From 3a3fca7f0ec151a6d325cf076c901f914befb0dc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Sun, 14 Jun 2026 22:33:34 +0100 Subject: [PATCH] feat: add resource metrics topbar (#1373) --- apps/backend/cmd/kandev/main.go | 5 + apps/backend/cmd/kandev/metrics_provider.go | 52 +++ .../cmd/kandev/metrics_provider_test.go | 123 +++++++ .../agent/runtime/agentctl/client_metrics.go | 33 ++ .../agentctl/server/api/metrics_handler.go | 40 +++ .../internal/agentctl/server/api/server.go | 4 + .../internal/gateway/websocket/client.go | 47 ++- .../backend/internal/gateway/websocket/hub.go | 115 ++++++- .../websocket/hub_session_mode_test.go | 1 + .../websocket/metrics_subscription_test.go | 122 +++++++ .../internal/system/metrics/collector.go | 312 ++++++++++++++++++ .../internal/system/metrics/disk_unix.go | 53 +++ .../internal/system/metrics/disk_unix_test.go | 49 +++ .../internal/system/metrics/disk_windows.go | 12 + .../internal/system/metrics/goleak_test.go | 11 + .../internal/system/metrics/handler.go | 44 +++ .../internal/system/metrics/service.go | 190 +++++++++++ .../internal/system/metrics/service_test.go | 163 +++++++++ .../internal/system/metrics/settings.go | 81 +++++ .../internal/system/metrics/settings_test.go | 98 ++++++ apps/backend/internal/system/metrics/store.go | 49 +++ apps/backend/internal/system/metrics/types.go | 28 ++ .../backend/internal/system/settings/store.go | 77 +++++ .../internal/system/settings/store_test.go | 84 +++++ apps/backend/internal/system/system.go | 18 + .../internal/user/controller/controller.go | 1 + apps/backend/internal/user/dto/dto.go | 107 +++--- apps/backend/internal/user/models/models.go | 5 + apps/backend/internal/user/service/service.go | 5 + apps/backend/internal/user/store/sqlite.go | 53 +-- .../internal/user/store/sqlite_test.go | 54 +++ apps/backend/pkg/websocket/actions.go | 15 +- apps/web/components/kanban/kanban-header.tsx | 3 + .../settings/editors-settings-state.tsx | 2 + .../components/settings/general-settings.tsx | 12 + .../settings/system-metrics-settings-card.tsx | 246 ++++++++++++++ apps/web/components/state-provider.test.tsx | 57 ++++ apps/web/components/state-provider.tsx | 11 +- .../system-metrics/topbar-metrics.tsx | 213 ++++++++++++ .../web/components/task/task-top-bar.test.tsx | 9 +- apps/web/components/task/task-top-bar.tsx | 2 + .../hooks/use-system-metrics-subscription.ts | 13 + apps/web/hooks/use-user-display-settings.ts | 1 + apps/web/lib/api/domains/settings-api.ts | 22 ++ apps/web/lib/ssr/user-settings.ts | 12 + .../state/slices/settings/settings-slice.ts | 1 + apps/web/lib/state/slices/settings/types.ts | 1 + .../lib/state/slices/system/system-slice.ts | 5 + apps/web/lib/state/slices/system/types.ts | 3 + apps/web/lib/types/backend.ts | 6 + apps/web/lib/types/http.ts | 5 +- apps/web/lib/types/system.ts | 43 +++ apps/web/lib/ws/client.ts | 34 ++ apps/web/lib/ws/handlers/system-events.ts | 3 + apps/web/lib/ws/handlers/users.ts | 7 +- .../0017-resource-metrics-sampling.md | 31 ++ docs/decisions/INDEX.md | 1 + 57 files changed, 2679 insertions(+), 115 deletions(-) create mode 100644 apps/backend/cmd/kandev/metrics_provider.go create mode 100644 apps/backend/cmd/kandev/metrics_provider_test.go create mode 100644 apps/backend/internal/agent/runtime/agentctl/client_metrics.go create mode 100644 apps/backend/internal/agentctl/server/api/metrics_handler.go create mode 100644 apps/backend/internal/gateway/websocket/metrics_subscription_test.go create mode 100644 apps/backend/internal/system/metrics/collector.go create mode 100644 apps/backend/internal/system/metrics/disk_unix.go create mode 100644 apps/backend/internal/system/metrics/disk_unix_test.go create mode 100644 apps/backend/internal/system/metrics/disk_windows.go create mode 100644 apps/backend/internal/system/metrics/goleak_test.go create mode 100644 apps/backend/internal/system/metrics/handler.go create mode 100644 apps/backend/internal/system/metrics/service.go create mode 100644 apps/backend/internal/system/metrics/service_test.go create mode 100644 apps/backend/internal/system/metrics/settings.go create mode 100644 apps/backend/internal/system/metrics/settings_test.go create mode 100644 apps/backend/internal/system/metrics/store.go create mode 100644 apps/backend/internal/system/metrics/types.go create mode 100644 apps/backend/internal/system/settings/store.go create mode 100644 apps/backend/internal/system/settings/store_test.go create mode 100644 apps/web/components/settings/system-metrics-settings-card.tsx create mode 100644 apps/web/components/state-provider.test.tsx create mode 100644 apps/web/components/system-metrics/topbar-metrics.tsx create mode 100644 apps/web/hooks/use-system-metrics-subscription.ts create mode 100644 docs/decisions/0017-resource-metrics-sampling.md diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index b7570ce6e0..5d3238318d 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -658,6 +658,11 @@ func startGatewayAndServe( }, systemsvc.Wiring{ OrchestratorShutdown: func() { _ = orchestratorSvc.Stop() }, }) + if systemSvc.Metrics != nil { + systemSvc.Metrics.SetBroadcaster(gateway.Hub.BroadcastToSystemMetrics) + gateway.Hub.SetSystemMetricsInterestTracker(systemSvc.Metrics) + systemSvc.Metrics.SetExecutionProvider(lifecycleMetricProvider{manager: lifecycleMgr}) + } systemSvc.StartBackground(ctx) gateways.RegisterSystemNotifications(ctx, eventBus, gateway.Hub, log) diff --git a/apps/backend/cmd/kandev/metrics_provider.go b/apps/backend/cmd/kandev/metrics_provider.go new file mode 100644 index 0000000000..e15c32da69 --- /dev/null +++ b/apps/backend/cmd/kandev/metrics_provider.go @@ -0,0 +1,52 @@ +package main + +import ( + "fmt" + + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" + "github.com/kandev/kandev/internal/agentruntime" + "github.com/kandev/kandev/internal/system/metrics" +) + +type lifecycleMetricProvider struct { + manager metricExecutionLister +} + +type metricExecutionLister interface { + ListExecutions() []*lifecycle.AgentExecution +} + +func (p lifecycleMetricProvider) MetricExecutions() []metrics.ExecutionSource { + if p.manager == nil { + return nil + } + executions := p.manager.ListExecutions() + sources := make([]metrics.ExecutionSource, 0, len(executions)) + for _, execution := range executions { + if execution == nil || execution.GetAgentCtlClient() == nil || !shouldCollectExecutionMetrics(execution.RuntimeName) { + continue + } + label := fmt.Sprintf("Execution %s", execution.SessionID) + if execution.TaskID != "" { + label = fmt.Sprintf("Task %s execution", execution.TaskID) + } + sources = append(sources, metrics.ExecutionSource{ + ID: execution.ID, + Label: label, + ExecutorType: execution.RuntimeName.String(), + SessionID: execution.SessionID, + TaskID: execution.TaskID, + Client: execution.GetAgentCtlClient(), + }) + } + return sources +} + +func shouldCollectExecutionMetrics(runtime agentruntime.Runtime) bool { + switch runtime { + case agentruntime.RuntimeDocker, agentruntime.RuntimeRemoteDocker, agentruntime.RuntimeSprites, agentruntime.RuntimeSSH: + return true + default: + return false + } +} diff --git a/apps/backend/cmd/kandev/metrics_provider_test.go b/apps/backend/cmd/kandev/metrics_provider_test.go new file mode 100644 index 0000000000..da61db1b5a --- /dev/null +++ b/apps/backend/cmd/kandev/metrics_provider_test.go @@ -0,0 +1,123 @@ +package main + +import ( + "reflect" + "testing" + "unsafe" + + agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" + "github.com/kandev/kandev/internal/agentruntime" + "github.com/kandev/kandev/internal/common/logger" + "go.uber.org/zap" +) + +type metricExecutionListStub struct { + executions []*lifecycle.AgentExecution +} + +func (s metricExecutionListStub) ListExecutions() []*lifecycle.AgentExecution { + return s.executions +} + +func TestLifecycleMetricProviderMetricExecutions(t *testing.T) { + docker := executionWithAgentCtl(t, &lifecycle.AgentExecution{ + ID: "exec-docker", + TaskID: "task-1", + SessionID: "session-1", + RuntimeName: agentruntime.RuntimeDocker, + }) + ssh := executionWithAgentCtl(t, &lifecycle.AgentExecution{ + ID: "exec-ssh", + SessionID: "session-2", + RuntimeName: agentruntime.RuntimeSSH, + }) + standalone := executionWithAgentCtl(t, &lifecycle.AgentExecution{ + ID: "exec-standalone", + TaskID: "task-2", + SessionID: "session-3", + RuntimeName: agentruntime.RuntimeStandalone, + }) + + provider := lifecycleMetricProvider{manager: metricExecutionListStub{executions: []*lifecycle.AgentExecution{ + nil, + docker, + ssh, + standalone, + { + ID: "exec-no-client", + TaskID: "task-3", + SessionID: "session-4", + RuntimeName: agentruntime.RuntimeDocker, + }, + }}} + + sources := provider.MetricExecutions() + if len(sources) != 2 { + t.Fatalf("expected 2 execution sources, got %d", len(sources)) + } + + if sources[0].ID != "exec-docker" { + t.Fatalf("expected docker source first, got %q", sources[0].ID) + } + if sources[0].Label != "Task task-1 execution" { + t.Fatalf("expected task label, got %q", sources[0].Label) + } + if sources[0].ExecutorType != string(agentruntime.RuntimeDocker) { + t.Fatalf("expected docker executor type, got %q", sources[0].ExecutorType) + } + if sources[0].Client == nil { + t.Fatal("expected docker source client") + } + + if sources[1].ID != "exec-ssh" { + t.Fatalf("expected ssh source second, got %q", sources[1].ID) + } + if sources[1].Label != "Execution session-2" { + t.Fatalf("expected session label, got %q", sources[1].Label) + } +} + +func TestLifecycleMetricProviderMetricExecutionsNilManager(t *testing.T) { + sources := (lifecycleMetricProvider{}).MetricExecutions() + if sources != nil { + t.Fatalf("expected nil sources, got %#v", sources) + } +} + +func TestShouldCollectExecutionMetrics(t *testing.T) { + tests := []struct { + runtime agentruntime.Runtime + want bool + }{ + {runtime: agentruntime.RuntimeDocker, want: true}, + {runtime: agentruntime.RuntimeRemoteDocker, want: true}, + {runtime: agentruntime.RuntimeSprites, want: true}, + {runtime: agentruntime.RuntimeSSH, want: true}, + {runtime: agentruntime.RuntimeStandalone, want: false}, + {runtime: agentruntime.Runtime("local_worktree"), want: false}, + {runtime: "", want: false}, + } + + for _, tt := range tests { + if got := shouldCollectExecutionMetrics(tt.runtime); got != tt.want { + t.Fatalf("shouldCollectExecutionMetrics(%q) = %v, want %v", tt.runtime, got, tt.want) + } + } +} + +func executionWithAgentCtl(t *testing.T, execution *lifecycle.AgentExecution) *lifecycle.AgentExecution { + t.Helper() + log, err := logger.NewFromZap(zap.NewNop()) + if err != nil { + t.Fatalf("new logger: %v", err) + } + setAgentCtlClient(t, execution, agentctl.NewClient("127.0.0.1", 1, log)) + return execution +} + +func setAgentCtlClient(t *testing.T, execution *lifecycle.AgentExecution, client *agentctl.Client) { + t.Helper() + field := reflect.ValueOf(execution).Elem().FieldByName("agentctl") + reflect.NewAt(field.Type(), unsafe.Pointer(field.UnsafeAddr())).Elem().Set(reflect.ValueOf(client)) +} diff --git a/apps/backend/internal/agent/runtime/agentctl/client_metrics.go b/apps/backend/internal/agent/runtime/agentctl/client_metrics.go new file mode 100644 index 0000000000..4cf60fc970 --- /dev/null +++ b/apps/backend/internal/agent/runtime/agentctl/client_metrics.go @@ -0,0 +1,33 @@ +package client + +import ( + "context" + "fmt" + "net/url" + "strings" + + "github.com/kandev/kandev/internal/system/metrics" +) + +func (c *Client) SystemMetrics(ctx context.Context, metricIDs []string, diskPath string) (*metrics.SourceSnapshot, error) { + values := url.Values{} + if len(metricIDs) > 0 { + values.Set("metrics", strings.Join(metricIDs, ",")) + } + if diskPath != "" { + values.Set("disk_path", diskPath) + } + path := "/api/v1/system/metrics" + if encoded := values.Encode(); encoded != "" { + path += "?" + encoded + } + var result metrics.SourceSnapshot + status, err := c.fetchJSONResult(ctx, path, &result) + if err != nil { + return nil, err + } + if status >= 400 { + return &result, fmt.Errorf("system metrics failed with status %d", status) + } + return &result, nil +} diff --git a/apps/backend/internal/agentctl/server/api/metrics_handler.go b/apps/backend/internal/agentctl/server/api/metrics_handler.go new file mode 100644 index 0000000000..672b8478d0 --- /dev/null +++ b/apps/backend/internal/agentctl/server/api/metrics_handler.go @@ -0,0 +1,40 @@ +package api + +import ( + "net/http" + "strings" + + "github.com/gin-gonic/gin" + "github.com/kandev/kandev/internal/system/metrics" +) + +func (s *Server) handleSystemMetrics(c *gin.Context) { + metricIDs := splitMetrics(c.Query("metrics")) + if len(metricIDs) == 0 { + metricIDs = metrics.DefaultSettings().Metrics + } + diskPath := c.Query("disk_path") + if diskPath == "" { + diskPath = "/" + } + snapshot := s.metricsCollector.Sample(c.Request.Context(), metricIDs, diskPath) + snapshot.ID = "agentctl" + snapshot.Label = "Execution" + snapshot.Kind = "execution" + c.JSON(http.StatusOK, snapshot) +} + +func splitMetrics(raw string) []string { + if raw == "" { + return nil + } + parts := strings.Split(raw, ",") + out := make([]string, 0, len(parts)) + for _, part := range parts { + part = strings.TrimSpace(part) + if part != "" { + out = append(out, part) + } + } + return out +} diff --git a/apps/backend/internal/agentctl/server/api/server.go b/apps/backend/internal/agentctl/server/api/server.go index 80e8b84b17..f8be960bf1 100644 --- a/apps/backend/internal/agentctl/server/api/server.go +++ b/apps/backend/internal/agentctl/server/api/server.go @@ -19,6 +19,7 @@ import ( "github.com/kandev/kandev/internal/common/httpmw" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/mcp/server" + "github.com/kandev/kandev/internal/system/metrics" "go.uber.org/zap" ) @@ -31,6 +32,7 @@ type Server struct { logger *logger.Logger router *gin.Engine portProxies *portProxyCache + metricsCollector *metrics.Collector upgrader websocket.Upgrader } @@ -49,6 +51,7 @@ func NewServer(cfg *config.InstanceConfig, procMgr *process.Manager, mcpServer * logger: log.WithFields(zap.String("component", "api-server")), router: gin.New(), portProxies: newPortProxyCache(), + metricsCollector: metrics.NewCollector(), upgrader: websocket.Upgrader{ CheckOrigin: func(r *http.Request) bool { return true // Allow all origins for container-local communication @@ -88,6 +91,7 @@ func (s *Server) setupRoutes() { // Status and info api.GET("/status", s.handleStatus) api.GET("/info", s.handleInfo) + api.GET("/system/metrics", s.handleSystemMetrics) // Process control api.POST("/agent/configure", s.handleAgentConfigure) diff --git a/apps/backend/internal/gateway/websocket/client.go b/apps/backend/internal/gateway/websocket/client.go index f9a039dfb6..3dfd4b98fc 100644 --- a/apps/backend/internal/gateway/websocket/client.go +++ b/apps/backend/internal/gateway/websocket/client.go @@ -30,18 +30,19 @@ const ( // Client represents a single WebSocket connection type Client struct { - ID string - conn *websocket.Conn - hub *Hub - send chan []byte - subscriptions map[string]bool // Task IDs this client is subscribed to - sessionSubscriptions map[string]bool // Session IDs this client is subscribed to - sessionFocus map[string]bool // Session IDs this client has focused (a strict subset of subscriptions, conceptually — see hub_session_mode.go) - userSubscriptions map[string]bool // User IDs this client is subscribed to - runSubscriptions map[string]bool // Office run IDs this client is subscribed to (for run.event.appended) - mu sync.RWMutex - closed bool - logger *logger.Logger + ID string + conn *websocket.Conn + hub *Hub + send chan []byte + subscriptions map[string]bool // Task IDs this client is subscribed to + sessionSubscriptions map[string]bool // Session IDs this client is subscribed to + sessionFocus map[string]bool // Session IDs this client has focused (a strict subset of subscriptions, conceptually — see hub_session_mode.go) + userSubscriptions map[string]bool // User IDs this client is subscribed to + runSubscriptions map[string]bool // Office run IDs this client is subscribed to (for run.event.appended) + systemMetricsSubscribed bool + mu sync.RWMutex + closed bool + logger *logger.Logger } // NewClient creates a new WebSocket client @@ -152,6 +153,12 @@ func (c *Client) handleMessage(msg *ws.Message) { case ws.ActionRunUnsubscribe: c.handleRunUnsubscribe(msg) return + case ws.ActionSystemMetricsSubscribe: + c.handleSystemMetricsSubscribe(msg) + return + case ws.ActionSystemMetricsUnsubscribe: + c.handleSystemMetricsUnsubscribe(msg) + return } // Dispatch to handler using the hub's lifetime context, not the per- @@ -377,6 +384,22 @@ func (c *Client) handleRunUnsubscribe(msg *ws.Message) { c.sendMessage(resp) } +func (c *Client) handleSystemMetricsSubscribe(msg *ws.Message) { + c.hub.SubscribeToSystemMetrics(c) + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + }) + c.sendMessage(resp) +} + +func (c *Client) handleSystemMetricsUnsubscribe(msg *ws.Message) { + c.hub.UnsubscribeFromSystemMetrics(c) + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + }) + c.sendMessage(resp) +} + // handleSessionUnsubscribe handles session.unsubscribe action func (c *Client) handleSessionUnsubscribe(msg *ws.Message) { var req SessionSubscribeRequest diff --git a/apps/backend/internal/gateway/websocket/hub.go b/apps/backend/internal/gateway/websocket/hub.go index aa1b183475..b215695feb 100644 --- a/apps/backend/internal/gateway/websocket/hub.go +++ b/apps/backend/internal/gateway/websocket/hub.go @@ -27,6 +27,8 @@ type Hub struct { userSubscribers map[string]map[*Client]bool // Clients subscribed to specific office run ids (for run.event.appended). runSubscribers map[string]map[*Client]bool + // Clients subscribed to backend/resource metrics. + systemMetricsSubscribers map[*Client]bool // Channels for client management register chan *Client @@ -43,7 +45,8 @@ type Hub struct { // sessionMode tracks per-session focus state and fires listeners when // effective mode (paused/slow/fast) transitions. See hub_session_mode.go. - sessionMode *sessionModeTracker + sessionMode *sessionModeTracker + metricsInterestTracker SystemMetricsInterestTracker // dispatchCtx is the hub's lifetime context, set by Run. Dispatched // message handlers use it instead of the per-connection context so that @@ -59,20 +62,26 @@ type Hub struct { // NewHub creates a new WebSocket hub func NewHub(dispatcher *ws.Dispatcher, log *logger.Logger) *Hub { return &Hub{ - clients: make(map[*Client]bool), - taskSubscribers: make(map[string]map[*Client]bool), - sessionSubscribers: make(map[string]map[*Client]bool), - userSubscribers: make(map[string]map[*Client]bool), - runSubscribers: make(map[string]map[*Client]bool), - register: make(chan *Client), - unregister: make(chan *Client), - broadcast: make(chan *ws.Message, 256), - dispatcher: dispatcher, - sessionMode: newSessionModeTracker(), - logger: log.WithFields(zap.String("component", "ws_hub")), + clients: make(map[*Client]bool), + taskSubscribers: make(map[string]map[*Client]bool), + sessionSubscribers: make(map[string]map[*Client]bool), + userSubscribers: make(map[string]map[*Client]bool), + runSubscribers: make(map[string]map[*Client]bool), + systemMetricsSubscribers: make(map[*Client]bool), + register: make(chan *Client), + unregister: make(chan *Client), + broadcast: make(chan *ws.Message, 256), + dispatcher: dispatcher, + sessionMode: newSessionModeTracker(), + logger: log.WithFields(zap.String("component", "ws_hub")), } } +type SystemMetricsInterestTracker interface { + MetricsSubscribe(clientID string) + MetricsUnsubscribe(clientID string) +} + // Run starts the hub's main processing loop func (h *Hub) Run(ctx context.Context) { h.logger.Info("WebSocket hub started") @@ -108,16 +117,29 @@ func (h *Hub) Run(ctx context.Context) { // after shutdown and call into listeners with stale state. func (h *Hub) closeAllClients() { h.mu.Lock() + metricClientIDs := make([]string, 0, len(h.systemMetricsSubscribers)) for client := range h.clients { + if client.systemMetricsSubscribed { + metricClientIDs = append(metricClientIDs, client.ID) + client.systemMetricsSubscribed = false + } client.closeSend() delete(h.clients, client) } + tracker := h.metricsInterestTracker h.taskSubscribers = make(map[string]map[*Client]bool) h.sessionSubscribers = make(map[string]map[*Client]bool) h.runSubscribers = make(map[string]map[*Client]bool) + h.systemMetricsSubscribers = make(map[*Client]bool) h.sessionMode.focusByClient = make(map[string]map[*Client]bool) h.mu.Unlock() + for _, clientID := range metricClientIDs { + if tracker != nil { + tracker.MetricsUnsubscribe(clientID) + } + } + h.stopAllPendingTransitions() } @@ -156,8 +178,20 @@ func (h *Hub) removeClient(client *Client) { for runID := range client.runSubscriptions { removeClientFromSubscriberMap(h.runSubscribers, runID, client) } + var metricClientID string + var tracker SystemMetricsInterestTracker + if client.systemMetricsSubscribed { + delete(h.systemMetricsSubscribers, client) + client.systemMetricsSubscribed = false + metricClientID = client.ID + tracker = h.metricsInterestTracker + } h.mu.Unlock() + if tracker != nil && metricClientID != "" { + tracker.MetricsUnsubscribe(metricClientID) + } + for _, sessionID := range dedupStrings(affectedSessions) { h.recomputeSessionMode(sessionID) } @@ -165,6 +199,12 @@ func (h *Hub) removeClient(client *Client) { h.logger.Debug("Client unregistered", zap.String("client_id", client.ID)) } +func (h *Hub) SetSystemMetricsInterestTracker(tracker SystemMetricsInterestTracker) { + h.mu.Lock() + defer h.mu.Unlock() + h.metricsInterestTracker = tracker +} + // dedupStrings returns the input with duplicates removed, preserving order. // Used to call recomputeSessionMode at most once per affected session when a // client is both subscribed and focused. @@ -425,6 +465,21 @@ func (h *Hub) BroadcastToRun(runID string, msg *ws.Message) { h.sendToClients(data, clients, msg.Action) } +func (h *Hub) BroadcastToSystemMetrics(msg *ws.Message) { + data, err := json.Marshal(msg) + if err != nil { + h.logger.Error("Failed to marshal message", zap.Error(err)) + return + } + h.mu.RLock() + clients := make([]*Client, 0, len(h.systemMetricsSubscribers)) + for client := range h.systemMetricsSubscribers { + clients = append(clients, client) + } + h.mu.RUnlock() + h.sendToClients(data, clients, msg.Action) +} + // SubscribeToRun subscribes a client to office run-event notifications. func (h *Hub) SubscribeToRun(client *Client, runID string) { h.mu.Lock() @@ -455,6 +510,42 @@ func (h *Hub) UnsubscribeFromRun(client *Client, runID string) { } } +func (h *Hub) SubscribeToSystemMetrics(client *Client) { + h.mu.Lock() + if _, ok := h.clients[client]; !ok { + h.mu.Unlock() + return + } + if client.systemMetricsSubscribed { + h.mu.Unlock() + return + } + h.systemMetricsSubscribers[client] = true + client.systemMetricsSubscribed = true + tracker := h.metricsInterestTracker + h.mu.Unlock() + + if tracker != nil { + tracker.MetricsSubscribe(client.ID) + } +} + +func (h *Hub) UnsubscribeFromSystemMetrics(client *Client) { + h.mu.Lock() + if !client.systemMetricsSubscribed { + h.mu.Unlock() + return + } + delete(h.systemMetricsSubscribers, client) + client.systemMetricsSubscribed = false + tracker := h.metricsInterestTracker + h.mu.Unlock() + + if tracker != nil { + tracker.MetricsUnsubscribe(client.ID) + } +} + // UnsubscribeFromTask unsubscribes a client from task notifications func (h *Hub) UnsubscribeFromTask(client *Client, taskID string) { h.mu.Lock() diff --git a/apps/backend/internal/gateway/websocket/hub_session_mode_test.go b/apps/backend/internal/gateway/websocket/hub_session_mode_test.go index 401fa4c0fc..f27e783bc9 100644 --- a/apps/backend/internal/gateway/websocket/hub_session_mode_test.go +++ b/apps/backend/internal/gateway/websocket/hub_session_mode_test.go @@ -35,6 +35,7 @@ func newTestClient(id string) *Client { sessionSubscriptions: map[string]bool{}, sessionFocus: map[string]bool{}, userSubscriptions: map[string]bool{}, + runSubscriptions: map[string]bool{}, logger: log, } } diff --git a/apps/backend/internal/gateway/websocket/metrics_subscription_test.go b/apps/backend/internal/gateway/websocket/metrics_subscription_test.go new file mode 100644 index 0000000000..49be7a501d --- /dev/null +++ b/apps/backend/internal/gateway/websocket/metrics_subscription_test.go @@ -0,0 +1,122 @@ +package websocket + +import ( + "encoding/json" + "testing" + + ws "github.com/kandev/kandev/pkg/websocket" +) + +type metricsInterestRecorder struct { + subs []string + unsubs []string +} + +func (r *metricsInterestRecorder) MetricsSubscribe(clientID string) { + r.subs = append(r.subs, clientID) +} + +func (r *metricsInterestRecorder) MetricsUnsubscribe(clientID string) { + r.unsubs = append(r.unsubs, clientID) +} + +func TestSystemMetricsSubscriptionBroadcastsOnlyToSubscribers(t *testing.T) { + h := newTestHub(t) + c1 := newTestClient("c1") + c2 := newTestClient("c2") + registerTestClient(h, c1) + registerTestClient(h, c2) + + h.SubscribeToSystemMetrics(c1) + + msg, err := ws.NewNotification(ws.ActionSystemMetricsUpdated, map[string]any{"ok": true}) + if err != nil { + t.Fatalf("notification: %v", err) + } + h.BroadcastToSystemMetrics(msg) + + if !clientReceived(c1) { + t.Fatal("subscribed client did not receive metrics update") + } + if clientReceived(c2) { + t.Fatal("unsubscribed client received metrics update") + } +} + +func TestSystemMetricsInterestTracksClientLifecycle(t *testing.T) { + h := newTestHub(t) + rec := &metricsInterestRecorder{} + h.SetSystemMetricsInterestTracker(rec) + c := newTestClient("c1") + registerTestClient(h, c) + + h.SubscribeToSystemMetrics(c) + h.SubscribeToSystemMetrics(c) + if len(rec.subs) != 1 { + t.Fatalf("subscribe calls=%d, want 1", len(rec.subs)) + } + + h.removeClient(c) + if len(rec.unsubs) != 1 { + t.Fatalf("unsubscribe calls=%d, want 1", len(rec.unsubs)) + } +} + +func TestSystemMetricsInterestTracksHubShutdown(t *testing.T) { + h := newTestHub(t) + rec := &metricsInterestRecorder{} + h.SetSystemMetricsInterestTracker(rec) + c := newTestClient("c1") + registerTestClient(h, c) + + h.SubscribeToSystemMetrics(c) + h.closeAllClients() + + if len(rec.unsubs) != 1 { + t.Fatalf("unsubscribe calls=%d, want 1", len(rec.unsubs)) + } +} + +func TestHandleSystemMetricsSubscribe(t *testing.T) { + h := newTestHub(t) + rec := &metricsInterestRecorder{} + h.SetSystemMetricsInterestTracker(rec) + c := newTestClient("c1") + c.hub = h + registerTestClient(h, c) + + msg, _ := ws.NewRequest("req-1", ws.ActionSystemMetricsSubscribe, map[string]any{}) + c.handleMessage(msg) + + if len(rec.subs) != 1 { + t.Fatalf("subscribe calls=%d, want 1", len(rec.subs)) + } + select { + case data := <-c.send: + var got ws.Message + if err := json.Unmarshal(data, &got); err != nil { + t.Fatalf("decode response: %v", err) + } + if got.Type != ws.MessageTypeResponse || got.Action != ws.ActionSystemMetricsSubscribe { + t.Fatalf("unexpected response type/action: %s %s", got.Type, got.Action) + } + default: + t.Fatal("expected subscribe response") + } +} + +func TestSystemMetricsSubscribeIgnoresDisconnectedClient(t *testing.T) { + h := newTestHub(t) + rec := &metricsInterestRecorder{} + h.SetSystemMetricsInterestTracker(rec) + c := newTestClient("c1") + + h.SubscribeToSystemMetrics(c) + + if len(rec.subs) != 0 { + t.Fatalf("subscribe calls=%d, want 0", len(rec.subs)) + } + if c.systemMetricsSubscribed { + t.Fatal("disconnected client should not be marked as subscribed") + } +} diff --git a/apps/backend/internal/system/metrics/collector.go b/apps/backend/internal/system/metrics/collector.go new file mode 100644 index 0000000000..d9977462ba --- /dev/null +++ b/apps/backend/internal/system/metrics/collector.go @@ -0,0 +1,312 @@ +package metrics + +import ( + "bufio" + "context" + "errors" + "fmt" + "math" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "time" +) + +type Collector struct { + procRoot string + sysRoot string + cgroupRoot string + prevCPU *cpuTimes + lastCPUAt time.Time + mu sync.Mutex +} + +func NewCollector() *Collector { + return &Collector{ + procRoot: "/proc", + sysRoot: "/sys", + cgroupRoot: "/sys/fs/cgroup", + } +} + +func (c *Collector) Sample(ctx context.Context, metricIDs []string, diskPath string) SourceSnapshot { + if err := ctx.Err(); err != nil { + return canceledBackendSnapshot(metricIDs, err) + } + for !c.mu.TryLock() { + select { + case <-ctx.Done(): + return canceledBackendSnapshot(metricIDs, ctx.Err()) + case <-time.After(10 * time.Millisecond): + } + } + defer c.mu.Unlock() + if diskPath == "" { + diskPath = "/" + } + samples := make([]MetricSample, 0, len(metricIDs)) + for _, id := range metricIDs { + if err := ctx.Err(); err != nil { + samples = append(samples, sample(id, metricLabel(id), metricUnit(id), 0, err)) + continue + } + samples = append(samples, c.sampleMetric(ctx, id, diskPath)) + } + return SourceSnapshot{ + ID: "kandev-backend", + Label: "Kandev backend", + Kind: "backend", + Metrics: samples, + } +} + +func (c *Collector) Reset() { + c.mu.Lock() + defer c.mu.Unlock() + c.prevCPU = nil + c.lastCPUAt = time.Time{} +} + +func (c *Collector) sampleMetric(ctx context.Context, id string, diskPath string) MetricSample { + switch id { + case MetricCPUPercent: + value, err := c.cpuPercent() + return sample(id, "CPU", "%", value, err) + case MetricMemoryPercent: + value, err := c.memoryPercent() + return sample(id, "Memory", "%", value, err) + case MetricDiskPercent: + value, err := diskPercent(ctx, diskPath) + return sample(id, "Disk", "%", value, err) + case MetricCPUTemp: + value, err := cpuTemp(c.sysRoot) + return sample(id, "CPU temp", "C", value, err) + case MetricIOLoad: + value, err := ioLoad(c.procRoot) + return sample(id, "Load avg", "", value, err) + default: + return MetricSample{ID: id, Label: id, Available: false, Error: "unknown metric"} + } +} + +func canceledBackendSnapshot(metricIDs []string, err error) SourceSnapshot { + samples := make([]MetricSample, 0, len(metricIDs)) + for _, id := range metricIDs { + samples = append(samples, sample(id, metricLabel(id), metricUnit(id), 0, err)) + } + return SourceSnapshot{ + ID: "kandev-backend", + Label: "Kandev backend", + Kind: "backend", + Metrics: samples, + } +} + +func metricLabel(id string) string { + switch id { + case MetricCPUPercent: + return "CPU" + case MetricMemoryPercent: + return "Memory" + case MetricDiskPercent: + return "Disk" + case MetricCPUTemp: + return "CPU temp" + case MetricIOLoad: + return "Load avg" + default: + return id + } +} + +func metricUnit(id string) string { + switch id { + case MetricCPUPercent, MetricMemoryPercent, MetricDiskPercent: + return "%" + case MetricCPUTemp: + return "C" + default: + return "" + } +} + +func sample(id, label, unit string, value float64, err error) MetricSample { + if err != nil { + return MetricSample{ID: id, Label: label, Unit: unit, Available: false, Error: err.Error()} + } + rounded := math.Round(value*10) / 10 + return MetricSample{ID: id, Label: label, Unit: unit, Value: &rounded, Available: true} +} + +type cpuTimes struct { + total uint64 + idle uint64 +} + +func (c *Collector) cpuPercent() (float64, error) { + current, err := readProcStat(filepath.Join(c.procRoot, "stat")) + if err != nil { + return 0, err + } + now := time.Now() + if c.prevCPU != nil && !c.lastCPUAt.IsZero() && now.Sub(c.lastCPUAt) > 2*time.Duration(MaxIntervalSeconds)*time.Second { + c.prevCPU = nil + } + if c.prevCPU == nil { + c.prevCPU = ¤t + c.lastCPUAt = now + return 0, nil + } + value := calculateCPUPercent(*c.prevCPU, current) + c.prevCPU = ¤t + c.lastCPUAt = now + return value, nil +} + +func readProcStat(path string) (cpuTimes, error) { + data, err := os.ReadFile(path) + if err != nil { + return cpuTimes{}, err + } + fields := strings.Fields(strings.SplitN(string(data), "\n", 2)[0]) + if len(fields) < 8 || fields[0] != "cpu" { + return cpuTimes{}, errors.New("invalid /proc/stat cpu line") + } + var total uint64 + var idle uint64 + for i, field := range fields[1:] { + v, err := strconv.ParseUint(field, 10, 64) + if err != nil { + return cpuTimes{}, err + } + total += v + if i == 3 || i == 4 { + idle += v + } + } + return cpuTimes{total: total, idle: idle}, nil +} + +func calculateCPUPercent(prev, current cpuTimes) float64 { + totalDelta := current.total - prev.total + if totalDelta == 0 { + return 0 + } + idleDelta := current.idle - prev.idle + return (1 - float64(idleDelta)/float64(totalDelta)) * 100 +} + +func (c *Collector) memoryPercent() (float64, error) { + if value, ok := c.cgroupMemoryPercent(); ok { + return value, nil + } + return memInfoPercent(filepath.Join(c.procRoot, "meminfo")) +} + +func (c *Collector) cgroupMemoryPercent() (float64, bool) { + usage, err := readUintFile(filepath.Join(c.cgroupRoot, "memory.current")) + if err != nil { + return 0, false + } + maxRaw, err := os.ReadFile(filepath.Join(c.cgroupRoot, "memory.max")) + if err != nil { + return 0, false + } + maxText := strings.TrimSpace(string(maxRaw)) + if maxText == "max" { + return 0, false + } + limit, err := strconv.ParseUint(maxText, 10, 64) + if err != nil || limit == 0 { + return 0, false + } + return float64(usage) / float64(limit) * 100, true +} + +func memInfoPercent(path string) (float64, error) { + data, err := os.ReadFile(path) + if err != nil { + return 0, err + } + values := map[string]uint64{} + scanner := bufio.NewScanner(strings.NewReader(string(data))) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) < 2 { + continue + } + key := strings.TrimSuffix(fields[0], ":") + v, err := strconv.ParseUint(fields[1], 10, 64) + if err == nil { + values[key] = v + } + } + total := values["MemTotal"] + available := values["MemAvailable"] + if total == 0 { + return 0, errors.New("MemTotal missing") + } + if available == 0 { + available = values["MemFree"] + } + return (1 - float64(available)/float64(total)) * 100, nil +} + +func cpuTemp(sysRoot string) (float64, error) { + paths, err := filepath.Glob(filepath.Join(sysRoot, "class/thermal/thermal_zone*/temp")) + if err != nil || len(paths) == 0 { + return 0, errors.New("cpu temperature unavailable") + } + for _, path := range paths { + raw, err := readUintFile(path) + if err != nil { + continue + } + value := float64(raw) + if value > 1000 { + value /= 1000 + } + return value, nil + } + return 0, errors.New("cpu temperature unavailable") +} + +func ioLoad(procRoot string) (float64, error) { + data, err := os.ReadFile(filepath.Join(procRoot, "loadavg")) + if err != nil { + return 0, err + } + fields := strings.Fields(string(data)) + if len(fields) == 0 { + return 0, errors.New("loadavg missing") + } + return strconv.ParseFloat(fields[0], 64) +} + +func readUintFile(path string) (uint64, error) { + data, err := os.ReadFile(path) + if err != nil { + return 0, err + } + return strconv.ParseUint(strings.TrimSpace(string(data)), 10, 64) +} + +func (c *Collector) SampleWithTimestamp(ctx context.Context, settings GlobalSettings) Snapshot { + source := c.Sample(ctx, settings.Metrics, settings.BackendDiskPath) + return Snapshot{ + Timestamp: time.Now().UTC(), + IntervalSeconds: settings.IntervalSeconds, + Sources: []SourceSnapshot{source}, + } +} + +func unavailableExecutionSource(id, label, kind, errText string) SourceSnapshot { + return SourceSnapshot{ + ID: id, + Label: label, + Kind: kind, + Metrics: []MetricSample{{ID: "error", Label: "Metrics", Available: false, Error: fmt.Sprintf("execution metrics unavailable: %s", errText)}}, + } +} diff --git a/apps/backend/internal/system/metrics/disk_unix.go b/apps/backend/internal/system/metrics/disk_unix.go new file mode 100644 index 0000000000..0f81aa5441 --- /dev/null +++ b/apps/backend/internal/system/metrics/disk_unix.go @@ -0,0 +1,53 @@ +//go:build !windows + +package metrics + +import ( + "context" + "errors" + "fmt" + "syscall" + "time" +) + +const diskStatfsTimeout = 2 * time.Second + +var statfs = syscall.Statfs + +type diskStatfsResult struct { + stat syscall.Statfs_t + err error +} + +func diskPercent(ctx context.Context, path string) (float64, error) { + result := make(chan diskStatfsResult, 1) + go func() { + var stat syscall.Statfs_t + err := statfs(path, &stat) + result <- diskStatfsResult{stat: stat, err: err} + }() + + timer := time.NewTimer(diskStatfsTimeout) + defer timer.Stop() + + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-timer.C: + return 0, fmt.Errorf("disk usage timed out for %q", path) + case res := <-result: + if res.err != nil { + return 0, res.err + } + return diskPercentFromStatfs(res.stat) + } +} + +func diskPercentFromStatfs(stat syscall.Statfs_t) (float64, error) { + total := stat.Blocks * uint64(stat.Bsize) + free := stat.Bavail * uint64(stat.Bsize) + if total == 0 { + return 0, errors.New("disk total is zero") + } + return (1 - float64(free)/float64(total)) * 100, nil +} diff --git a/apps/backend/internal/system/metrics/disk_unix_test.go b/apps/backend/internal/system/metrics/disk_unix_test.go new file mode 100644 index 0000000000..d4424e7cbe --- /dev/null +++ b/apps/backend/internal/system/metrics/disk_unix_test.go @@ -0,0 +1,49 @@ +//go:build !windows + +package metrics + +import ( + "context" + "errors" + "syscall" + "testing" + "time" +) + +func TestDiskPercentReturnsWhenContextCancelsWhileStatfsBlocks(t *testing.T) { + original := statfs + block := make(chan struct{}) + started := make(chan struct{}) + statfs = func(_ string, _ *syscall.Statfs_t) error { + close(started) + <-block + return errors.New("unblocked") + } + t.Cleanup(func() { + close(block) + statfs = original + }) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + _, err := diskPercent(ctx, "/slow") + done <- err + }() + + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("statfs was not called") + } + cancel() + + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("diskPercent error = %v, want context.Canceled", err) + } + case <-time.After(500 * time.Millisecond): + t.Fatal("diskPercent did not return after context cancellation") + } +} diff --git a/apps/backend/internal/system/metrics/disk_windows.go b/apps/backend/internal/system/metrics/disk_windows.go new file mode 100644 index 0000000000..f538414d99 --- /dev/null +++ b/apps/backend/internal/system/metrics/disk_windows.go @@ -0,0 +1,12 @@ +//go:build windows + +package metrics + +import ( + "context" + "errors" +) + +func diskPercent(_ context.Context, _ string) (float64, error) { + return 0, errors.New("disk usage unavailable on windows") +} diff --git a/apps/backend/internal/system/metrics/goleak_test.go b/apps/backend/internal/system/metrics/goleak_test.go new file mode 100644 index 0000000000..33a11e122f --- /dev/null +++ b/apps/backend/internal/system/metrics/goleak_test.go @@ -0,0 +1,11 @@ +package metrics + +import ( + "testing" + + "go.uber.org/goleak" +) + +func TestMain(m *testing.M) { + goleak.VerifyTestMain(m) +} diff --git a/apps/backend/internal/system/metrics/handler.go b/apps/backend/internal/system/metrics/handler.go new file mode 100644 index 0000000000..70079d773a --- /dev/null +++ b/apps/backend/internal/system/metrics/handler.go @@ -0,0 +1,44 @@ +package metrics + +import ( + "errors" + "net/http" + + "github.com/gin-gonic/gin" +) + +func RegisterRoutes(g *gin.RouterGroup, svc *Service) { + g.GET("/metrics/settings", handleGetSettings(svc)) + g.PATCH("/metrics/settings", handleSaveSettings(svc)) +} + +func handleGetSettings(svc *Service) gin.HandlerFunc { + return func(c *gin.Context) { + settings, err := svc.GetSettings(c.Request.Context()) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load metrics settings"}) + return + } + c.JSON(http.StatusOK, gin.H{"settings": settings}) + } +} + +func handleSaveSettings(svc *Service) gin.HandlerFunc { + return func(c *gin.Context) { + var req GlobalSettings + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request"}) + return + } + settings, err := svc.SaveSettings(c.Request.Context(), req) + if err != nil { + if errors.Is(err, ErrValidation) { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to save metrics settings"}) + return + } + c.JSON(http.StatusOK, gin.H{"settings": settings}) + } +} diff --git a/apps/backend/internal/system/metrics/service.go b/apps/backend/internal/system/metrics/service.go new file mode 100644 index 0000000000..f8e373d0e2 --- /dev/null +++ b/apps/backend/internal/system/metrics/service.go @@ -0,0 +1,190 @@ +package metrics + +import ( + "context" + "sync" + "time" + + ws "github.com/kandev/kandev/pkg/websocket" +) + +type Broadcaster func(*ws.Message) + +type AgentctlMetricsClient interface { + SystemMetrics(ctx context.Context, metricIDs []string, diskPath string) (*SourceSnapshot, error) +} + +type ExecutionSource struct { + ID string + Label string + ExecutorType string + SessionID string + TaskID string + Client AgentctlMetricsClient +} + +type ExecutionProvider interface { + MetricExecutions() []ExecutionSource +} + +const executionMetricsDiskPath = "/" + +type Service struct { + store *Store + collector *Collector + + mu sync.Mutex + interested map[string]struct{} + cancel context.CancelFunc + runDone chan struct{} + broadcaster Broadcaster + executions ExecutionProvider +} + +func NewService(store *Store, collector *Collector) *Service { + return &Service{ + store: store, + collector: collector, + interested: make(map[string]struct{}), + } +} + +func (s *Service) SetBroadcaster(b Broadcaster) { + s.mu.Lock() + defer s.mu.Unlock() + s.broadcaster = b +} + +func (s *Service) SetExecutionProvider(provider ExecutionProvider) { + s.mu.Lock() + defer s.mu.Unlock() + s.executions = provider +} + +func (s *Service) GetSettings(ctx context.Context) (GlobalSettings, error) { + return s.store.GetSettings(ctx) +} + +func (s *Service) SaveSettings(ctx context.Context, settings GlobalSettings) (GlobalSettings, error) { + return s.store.SaveSettings(ctx, settings) +} + +func (s *Service) MetricsSubscribe(clientID string) { + s.mu.Lock() + defer s.mu.Unlock() + if _, ok := s.interested[clientID]; ok { + return + } + s.interested[clientID] = struct{}{} + if len(s.interested) == 1 { + s.startLocked() + } +} + +func (s *Service) MetricsUnsubscribe(clientID string) { + s.mu.Lock() + delete(s.interested, clientID) + var done chan struct{} + if len(s.interested) == 0 && s.cancel != nil { + s.cancel() + done = s.runDone + } + s.mu.Unlock() + if done != nil { + go s.finishStop(done) + } +} + +func (s *Service) finishStop(done chan struct{}) { + <-done + s.mu.Lock() + defer s.mu.Unlock() + if s.runDone == done { + s.cancel = nil + s.runDone = nil + if len(s.interested) > 0 { + s.startLocked() + } + } +} + +func (s *Service) startLocked() { + if s.cancel != nil { + return + } + s.collector.Reset() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + s.cancel = cancel + s.runDone = done + go s.run(ctx, done) +} + +func (s *Service) run(ctx context.Context, done chan struct{}) { + defer close(done) + for { + settings, err := s.store.GetSettings(ctx) + if err == nil { + s.publish(ctx, settings) + } + interval := time.Duration(DefaultIntervalSeconds) * time.Second + if err == nil { + interval = time.Duration(settings.IntervalSeconds) * time.Second + } + timer := time.NewTimer(interval) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + } +} + +func (s *Service) publish(ctx context.Context, settings GlobalSettings) { + s.mu.Lock() + broadcaster := s.broadcaster + hasInterest := len(s.interested) > 0 + s.mu.Unlock() + if broadcaster == nil || !hasInterest { + return + } + snapshot := s.collector.SampleWithTimestamp(ctx, settings) + if settings.CollectExecution { + snapshot.Sources = append(snapshot.Sources, s.collectExecutions(ctx, settings)...) + } + msg, err := ws.NewNotification(ws.ActionSystemMetricsUpdated, snapshot) + if err != nil { + return + } + broadcaster(msg) +} + +func (s *Service) collectExecutions(ctx context.Context, settings GlobalSettings) []SourceSnapshot { + s.mu.Lock() + provider := s.executions + s.mu.Unlock() + if provider == nil { + return nil + } + executions := provider.MetricExecutions() + sources := make([]SourceSnapshot, 0, len(executions)) + for _, execution := range executions { + if execution.Client == nil { + continue + } + source, err := execution.Client.SystemMetrics(ctx, settings.Metrics, executionMetricsDiskPath) + if err != nil { + sources = append(sources, unavailableExecutionSource(execution.ID, execution.Label, "execution", err.Error())) + continue + } + source.ID = execution.ID + source.Label = execution.Label + source.Kind = "execution" + source.ExecutorType = execution.ExecutorType + source.SessionID = execution.SessionID + source.TaskID = execution.TaskID + sources = append(sources, *source) + } + return sources +} diff --git a/apps/backend/internal/system/metrics/service_test.go b/apps/backend/internal/system/metrics/service_test.go new file mode 100644 index 0000000000..862c7b1825 --- /dev/null +++ b/apps/backend/internal/system/metrics/service_test.go @@ -0,0 +1,163 @@ +package metrics + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/jmoiron/sqlx" + _ "github.com/mattn/go-sqlite3" + + "github.com/kandev/kandev/internal/db" + systemsettings "github.com/kandev/kandev/internal/system/settings" + ws "github.com/kandev/kandev/pkg/websocket" +) + +func TestServiceSubscribeLifecycle(t *testing.T) { + svc := newTestService(t) + published := make(chan struct{}, 4) + svc.SetBroadcaster(func(_ *ws.Message) { + published <- struct{}{} + }) + + svc.MetricsSubscribe("client-1") + t.Cleanup(func() { unsubscribeAll(t, svc) }) + waitForPublish(t, published) + + svc.mu.Lock() + if len(svc.interested) != 1 { + t.Fatalf("interested=%d, want 1", len(svc.interested)) + } + if svc.cancel == nil { + t.Fatal("expected sampler to start on first subscriber") + } + svc.mu.Unlock() + + svc.MetricsSubscribe("client-1") + svc.mu.Lock() + if len(svc.interested) != 1 { + t.Fatalf("interested after duplicate subscribe=%d, want 1", len(svc.interested)) + } + svc.mu.Unlock() + + svc.MetricsSubscribe("client-2") + svc.MetricsUnsubscribe("client-1") + svc.mu.Lock() + if svc.cancel == nil { + t.Fatal("expected sampler to keep running while another subscriber remains") + } + svc.mu.Unlock() + + svc.MetricsUnsubscribe("client-2") + waitForStoppedSampler(t, svc) +} + +func TestServiceCollectExecutionsUsesContainerRootDiskPath(t *testing.T) { + svc := newTestService(t) + client := &recordingMetricsClient{} + svc.SetExecutionProvider(staticExecutionProvider{sources: []ExecutionSource{{ + ID: "exec-1", + Label: "Executor", + Client: client, + }}}) + + settings := DefaultSettings() + settings.BackendDiskPath = "/mnt/backend-data" + settings.CollectExecution = true + + sources := svc.collectExecutions(context.Background(), settings) + if len(sources) != 1 { + t.Fatalf("sources=%d, want 1", len(sources)) + } + if client.diskPath != "/" { + t.Fatalf("disk path=%q, want /", client.diskPath) + } +} + +func newTestService(t *testing.T) *Service { + t.Helper() + conn, err := sqlx.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + conn.SetMaxOpenConns(1) + t.Cleanup(func() { _ = conn.Close() }) + settingsStore, err := systemsettings.NewStore(db.NewPool(conn, conn)) + if err != nil { + t.Fatalf("new settings store: %v", err) + } + store := NewStore(settingsStore) + if _, err := store.SaveSettings(context.Background(), GlobalSettings{ + Metrics: []string{MetricCPUPercent}, + IntervalSeconds: 1, + BackendDiskPath: "/", + }); err != nil { + t.Fatalf("save settings: %v", err) + } + return NewService(store, NewCollector()) +} + +func waitForPublish(t *testing.T, ch <-chan struct{}) { + t.Helper() + select { + case <-ch: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for metrics publish") + } +} + +func waitForStoppedSampler(t *testing.T, svc *Service) { + t.Helper() + deadline := time.After(2 * time.Second) + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-deadline: + t.Fatal("timed out waiting for sampler to stop") + case <-ticker.C: + svc.mu.Lock() + interested := len(svc.interested) + cancel := svc.cancel + svc.mu.Unlock() + if interested == 0 && cancel == nil { + return + } + } + } +} + +func unsubscribeAll(t *testing.T, svc *Service) { + t.Helper() + svc.mu.Lock() + clients := make([]string, 0, len(svc.interested)) + for client := range svc.interested { + clients = append(clients, client) + } + svc.mu.Unlock() + for _, client := range clients { + svc.MetricsUnsubscribe(client) + } + waitForStoppedSampler(t, svc) +} + +type staticExecutionProvider struct { + sources []ExecutionSource +} + +func (p staticExecutionProvider) MetricExecutions() []ExecutionSource { + return p.sources +} + +type recordingMetricsClient struct { + mu sync.Mutex + diskPath string +} + +func (c *recordingMetricsClient) SystemMetrics(_ context.Context, _ []string, diskPath string) (*SourceSnapshot, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.diskPath = diskPath + return &SourceSnapshot{Metrics: []MetricSample{}}, nil +} diff --git a/apps/backend/internal/system/metrics/settings.go b/apps/backend/internal/system/metrics/settings.go new file mode 100644 index 0000000000..cc6b6a0384 --- /dev/null +++ b/apps/backend/internal/system/metrics/settings.go @@ -0,0 +1,81 @@ +package metrics + +import ( + "errors" + "fmt" +) + +var ErrValidation = errors.New("metrics settings validation") + +const ( + MetricCPUPercent = "cpu_percent" + MetricMemoryPercent = "memory_percent" + MetricDiskPercent = "disk_percent" + MetricCPUTemp = "cpu_temp" + MetricIOLoad = "io_load" + + DefaultIntervalSeconds = 5 + MinIntervalSeconds = 1 + MaxIntervalSeconds = 5 * 60 +) + +type GlobalSettings struct { + Metrics []string `json:"metrics"` + IntervalSeconds int `json:"interval_seconds"` + BackendDiskPath string `json:"backend_disk_path"` + CollectExecution bool `json:"collect_execution"` +} + +func DefaultSettings() GlobalSettings { + return GlobalSettings{ + Metrics: []string{MetricCPUPercent, MetricMemoryPercent, MetricDiskPercent}, + IntervalSeconds: DefaultIntervalSeconds, + BackendDiskPath: "/", + CollectExecution: false, + } +} + +func NormalizeSettings(in GlobalSettings) (GlobalSettings, error) { + defaults := DefaultSettings() + out := in + if out.IntervalSeconds == 0 { + out.IntervalSeconds = defaults.IntervalSeconds + } + if out.IntervalSeconds < MinIntervalSeconds || out.IntervalSeconds > MaxIntervalSeconds { + return GlobalSettings{}, fmt.Errorf("%w: interval_seconds must be between %d and %d", ErrValidation, MinIntervalSeconds, MaxIntervalSeconds) + } + if out.BackendDiskPath == "" { + out.BackendDiskPath = defaults.BackendDiskPath + } + if len(out.Metrics) == 0 { + out.Metrics = defaults.Metrics + return out, nil + } + + seen := make(map[string]struct{}, len(out.Metrics)) + metrics := make([]string, 0, len(out.Metrics)) + for _, metric := range out.Metrics { + if !isKnownMetric(metric) { + return GlobalSettings{}, fmt.Errorf("%w: unknown metric %q", ErrValidation, metric) + } + if _, ok := seen[metric]; ok { + continue + } + seen[metric] = struct{}{} + metrics = append(metrics, metric) + } + if len(metrics) == 0 { + return GlobalSettings{}, fmt.Errorf("%w: at least one metric is required", ErrValidation) + } + out.Metrics = metrics + return out, nil +} + +func isKnownMetric(metric string) bool { + switch metric { + case MetricCPUPercent, MetricMemoryPercent, MetricDiskPercent, MetricCPUTemp, MetricIOLoad: + return true + default: + return false + } +} diff --git a/apps/backend/internal/system/metrics/settings_test.go b/apps/backend/internal/system/metrics/settings_test.go new file mode 100644 index 0000000000..6e9a986dd8 --- /dev/null +++ b/apps/backend/internal/system/metrics/settings_test.go @@ -0,0 +1,98 @@ +package metrics + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +func TestNormalizeSettingsDefaults(t *testing.T) { + got, err := NormalizeSettings(GlobalSettings{}) + if err != nil { + t.Fatalf("NormalizeSettings returned error: %v", err) + } + + if got.IntervalSeconds != DefaultIntervalSeconds { + t.Fatalf("IntervalSeconds=%d, want %d", got.IntervalSeconds, DefaultIntervalSeconds) + } + if len(got.Metrics) != 3 { + t.Fatalf("metrics len=%d, want 3", len(got.Metrics)) + } + if got.CollectExecution { + t.Fatal("CollectExecution should default to false") + } +} + +func TestNormalizeSettingsValidatesIntervalAndMetrics(t *testing.T) { + _, err := NormalizeSettings(GlobalSettings{ + IntervalSeconds: 6 * 60, + Metrics: []string{MetricCPUPercent}, + }) + if err == nil { + t.Fatal("expected error for interval above max") + } + + got, err := NormalizeSettings(GlobalSettings{ + IntervalSeconds: 1, + Metrics: []string{MetricCPUPercent, MetricCPUPercent, MetricMemoryPercent}, + }) + if err != nil { + t.Fatalf("NormalizeSettings returned error: %v", err) + } + if len(got.Metrics) != 2 { + t.Fatalf("deduped metrics len=%d, want 2", len(got.Metrics)) + } + + _, err = NormalizeSettings(GlobalSettings{ + IntervalSeconds: 5, + Metrics: []string{"unknown"}, + }) + if err == nil { + t.Fatal("expected error for unknown metric") + } +} + +func TestCollectorResetClearsCPUBaseline(t *testing.T) { + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "stat"), []byte("cpu 1 0 0 9 0 0 0 0\n"), 0o600); err != nil { + t.Fatalf("write stat: %v", err) + } + collector := NewCollector() + collector.procRoot = dir + + if _, err := collector.cpuPercent(); err != nil { + t.Fatalf("cpuPercent baseline: %v", err) + } + if collector.prevCPU == nil { + t.Fatal("expected CPU baseline to be stored") + } + + collector.Reset() + if collector.prevCPU != nil { + t.Fatal("expected Reset to clear CPU baseline") + } + if !collector.lastCPUAt.IsZero() { + t.Fatal("expected Reset to clear CPU timestamp") + } +} + +func TestCollectorCPUPercentResetsStaleBaseline(t *testing.T) { + dir := t.TempDir() + statPath := filepath.Join(dir, "stat") + if err := os.WriteFile(statPath, []byte("cpu 1 0 0 9 0 0 0 0\n"), 0o600); err != nil { + t.Fatalf("write stat: %v", err) + } + collector := NewCollector() + collector.procRoot = dir + if value, err := collector.cpuPercent(); err != nil || value != 0 { + t.Fatalf("cpuPercent baseline=(%v, %v), want 0 nil", value, err) + } + collector.lastCPUAt = time.Now().Add(-time.Duration(2*MaxIntervalSeconds+1) * time.Second) + if err := os.WriteFile(statPath, []byte("cpu 10 0 0 10 0 0 0 0\n"), 0o600); err != nil { + t.Fatalf("rewrite stat: %v", err) + } + if value, err := collector.cpuPercent(); err != nil || value != 0 { + t.Fatalf("stale cpuPercent=(%v, %v), want 0 nil", value, err) + } +} diff --git a/apps/backend/internal/system/metrics/store.go b/apps/backend/internal/system/metrics/store.go new file mode 100644 index 0000000000..95ebc14e23 --- /dev/null +++ b/apps/backend/internal/system/metrics/store.go @@ -0,0 +1,49 @@ +package metrics + +import ( + "context" + "encoding/json" + "fmt" + + systemsettings "github.com/kandev/kandev/internal/system/settings" +) + +const settingsKey = "system_metrics" + +type Store struct { + settings *systemsettings.Store +} + +func NewStore(settings *systemsettings.Store) *Store { + return &Store{settings: settings} +} + +func (s *Store) GetSettings(ctx context.Context) (GlobalSettings, error) { + raw, found, err := s.settings.Get(ctx, settingsKey) + if err != nil { + return GlobalSettings{}, err + } + if !found { + return DefaultSettings(), nil + } + var settings GlobalSettings + if err := json.Unmarshal(raw, &settings); err != nil { + return GlobalSettings{}, err + } + return NormalizeSettings(settings) +} + +func (s *Store) SaveSettings(ctx context.Context, settings GlobalSettings) (GlobalSettings, error) { + normalized, err := NormalizeSettings(settings) + if err != nil { + return GlobalSettings{}, err + } + data, err := json.Marshal(normalized) + if err != nil { + return GlobalSettings{}, err + } + if err := s.settings.Save(ctx, settingsKey, data); err != nil { + return GlobalSettings{}, fmt.Errorf("save metrics settings: %w", err) + } + return normalized, nil +} diff --git a/apps/backend/internal/system/metrics/types.go b/apps/backend/internal/system/metrics/types.go new file mode 100644 index 0000000000..f528a90adf --- /dev/null +++ b/apps/backend/internal/system/metrics/types.go @@ -0,0 +1,28 @@ +package metrics + +import "time" + +type Snapshot struct { + Timestamp time.Time `json:"timestamp"` + IntervalSeconds int `json:"interval_seconds"` + Sources []SourceSnapshot `json:"sources"` +} + +type SourceSnapshot struct { + ID string `json:"id"` + Label string `json:"label"` + Kind string `json:"kind"` + ExecutorType string `json:"executor_type,omitempty"` + SessionID string `json:"session_id,omitempty"` + TaskID string `json:"task_id,omitempty"` + Metrics []MetricSample `json:"metrics"` +} + +type MetricSample struct { + ID string `json:"id"` + Label string `json:"label"` + Unit string `json:"unit,omitempty"` + Value *float64 `json:"value,omitempty"` + Available bool `json:"available"` + Error string `json:"error,omitempty"` +} diff --git a/apps/backend/internal/system/settings/store.go b/apps/backend/internal/system/settings/store.go new file mode 100644 index 0000000000..31f7ae1799 --- /dev/null +++ b/apps/backend/internal/system/settings/store.go @@ -0,0 +1,77 @@ +package settings + +import ( + "context" + "database/sql" + "time" + + "github.com/jmoiron/sqlx" + + "github.com/kandev/kandev/internal/db" +) + +// Store persists install-wide Kandev settings that are not scoped to a user, +// workspace, agent profile, or integration. +type Store struct { + db *sqlx.DB + ro *sqlx.DB +} + +func NewStore(pool *db.Pool) (*Store, error) { + store := &Store{db: pool.Writer(), ro: pool.Reader()} + if err := store.initSchema(); err != nil { + return nil, err + } + return store, nil +} + +func (s *Store) initSchema() error { + if _, err := s.db.Exec(` + CREATE TABLE IF NOT EXISTS settings ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + updated_at TIMESTAMP NOT NULL + ); + `); err != nil { + return err + } + return s.migrateLegacySystemSettings() +} + +func (s *Store) migrateLegacySystemSettings() error { + var exists int + if err := s.db.QueryRow(` + SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'system_settings' + `).Scan(&exists); err != nil { + return err + } + if exists == 0 { + return nil + } + _, err := s.db.Exec(` + INSERT OR IGNORE INTO settings (key, value, updated_at) + SELECT key, value, updated_at FROM system_settings + `) + return err +} + +func (s *Store) Get(ctx context.Context, key string) ([]byte, bool, error) { + var raw string + err := s.ro.QueryRowContext(ctx, s.ro.Rebind(`SELECT value FROM settings WHERE key = ?`), key).Scan(&raw) + if err != nil { + if err == sql.ErrNoRows { + return nil, false, nil + } + return nil, false, err + } + return []byte(raw), true, nil +} + +func (s *Store) Save(ctx context.Context, key string, value []byte) error { + _, err := s.db.ExecContext(ctx, s.db.Rebind(` + INSERT INTO settings (key, value, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at + `), key, string(value), time.Now().UTC()) + return err +} diff --git a/apps/backend/internal/system/settings/store_test.go b/apps/backend/internal/system/settings/store_test.go new file mode 100644 index 0000000000..a6b5bc2133 --- /dev/null +++ b/apps/backend/internal/system/settings/store_test.go @@ -0,0 +1,84 @@ +package settings + +import ( + "context" + "testing" + + "github.com/jmoiron/sqlx" + _ "github.com/mattn/go-sqlite3" + + "github.com/kandev/kandev/internal/db" +) + +func TestStoreSaveAndGet(t *testing.T) { + store := newTestStore(t) + ctx := context.Background() + + value, found, err := store.Get(ctx, "missing") + if err != nil { + t.Fatalf("get missing: %v", err) + } + if found || value != nil { + t.Fatalf("missing value = (%q, %v), want nil false", value, found) + } + + if err := store.Save(ctx, "system_metrics", []byte(`{"interval_seconds":5}`)); err != nil { + t.Fatalf("save: %v", err) + } + value, found, err = store.Get(ctx, "system_metrics") + if err != nil { + t.Fatalf("get saved: %v", err) + } + if !found || string(value) != `{"interval_seconds":5}` { + t.Fatalf("saved value = (%q, %v)", value, found) + } +} + +func TestStoreMigratesLegacySystemSettings(t *testing.T) { + conn := newSQLite(t) + if _, err := conn.Exec(` + CREATE TABLE system_settings ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + updated_at TIMESTAMP NOT NULL + ); + INSERT INTO system_settings (key, value, updated_at) + VALUES ('system_metrics', '{"interval_seconds":5}', CURRENT_TIMESTAMP); + `); err != nil { + t.Fatalf("seed legacy table: %v", err) + } + + store, err := NewStore(db.NewPool(conn, conn)) + if err != nil { + t.Fatalf("new store: %v", err) + } + + value, found, err := store.Get(context.Background(), "system_metrics") + if err != nil { + t.Fatalf("get migrated: %v", err) + } + if !found || string(value) != `{"interval_seconds":5}` { + t.Fatalf("migrated value = (%q, %v)", value, found) + } +} + +func newTestStore(t *testing.T) *Store { + t.Helper() + conn := newSQLite(t) + store, err := NewStore(db.NewPool(conn, conn)) + if err != nil { + t.Fatalf("new store: %v", err) + } + return store +} + +func newSQLite(t *testing.T) *sqlx.DB { + t.Helper() + conn, err := sqlx.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + conn.SetMaxOpenConns(1) + t.Cleanup(func() { _ = conn.Close() }) + return conn +} diff --git a/apps/backend/internal/system/system.go b/apps/backend/internal/system/system.go index e0e720682c..88a160d62b 100644 --- a/apps/backend/internal/system/system.go +++ b/apps/backend/internal/system/system.go @@ -25,7 +25,10 @@ import ( "github.com/kandev/kandev/internal/system/info" "github.com/kandev/kandev/internal/system/jobs" "github.com/kandev/kandev/internal/system/logs" + "github.com/kandev/kandev/internal/system/metrics" + systemsettings "github.com/kandev/kandev/internal/system/settings" "github.com/kandev/kandev/internal/system/updates" + "go.uber.org/zap" ) // BuildInfo holds the ldflag-injected build metadata that cmd/kandev @@ -55,6 +58,7 @@ type Service struct { Database *database.Service Backups *backups.Service Logs *logs.Service + Metrics *metrics.Service Updates *updates.Service } @@ -81,6 +85,15 @@ func Provide(cfg *config.Config, log *logger.Logger, pool *db.Pool, eventBus bus logDir := log.LogDirectory() logFile := log.LogFilename() + settingsStore, err := systemsettings.NewStore(pool) + if err != nil { + log.Error("Failed to initialize system settings store", zap.Error(err)) + } + var metricsSvc *metrics.Service + if settingsStore != nil { + metricsStore := metrics.NewStore(settingsStore) + metricsSvc = metrics.NewService(metricsStore, metrics.NewCollector()) + } return &Service{ Info: info.NewService(build.Version, build.Commit, build.BuildTime), @@ -89,6 +102,7 @@ func Provide(cfg *config.Config, log *logger.Logger, pool *db.Pool, eventBus bus Database: dbSvc, Backups: backupsSvc, Logs: logs.NewService(logDir, logFile, log), + Metrics: metricsSvc, Updates: updates.NewService(pool, build.Version, nil, log, updates.WithHomeDir(homeDir), updates.WithJobs(tracker)), } } @@ -116,6 +130,10 @@ func (s *Service) RegisterRoutes(router *gin.Engine, log *logger.Logger) { g.GET("/logs/tail", logs.HandleTail(s.Logs)) g.GET("/logs/:name/download", logs.HandleDownload(s.Logs)) + if s.Metrics != nil { + metrics.RegisterRoutes(g, s.Metrics) + } + g.GET("/updates", updates.HandleGet(s.Updates)) g.POST("/updates/check", updates.HandleCheck(s.Updates)) g.POST("/updates/apply", updates.HandleApply(s.Updates)) diff --git a/apps/backend/internal/user/controller/controller.go b/apps/backend/internal/user/controller/controller.go index 1ca49c98e6..700a00224a 100644 --- a/apps/backend/internal/user/controller/controller.go +++ b/apps/backend/internal/user/controller/controller.go @@ -68,6 +68,7 @@ func (c *Controller) UpdateUserSettings(ctx context.Context, req dto.UpdateUserS TerminalFontFamily: req.TerminalFontFamily, TerminalFontSize: req.TerminalFontSize, ChangesPanelLayout: req.ChangesPanelLayout, + SystemMetricsDisplay: req.SystemMetricsDisplay, VoiceMode: req.VoiceMode, }) if err != nil { diff --git a/apps/backend/internal/user/dto/dto.go b/apps/backend/internal/user/dto/dto.go index 3329aeb1ab..f0cac76e25 100644 --- a/apps/backend/internal/user/dto/dto.go +++ b/apps/backend/internal/user/dto/dto.go @@ -14,33 +14,34 @@ type UserDTO struct { } type UserSettingsDTO struct { - UserID string `json:"user_id"` - WorkspaceID string `json:"workspace_id"` - KanbanViewMode string `json:"kanban_view_mode"` - WorkflowFilterID string `json:"workflow_filter_id"` - RepositoryIDs []string `json:"repository_ids"` - InitialSetupComplete bool `json:"initial_setup_complete"` - PreferredShell string `json:"preferred_shell"` - DefaultEditorID string `json:"default_editor_id"` - EnablePreviewOnClick bool `json:"enable_preview_on_click"` - ChatSubmitKey string `json:"chat_submit_key"` - ReviewAutoMarkOnScroll bool `json:"review_auto_mark_on_scroll"` - ShowReleaseNotification bool `json:"show_release_notification"` - ReleaseNotesLastSeenVersion string `json:"release_notes_last_seen_version"` - LspAutoStartLanguages []string `json:"lsp_auto_start_languages"` - LspAutoInstallLanguages []string `json:"lsp_auto_install_languages"` - LspServerConfigs map[string]map[string]interface{} `json:"lsp_server_configs,omitempty"` - SavedLayouts []models.SavedLayout `json:"saved_layouts"` - SidebarViews []models.SidebarView `json:"sidebar_views"` - DefaultUtilityAgentID string `json:"default_utility_agent_id"` - DefaultUtilityModel string `json:"default_utility_model"` - KeyboardShortcuts map[string]interface{} `json:"keyboard_shortcuts,omitempty"` - TerminalLinkBehavior string `json:"terminal_link_behavior"` - TerminalFontFamily string `json:"terminal_font_family"` - TerminalFontSize int `json:"terminal_font_size"` - ChangesPanelLayout string `json:"changes_panel_layout"` - VoiceMode models.VoiceModeSettings `json:"voice_mode"` - UpdatedAt string `json:"updated_at"` + UserID string `json:"user_id"` + WorkspaceID string `json:"workspace_id"` + KanbanViewMode string `json:"kanban_view_mode"` + WorkflowFilterID string `json:"workflow_filter_id"` + RepositoryIDs []string `json:"repository_ids"` + InitialSetupComplete bool `json:"initial_setup_complete"` + PreferredShell string `json:"preferred_shell"` + DefaultEditorID string `json:"default_editor_id"` + EnablePreviewOnClick bool `json:"enable_preview_on_click"` + ChatSubmitKey string `json:"chat_submit_key"` + ReviewAutoMarkOnScroll bool `json:"review_auto_mark_on_scroll"` + ShowReleaseNotification bool `json:"show_release_notification"` + ReleaseNotesLastSeenVersion string `json:"release_notes_last_seen_version"` + LspAutoStartLanguages []string `json:"lsp_auto_start_languages"` + LspAutoInstallLanguages []string `json:"lsp_auto_install_languages"` + LspServerConfigs map[string]map[string]interface{} `json:"lsp_server_configs,omitempty"` + SavedLayouts []models.SavedLayout `json:"saved_layouts"` + SidebarViews []models.SidebarView `json:"sidebar_views"` + DefaultUtilityAgentID string `json:"default_utility_agent_id"` + DefaultUtilityModel string `json:"default_utility_model"` + KeyboardShortcuts map[string]interface{} `json:"keyboard_shortcuts,omitempty"` + TerminalLinkBehavior string `json:"terminal_link_behavior"` + TerminalFontFamily string `json:"terminal_font_family"` + TerminalFontSize int `json:"terminal_font_size"` + ChangesPanelLayout string `json:"changes_panel_layout"` + SystemMetricsDisplay models.SystemMetricsDisplaySettings `json:"system_metrics_display"` + VoiceMode models.VoiceModeSettings `json:"voice_mode"` + UpdatedAt string `json:"updated_at"` } type UserResponse struct { @@ -59,31 +60,32 @@ type ShellOption struct { } type UpdateUserSettingsRequest struct { - WorkspaceID *string `json:"workspace_id,omitempty"` - KanbanViewMode *string `json:"kanban_view_mode,omitempty"` - WorkflowFilterID *string `json:"workflow_filter_id,omitempty"` - RepositoryIDs *[]string `json:"repository_ids,omitempty"` - InitialSetupComplete *bool `json:"initial_setup_complete,omitempty"` - PreferredShell *string `json:"preferred_shell,omitempty"` - DefaultEditorID *string `json:"default_editor_id,omitempty"` - EnablePreviewOnClick *bool `json:"enable_preview_on_click,omitempty"` - ChatSubmitKey *string `json:"chat_submit_key,omitempty"` - ReviewAutoMarkOnScroll *bool `json:"review_auto_mark_on_scroll,omitempty"` - ShowReleaseNotification *bool `json:"show_release_notification,omitempty"` - ReleaseNotesLastSeenVersion *string `json:"release_notes_last_seen_version,omitempty"` - LspAutoStartLanguages *[]string `json:"lsp_auto_start_languages,omitempty"` - LspAutoInstallLanguages *[]string `json:"lsp_auto_install_languages,omitempty"` - LspServerConfigs *map[string]map[string]interface{} `json:"lsp_server_configs,omitempty"` - SavedLayouts *[]models.SavedLayout `json:"saved_layouts,omitempty"` - SidebarViews *[]models.SidebarView `json:"sidebar_views,omitempty"` - DefaultUtilityAgentID *string `json:"default_utility_agent_id,omitempty"` - DefaultUtilityModel *string `json:"default_utility_model,omitempty"` - KeyboardShortcuts *map[string]interface{} `json:"keyboard_shortcuts,omitempty"` - TerminalLinkBehavior *string `json:"terminal_link_behavior,omitempty"` - TerminalFontFamily *string `json:"terminal_font_family,omitempty"` - TerminalFontSize *int `json:"terminal_font_size,omitempty"` - ChangesPanelLayout *string `json:"changes_panel_layout,omitempty"` - VoiceMode *models.VoiceModeSettings `json:"voice_mode,omitempty"` + WorkspaceID *string `json:"workspace_id,omitempty"` + KanbanViewMode *string `json:"kanban_view_mode,omitempty"` + WorkflowFilterID *string `json:"workflow_filter_id,omitempty"` + RepositoryIDs *[]string `json:"repository_ids,omitempty"` + InitialSetupComplete *bool `json:"initial_setup_complete,omitempty"` + PreferredShell *string `json:"preferred_shell,omitempty"` + DefaultEditorID *string `json:"default_editor_id,omitempty"` + EnablePreviewOnClick *bool `json:"enable_preview_on_click,omitempty"` + ChatSubmitKey *string `json:"chat_submit_key,omitempty"` + ReviewAutoMarkOnScroll *bool `json:"review_auto_mark_on_scroll,omitempty"` + ShowReleaseNotification *bool `json:"show_release_notification,omitempty"` + ReleaseNotesLastSeenVersion *string `json:"release_notes_last_seen_version,omitempty"` + LspAutoStartLanguages *[]string `json:"lsp_auto_start_languages,omitempty"` + LspAutoInstallLanguages *[]string `json:"lsp_auto_install_languages,omitempty"` + LspServerConfigs *map[string]map[string]interface{} `json:"lsp_server_configs,omitempty"` + SavedLayouts *[]models.SavedLayout `json:"saved_layouts,omitempty"` + SidebarViews *[]models.SidebarView `json:"sidebar_views,omitempty"` + DefaultUtilityAgentID *string `json:"default_utility_agent_id,omitempty"` + DefaultUtilityModel *string `json:"default_utility_model,omitempty"` + KeyboardShortcuts *map[string]interface{} `json:"keyboard_shortcuts,omitempty"` + TerminalLinkBehavior *string `json:"terminal_link_behavior,omitempty"` + TerminalFontFamily *string `json:"terminal_font_family,omitempty"` + TerminalFontSize *int `json:"terminal_font_size,omitempty"` + ChangesPanelLayout *string `json:"changes_panel_layout,omitempty"` + SystemMetricsDisplay *models.SystemMetricsDisplaySettings `json:"system_metrics_display,omitempty"` + VoiceMode *models.VoiceModeSettings `json:"voice_mode,omitempty"` } func FromUser(user *models.User) UserDTO { @@ -122,6 +124,7 @@ func FromUserSettings(settings *models.UserSettings) UserSettingsDTO { TerminalFontFamily: settings.TerminalFontFamily, TerminalFontSize: settings.TerminalFontSize, ChangesPanelLayout: settings.ChangesPanelLayout, + SystemMetricsDisplay: settings.SystemMetricsDisplay, VoiceMode: settings.VoiceMode, UpdatedAt: settings.UpdatedAt.Format(time.RFC3339), } diff --git a/apps/backend/internal/user/models/models.go b/apps/backend/internal/user/models/models.go index 80475e9041..3e29cb1613 100644 --- a/apps/backend/internal/user/models/models.go +++ b/apps/backend/internal/user/models/models.go @@ -38,11 +38,16 @@ type UserSettings struct { TerminalFontFamily string `json:"terminal_font_family"` TerminalFontSize int `json:"terminal_font_size"` ChangesPanelLayout string `json:"changes_panel_layout"` // "flat" | "tree" + SystemMetricsDisplay SystemMetricsDisplaySettings `json:"system_metrics_display"` VoiceMode VoiceModeSettings `json:"voice_mode"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } +type SystemMetricsDisplaySettings struct { + ShowInTopbar bool `json:"show_in_topbar"` +} + // VoiceModeSettings is the per-user configuration surface for the chat // voice-input feature. Stored as a nested JSON object inside the `users.settings` // blob — adding fields here does not require a schema migration. diff --git a/apps/backend/internal/user/service/service.go b/apps/backend/internal/user/service/service.go index 1a27093354..59b83e0ce4 100644 --- a/apps/backend/internal/user/service/service.go +++ b/apps/backend/internal/user/service/service.go @@ -58,6 +58,7 @@ type UpdateUserSettingsRequest struct { TerminalFontFamily *string TerminalFontSize *int ChangesPanelLayout *string + SystemMetricsDisplay *models.SystemMetricsDisplaySettings VoiceMode *models.VoiceModeSettings } @@ -187,6 +188,9 @@ func applyBasicSettings(settings *models.UserSettings, req *UpdateUserSettingsRe if err := applyChangesPanelLayout(settings, req.ChangesPanelLayout); err != nil { return err } + if req.SystemMetricsDisplay != nil { + settings.SystemMetricsDisplay = *req.SystemMetricsDisplay + } if req.TerminalFontFamily != nil { settings.TerminalFontFamily = strings.TrimSpace(*req.TerminalFontFamily) } @@ -396,6 +400,7 @@ func (s *Service) publishUserSettingsEvent(ctx context.Context, settings *models "terminal_font_family": settings.TerminalFontFamily, "terminal_font_size": settings.TerminalFontSize, "changes_panel_layout": settings.ChangesPanelLayout, + "system_metrics_display": settings.SystemMetricsDisplay, "voice_mode": settings.VoiceMode, "updated_at": settings.UpdatedAt.Format(time.RFC3339), } diff --git a/apps/backend/internal/user/store/sqlite.go b/apps/backend/internal/user/store/sqlite.go index b9dcf18a2a..daf7815358 100644 --- a/apps/backend/internal/user/store/sqlite.go +++ b/apps/backend/internal/user/store/sqlite.go @@ -162,6 +162,7 @@ func (r *sqliteRepository) UpsertUserSettings(ctx context.Context, settings *mod "terminal_font_family": settings.TerminalFontFamily, "terminal_font_size": settings.TerminalFontSize, "changes_panel_layout": settings.ChangesPanelLayout, + "system_metrics_display": settings.SystemMetricsDisplay, "voice_mode": settings.VoiceMode, }) if err != nil { @@ -265,31 +266,32 @@ func scanUserSettings(scanner interface{ Scan(dest ...any) error }, userID strin return settings, nil } var payload struct { - WorkspaceID string `json:"workspace_id"` - KanbanViewMode string `json:"kanban_view_mode"` - WorkflowFilterID string `json:"workflow_filter_id"` - RepositoryIDs []string `json:"repository_ids"` - InitialSetupComplete bool `json:"initial_setup_complete"` - PreferredShell string `json:"preferred_shell"` - DefaultEditorID string `json:"default_editor_id"` - EnablePreviewOnClick bool `json:"enable_preview_on_click"` - ChatSubmitKey string `json:"chat_submit_key"` - ReviewAutoMarkOnScroll *bool `json:"review_auto_mark_on_scroll"` - ShowReleaseNotification *bool `json:"show_release_notification"` - ReleaseNotesLastSeenVersion string `json:"release_notes_last_seen_version"` - LspAutoStartLanguages []string `json:"lsp_auto_start_languages"` - LspAutoInstallLanguages []string `json:"lsp_auto_install_languages"` - LspServerConfigs map[string]map[string]interface{} `json:"lsp_server_configs"` - SavedLayouts []models.SavedLayout `json:"saved_layouts"` - SidebarViews []models.SidebarView `json:"sidebar_views"` - DefaultUtilityAgentID string `json:"default_utility_agent_id"` - DefaultUtilityModel string `json:"default_utility_model"` - KeyboardShortcuts map[string]interface{} `json:"keyboard_shortcuts"` - TerminalLinkBehavior string `json:"terminal_link_behavior"` - TerminalFontFamily string `json:"terminal_font_family"` - TerminalFontSize int `json:"terminal_font_size"` - ChangesPanelLayout string `json:"changes_panel_layout"` - VoiceMode *storedVoiceMode `json:"voice_mode"` + WorkspaceID string `json:"workspace_id"` + KanbanViewMode string `json:"kanban_view_mode"` + WorkflowFilterID string `json:"workflow_filter_id"` + RepositoryIDs []string `json:"repository_ids"` + InitialSetupComplete bool `json:"initial_setup_complete"` + PreferredShell string `json:"preferred_shell"` + DefaultEditorID string `json:"default_editor_id"` + EnablePreviewOnClick bool `json:"enable_preview_on_click"` + ChatSubmitKey string `json:"chat_submit_key"` + ReviewAutoMarkOnScroll *bool `json:"review_auto_mark_on_scroll"` + ShowReleaseNotification *bool `json:"show_release_notification"` + ReleaseNotesLastSeenVersion string `json:"release_notes_last_seen_version"` + LspAutoStartLanguages []string `json:"lsp_auto_start_languages"` + LspAutoInstallLanguages []string `json:"lsp_auto_install_languages"` + LspServerConfigs map[string]map[string]interface{} `json:"lsp_server_configs"` + SavedLayouts []models.SavedLayout `json:"saved_layouts"` + SidebarViews []models.SidebarView `json:"sidebar_views"` + DefaultUtilityAgentID string `json:"default_utility_agent_id"` + DefaultUtilityModel string `json:"default_utility_model"` + KeyboardShortcuts map[string]interface{} `json:"keyboard_shortcuts"` + TerminalLinkBehavior string `json:"terminal_link_behavior"` + TerminalFontFamily string `json:"terminal_font_family"` + TerminalFontSize int `json:"terminal_font_size"` + ChangesPanelLayout string `json:"changes_panel_layout"` + SystemMetricsDisplay models.SystemMetricsDisplaySettings `json:"system_metrics_display"` + VoiceMode *storedVoiceMode `json:"voice_mode"` } if err := json.Unmarshal([]byte(settingsRaw), &payload); err != nil { return nil, err @@ -350,6 +352,7 @@ func scanUserSettings(scanner interface{ Scan(dest ...any) error }, userID strin settings.TerminalFontFamily = payload.TerminalFontFamily settings.TerminalFontSize = payload.TerminalFontSize settings.VoiceMode = mergeVoiceModeDefaults(payload.VoiceMode) + settings.SystemMetricsDisplay = payload.SystemMetricsDisplay if payload.ChangesPanelLayout == "flat" { settings.ChangesPanelLayout = "flat" } else { diff --git a/apps/backend/internal/user/store/sqlite_test.go b/apps/backend/internal/user/store/sqlite_test.go index c6c7d92d5b..2a762818d1 100644 --- a/apps/backend/internal/user/store/sqlite_test.go +++ b/apps/backend/internal/user/store/sqlite_test.go @@ -1,8 +1,14 @@ package store import ( + "context" "testing" "time" + + "github.com/jmoiron/sqlx" + _ "github.com/mattn/go-sqlite3" + + "github.com/kandev/kandev/internal/user/models" ) type settingsScanner struct { @@ -46,3 +52,51 @@ func TestScanUserSettingsChangesPanelLayoutDefault(t *testing.T) { } }) } + +func TestScanUserSettingsSystemMetricsDisplayDefault(t *testing.T) { + settings, err := scanUserSettings(settingsScanner{raw: "{}"}, DefaultUserID) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if settings.SystemMetricsDisplay.ShowInTopbar { + t.Fatal("system metrics display should default to disabled") + } + + settings, err = scanUserSettings(settingsScanner{raw: `{"system_metrics_display":{"show_in_topbar":true}}`}, DefaultUserID) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !settings.SystemMetricsDisplay.ShowInTopbar { + t.Fatal("expected stored system metrics display preference") + } +} + +func TestSQLiteRepositorySystemMetricsDisplayRoundTrip(t *testing.T) { + conn, err := sqlx.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + conn.SetMaxOpenConns(1) + t.Cleanup(func() { _ = conn.Close() }) + repo, err := newSQLiteRepositoryWithDB(conn, conn) + if err != nil { + t.Fatalf("new repo: %v", err) + } + + ctx := context.Background() + settings, err := repo.GetUserSettings(ctx, DefaultUserID) + if err != nil { + t.Fatalf("get defaults: %v", err) + } + settings.SystemMetricsDisplay = models.SystemMetricsDisplaySettings{ShowInTopbar: true} + if err := repo.UpsertUserSettings(ctx, settings); err != nil { + t.Fatalf("upsert settings: %v", err) + } + got, err := repo.GetUserSettings(ctx, DefaultUserID) + if err != nil { + t.Fatalf("get settings: %v", err) + } + if !got.SystemMetricsDisplay.ShowInTopbar { + t.Fatal("expected system metrics display preference to round-trip") + } +} diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 9eb5cec343..0184d436e8 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -134,12 +134,14 @@ const ( // Focus signals are layered on top of subscriptions to indicate which // session the user is actively viewing (task details page or task panel), // vs merely subscribed (sidebar diff badges). Drives backend polling tier. - ActionSessionFocus = "session.focus" - ActionSessionUnfocus = "session.unfocus" - ActionUserSubscribe = "user.subscribe" - ActionUserUnsubscribe = "user.unsubscribe" - ActionRunSubscribe = "run.subscribe" - ActionRunUnsubscribe = "run.unsubscribe" + ActionSessionFocus = "session.focus" + ActionSessionUnfocus = "session.unfocus" + ActionUserSubscribe = "user.subscribe" + ActionUserUnsubscribe = "user.unsubscribe" + ActionRunSubscribe = "run.subscribe" + ActionRunUnsubscribe = "run.unsubscribe" + ActionSystemMetricsSubscribe = "system.metrics.subscribe" + ActionSystemMetricsUnsubscribe = "system.metrics.unsubscribe" // Message actions ActionMessageAdd = "message.add" @@ -213,6 +215,7 @@ const ( ActionExecutorProfileDeleted = "executor.profile.deleted" ActionExecutorPrepareProgress = "executor.prepare.progress" ActionExecutorPrepareCompleted = "executor.prepare.completed" + ActionSystemMetricsUpdated = "system.metrics.updated" ActionAgentProfileDeleted = "agent.profile.deleted" ActionAgentProfileCreated = "agent.profile.created" diff --git a/apps/web/components/kanban/kanban-header.tsx b/apps/web/components/kanban/kanban-header.tsx index 5ed045331c..d4ec212361 100644 --- a/apps/web/components/kanban/kanban-header.tsx +++ b/apps/web/components/kanban/kanban-header.tsx @@ -19,6 +19,7 @@ import { useAppStore } from "@/components/state-provider"; import { useKanbanDisplaySettings } from "@/hooks/use-kanban-display-settings"; import { useReleaseNotes } from "@/hooks/use-release-notes"; import { useSystemHealthIndicator } from "@/hooks/use-system-health-indicator"; +import { TopbarMetrics } from "@/components/system-metrics/topbar-metrics"; import type { ComponentProps, RefObject } from "react"; type KanbanHeaderProps = { @@ -169,6 +170,7 @@ function TabletHeader({ className="hidden md:flex w-48 lg:w-56 [&_input]:h-8" /> )} + @@ -244,6 +246,7 @@ function DesktopHeader({ actions={ <> {actionsSearch} + diff --git a/apps/web/components/settings/editors-settings-state.tsx b/apps/web/components/settings/editors-settings-state.tsx index bb9bbadf7a..a92de77506 100644 --- a/apps/web/components/settings/editors-settings-state.tsx +++ b/apps/web/components/settings/editors-settings-state.tsx @@ -9,6 +9,7 @@ import type { EditorOption } from "@/lib/types/http"; import { type ComboboxOption } from "@/components/combobox"; import { parseChangesPanelLayout, + parseSystemMetricsDisplay, parseTerminalLinkBehavior, parseVoiceMode, } from "@/lib/ssr/user-settings"; @@ -249,6 +250,7 @@ function buildUserSettingsFromResponse( terminalFontFamily: s.terminal_font_family || null, terminalFontSize: s.terminal_font_size || null, changesPanelLayout: parseChangesPanelLayout(s.changes_panel_layout), + systemMetricsDisplay: parseSystemMetricsDisplay(s.system_metrics_display), voiceMode: parseVoiceMode(s.voice_mode), ...mapEditorSettingsFields(s), }; diff --git a/apps/web/components/settings/general-settings.tsx b/apps/web/components/settings/general-settings.tsx index 0203d71311..bea34b80e4 100644 --- a/apps/web/components/settings/general-settings.tsx +++ b/apps/web/components/settings/general-settings.tsx @@ -9,6 +9,7 @@ import { IconKeyboard, IconTerminal2, IconGitBranch, + IconActivity, } from "@tabler/icons-react"; import { Badge } from "@kandev/ui/badge"; import { Card, CardContent, CardHeader, CardTitle } from "@kandev/ui/card"; @@ -28,6 +29,7 @@ import { Separator } from "@kandev/ui/separator"; import { SettingsSection } from "@/components/settings/settings-section"; import { ShellSettingsCard } from "@/components/settings/shell-settings-card"; import { KeyboardShortcutsCard } from "@/components/settings/keyboard-shortcuts-card"; +import { SystemMetricsSettingsCard } from "@/components/settings/system-metrics-settings-card"; import { getBackendConfig } from "@/lib/config"; import { useAppStore, useAppStoreApi } from "@/components/state-provider"; import { updateUserSettings } from "@/lib/api"; @@ -500,6 +502,16 @@ export function GeneralSettings() { + } + title="Resource Metrics" + description="Configure backend and execution resource sampling" + > + + + + + } title="Chat Input" diff --git a/apps/web/components/settings/system-metrics-settings-card.tsx b/apps/web/components/settings/system-metrics-settings-card.tsx new file mode 100644 index 0000000000..4ffc54d869 --- /dev/null +++ b/apps/web/components/settings/system-metrics-settings-card.tsx @@ -0,0 +1,246 @@ +"use client"; + +import { useEffect, useRef, useState } from "react"; +import { Checkbox } from "@kandev/ui/checkbox"; +import { Card, CardContent, CardHeader, CardTitle } from "@kandev/ui/card"; +import { Input } from "@kandev/ui/input"; +import { Label } from "@kandev/ui/label"; +import { Switch } from "@kandev/ui/switch"; +import { useAppStore, useAppStoreApi } from "@/components/state-provider"; +import { + fetchSystemMetricsSettings, + updateSystemMetricsSettings, + updateUserSettings, +} from "@/lib/api"; +import type { SystemMetricId, SystemMetricsGlobalSettings } from "@/lib/types/system"; + +const METRIC_OPTIONS: Array<{ id: SystemMetricId; label: string }> = [ + { id: "cpu_percent", label: "CPU %" }, + { id: "memory_percent", label: "Memory %" }, + { id: "disk_percent", label: "Disk %" }, + { id: "cpu_temp", label: "CPU temp" }, + { id: "io_load", label: "Load avg" }, +]; + +const DEFAULT_METRICS_SETTINGS: SystemMetricsGlobalSettings = { + metrics: ["cpu_percent", "memory_percent", "disk_percent"], + interval_seconds: 5, + backend_disk_path: "/", + collect_execution: false, +}; + +export function SystemMetricsSettingsCard() { + const storeApi = useAppStoreApi(); + const userSettings = useAppStore((state) => state.userSettings); + const setUserSettings = useAppStore((state) => state.setUserSettings); + const [settings, setSettings] = useState(DEFAULT_METRICS_SETTINGS); + const [isSaving, setIsSaving] = useState(false); + const saveSeqRef = useRef(0); + + useEffect(() => { + let cancelled = false; + fetchSystemMetricsSettings() + .then((res) => { + if (!cancelled) setSettings(res.settings); + }) + .catch(() => {}); + return () => { + cancelled = true; + }; + }, []); + + const saveGlobal = async (next: SystemMetricsGlobalSettings) => { + const previous = settings; + const seq = saveSeqRef.current + 1; + saveSeqRef.current = seq; + setSettings(next); + setIsSaving(true); + try { + const res = await updateSystemMetricsSettings(next); + if (seq === saveSeqRef.current) setSettings(res.settings); + } catch { + if (seq === saveSeqRef.current) setSettings(previous); + } finally { + if (seq === saveSeqRef.current) setIsSaving(false); + } + }; + + const toggleMetric = (metric: SystemMetricId, checked: boolean) => { + const nextMetrics = checked + ? Array.from(new Set([...settings.metrics, metric])) + : settings.metrics.filter((id) => id !== metric); + if (nextMetrics.length > 0) void saveGlobal({ ...settings, metrics: nextMetrics }); + }; + + const toggleDisplay = async (checked: boolean) => { + const current = storeApi.getState().userSettings; + const previous = current.systemMetricsDisplay; + try { + setUserSettings({ ...current, systemMetricsDisplay: { showInTopbar: checked } }); + await updateUserSettings({ + workspace_id: current.workspaceId || "", + repository_ids: current.repositoryIds || [], + system_metrics_display: { show_in_topbar: checked }, + }); + } catch { + setUserSettings({ ...storeApi.getState().userSettings, systemMetricsDisplay: previous }); + } + }; + + return ( + + + Resource Metrics + + +

+ Useful when Kandev is self-hosted on a remote server and you want a lightweight view of + the machine resources from the kanban or task topbar. +

+ + void saveGlobal(next)} + onDraftSettings={setSettings} + /> + + void saveGlobal({ ...settings, collect_execution: checked }) + } + /> + + + ); +} + +function MetricsDisplayToggle({ + checked, + onCheckedChange, +}: { + checked: boolean; + onCheckedChange: (checked: boolean) => void; +}) { + return ( +
+
+ +

+ Collection starts only while at least one desktop or tablet client displays metrics. +

+
+ +
+ ); +} + +function MetricsSamplerControls({ + settings, + isSaving, + onToggleMetric, + onChangeSettings, + onDraftSettings, +}: { + settings: SystemMetricsGlobalSettings; + isSaving: boolean; + onToggleMetric: (metric: SystemMetricId, checked: boolean) => void; + onChangeSettings: (settings: SystemMetricsGlobalSettings) => void; + onDraftSettings: (settings: SystemMetricsGlobalSettings) => void; +}) { + return ( +
+ +
+ + + onChangeSettings({ ...settings, interval_seconds: clampInterval(event.target.value) }) + } + /> +
+
+ + + onDraftSettings({ ...settings, backend_disk_path: event.target.value }) + } + onBlur={() => onChangeSettings(settings)} + /> +
+
+ ); +} + +function MetricCheckboxes({ + settings, + isSaving, + onToggleMetric, +}: { + settings: SystemMetricsGlobalSettings; + isSaving: boolean; + onToggleMetric: (metric: SystemMetricId, checked: boolean) => void; +}) { + return ( +
+ +
+ {METRIC_OPTIONS.map((metric) => ( + + ))} +
+
+ ); +} + +function ExecutionMetricsToggle({ + checked, + disabled, + onCheckedChange, +}: { + checked: boolean; + disabled: boolean; + onCheckedChange: (checked: boolean) => void; +}) { + return ( +
+
+ +

+ Adds agentctl values for Docker, SSH, Sprites, and remote executors. +

+
+ +
+ ); +} + +function clampInterval(raw: string) { + return Math.min(300, Math.max(1, Number(raw) || 5)); +} diff --git a/apps/web/components/state-provider.test.tsx b/apps/web/components/state-provider.test.tsx new file mode 100644 index 0000000000..bbd5014a7e --- /dev/null +++ b/apps/web/components/state-provider.test.tsx @@ -0,0 +1,57 @@ +import { fireEvent, render, screen } from "@testing-library/react"; +import { describe, expect, it } from "vitest"; +import { defaultSettingsState } from "@/lib/state/slices/settings/settings-slice"; +import { StateProvider, useAppStore } from "./state-provider"; + +function ShowMetricsPreference({ label }: { label: string }) { + const enabled = useAppStore((state) => state.userSettings.systemMetricsDisplay.showInTopbar); + return ( +
+ {label}:{enabled ? "on" : "off"} +
+ ); +} + +function EnableMetricsFromNestedProvider() { + const setUserSettings = useAppStore((state) => state.setUserSettings); + const userSettings = useAppStore((state) => state.userSettings); + return ( + + ); +} + +describe("StateProvider", () => { + it("reuses the parent store for nested route providers", async () => { + render( + + + + + + , + ); + + expect(screen.getByText("root:off")).toBeTruthy(); + fireEvent.click(screen.getByRole("button", { name: "Enable metrics" })); + expect(await screen.findByText("root:on")).toBeTruthy(); + }); +}); diff --git a/apps/web/components/state-provider.tsx b/apps/web/components/state-provider.tsx index 3d2f1dc68d..02201b37cb 100644 --- a/apps/web/components/state-provider.tsx +++ b/apps/web/components/state-provider.tsx @@ -1,6 +1,6 @@ "use client"; -import { createContext, useContext, useEffect, useState } from "react"; +import { createContext, useContext, useEffect, useLayoutEffect, useState } from "react"; import type { StoreApi } from "zustand"; import { useStore } from "zustand"; import { isDebug, registerSessionTaskResolver } from "@/lib/debug/log"; @@ -15,7 +15,14 @@ type E2EWindow = Window & { }; export function StateProvider({ children, initialState }: StoreProviderProps) { - const [store] = useState(() => createAppStore(initialState)); + const parentStore = useContext(StoreContext); + const [ownStore] = useState(() => createAppStore(parentStore ? undefined : initialState)); + const store = parentStore ?? ownStore; + + useLayoutEffect(() => { + if (!parentStore || !initialState || Object.keys(initialState).length === 0) return; + store.getState().hydrate(initialState); + }, [initialState, parentStore, store]); useEffect(() => { const win = window as E2EWindow; diff --git a/apps/web/components/system-metrics/topbar-metrics.tsx b/apps/web/components/system-metrics/topbar-metrics.tsx new file mode 100644 index 0000000000..970c14cd2c --- /dev/null +++ b/apps/web/components/system-metrics/topbar-metrics.tsx @@ -0,0 +1,213 @@ +"use client"; + +import { + IconActivity, + IconCpu, + IconDatabase, + IconDeviceDesktopAnalytics, + IconFlame, + IconGauge, + IconDisc, + IconServer, +} from "@tabler/icons-react"; +import { Tooltip, TooltipContent, TooltipTrigger } from "@kandev/ui/tooltip"; +import { formatDistanceToNow } from "date-fns"; +import { useAppStore } from "@/components/state-provider"; +import { useResponsiveBreakpoint } from "@/hooks/use-responsive-breakpoint"; +import { useSystemMetricsSubscription } from "@/hooks/use-system-metrics-subscription"; +import type { SystemMetricSample, SystemMetricsSource } from "@/lib/types/system"; + +type TopbarMetricsProps = { + activeSessionId?: string | null; +}; + +export function TopbarMetrics({ activeSessionId }: TopbarMetricsProps) { + const enabled = useAppStore((s) => s.userSettings.systemMetricsDisplay.showInTopbar); + const snapshot = useAppStore((s) => s.system.metrics); + const { isMobile } = useResponsiveBreakpoint(); + const shouldRender = enabled && !isMobile; + useSystemMetricsSubscription(shouldRender); + + if (!shouldRender) return null; + const sources = selectSources(snapshot?.sources ?? [], activeSessionId); + if (sources.length === 0) { + return ( +
+ + Metrics +
+ ); + } + return ( +
+ {sources.map((source) => ( + 1} + /> + ))} +
+ ); +} + +function selectSources(sources: SystemMetricsSource[], activeSessionId?: string | null) { + if (!activeSessionId) return sources.slice(0, 2); + const backend = sources.find((source) => source.kind === "backend"); + const execution = sources.find((source) => source.session_id === activeSessionId); + return [backend, execution].filter(Boolean) as SystemMetricsSource[]; +} + +function SourceMetrics({ + source, + updatedAt, + showSource, +}: { + source: SystemMetricsSource; + updatedAt?: string; + showSource: boolean; +}) { + const metrics = source.metrics.slice(0, 4); + + return ( +
+ {showSource ? : null} + {metrics.length > 0 ? ( + metrics.map((metric) => ( + + )) + ) : ( + - + )} +
+ ); +} + +function SourceBadge({ source, updatedAt }: { source: SystemMetricsSource; updatedAt?: string }) { + const isHost = source.kind === "backend"; + const label = isHost ? "Host" : "Executor"; + + return ( + + + + {isHost ? ( + + ) : ( + + )} + + + +
+
{label}
+
{source.label}
+
{lastUpdatedText(updatedAt)}
+
+ + + ); +} + +function MetricChip({ + metric, + source, + updatedAt, +}: { + metric: SystemMetricSample; + source: SystemMetricsSource; + updatedAt?: string; +}) { + return ( + + + + {metricIcon(metric.id)} + {formatMetric(metric)} + + + +
+
{metricLabel(metric.id)}
+
+ {source.kind === "backend" ? "Host" : "Executor"}: {source.label} +
+
{formatMetric(metric)}
+ {metric.error ? ( +
{metric.error}
+ ) : null} +
{lastUpdatedText(updatedAt)}
+
+ + + ); +} + +function metricLabel(id: string) { + switch (id) { + case "cpu_percent": + return "CPU"; + case "memory_percent": + return "Memory"; + case "disk_percent": + return "Disk"; + case "cpu_temp": + return "CPU temperature"; + case "io_load": + return "Load avg"; + default: + return id; + } +} + +function metricIcon(id: string) { + switch (id) { + case "cpu_percent": + return ; + case "memory_percent": + return ; + case "disk_percent": + return ; + case "cpu_temp": + return ; + case "io_load": + return ; + default: + return ; + } +} + +function formatMetric(metric: SystemMetricSample) { + if (typeof metric.value !== "number") return "-"; + const value = metric.unit === "%" ? Math.round(metric.value) : Math.round(metric.value * 10) / 10; + return `${value}${metric.unit ?? ""}`; +} + +function metricColor(metric: SystemMetricSample) { + if (!metric.available) return "text-muted-foreground"; + const thresholdValue = metricThresholdValue(metric); + if (thresholdValue === null) return "text-muted-foreground"; + if (thresholdValue > 95) return "text-destructive"; + if (thresholdValue >= 80) return "text-yellow-500 dark:text-yellow-400"; + return "text-muted-foreground"; +} + +function metricThresholdValue(metric: SystemMetricSample) { + if (typeof metric.value !== "number") return null; + if (metric.unit === "%" || metric.id === "cpu_temp") return metric.value; + return null; +} + +function lastUpdatedText(updatedAt?: string) { + if (!updatedAt) return "Last update unknown"; + const date = new Date(updatedAt); + if (Number.isNaN(date.getTime())) return "Last update unknown"; + return `Updated ${formatDistanceToNow(date, { addSuffix: true })}`; +} diff --git a/apps/web/components/task/task-top-bar.test.tsx b/apps/web/components/task/task-top-bar.test.tsx index f79e9e8a27..edaddfb4fb 100644 --- a/apps/web/components/task/task-top-bar.test.tsx +++ b/apps/web/components/task/task-top-bar.test.tsx @@ -1,5 +1,6 @@ import { cleanup, render, screen } from "@testing-library/react"; import { afterEach, describe, expect, it, vi } from "vitest"; +import { StateProvider } from "@/components/state-provider"; import { TaskTopBar } from "./task-top-bar"; afterEach(() => cleanup()); @@ -90,14 +91,18 @@ vi.mock("@/components/task/branch-path-popover", () => ({ describe("TaskTopBar executor environment controls", () => { it("hides the executor environment button for filesystem executors", () => { - render(); + renderTopBar(); expect(screen.queryByTestId("executor-settings-button")).toBeNull(); }); it("shows the executor environment button for Docker executors", () => { - render(); + renderTopBar(); expect(screen.getByTestId("executor-settings-button")).toBeTruthy(); }); }); + +function renderTopBar(ui: React.ReactNode) { + return render({ui}); +} diff --git a/apps/web/components/task/task-top-bar.tsx b/apps/web/components/task/task-top-bar.tsx index e9190cee81..4a9d4325a5 100644 --- a/apps/web/components/task/task-top-bar.tsx +++ b/apps/web/components/task/task-top-bar.tsx @@ -20,6 +20,7 @@ import { useLinearAvailable } from "@/hooks/domains/linear/use-linear-availabili import { PortForwardButton } from "@/components/task/port-forward-dialog"; import { ExecutorSettingsButton } from "@/components/task/executor-settings-button"; import { WorkflowStepper, type WorkflowStepperStep } from "@/components/task/workflow-stepper"; +import { TopbarMetrics } from "@/components/system-metrics/topbar-metrics"; import { isDebugUI } from "@/lib/config"; type TaskTopBarProps = { @@ -317,6 +318,7 @@ function TopBarRight({ }) { return (
+ {officeTaskHref && (