feat: add resource metrics topbar (#1373)

This commit is contained in:
Carlos Florêncio
2026-06-14 22:33:34 +01:00
committed by GitHub
parent d6cf4a1acc
commit 3a3fca7f0e
57 changed files with 2679 additions and 115 deletions
+5
View File
@@ -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)
@@ -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
}
}
@@ -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))
}
@@ -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
}
@@ -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
}
@@ -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)
@@ -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
+103 -12
View File
@@ -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()
@@ -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,
}
}
@@ -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")
}
}
@@ -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 = &current
c.lastCPUAt = now
return 0, nil
}
value := calculateCPUPercent(*c.prevCPU, current)
c.prevCPU = &current
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)}},
}
}
@@ -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
}
@@ -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")
}
}
@@ -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")
}
@@ -0,0 +1,11 @@
package metrics
import (
"testing"
"go.uber.org/goleak"
)
func TestMain(m *testing.M) {
goleak.VerifyTestMain(m)
}
@@ -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})
}
}
@@ -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
}
@@ -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
}
@@ -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
}
}
@@ -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)
}
}
@@ -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
}
@@ -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"`
}
@@ -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
}
@@ -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
}
+18
View File
@@ -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))
@@ -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 {
+55 -52
View File
@@ -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),
}
@@ -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.
@@ -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),
}
+28 -25
View File
@@ -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 {
@@ -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")
}
}
+9 -6
View File
@@ -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"
@@ -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"
/>
)}
<TopbarMetrics />
<TooltipProvider>
<ViewToggleGroup toggleValue={toggleValue} onValueChange={handleViewChange} size="lg" />
</TooltipProvider>
@@ -244,6 +246,7 @@ function DesktopHeader({
actions={
<>
{actionsSearch}
<TopbarMetrics />
<TooltipProvider>
<ViewToggleGroup toggleValue={toggleValue} onValueChange={handleViewChange} size="lg" />
</TooltipProvider>
@@ -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),
};
@@ -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() {
<Separator />
<SettingsSection
icon={<IconActivity className="h-5 w-5" />}
title="Resource Metrics"
description="Configure backend and execution resource sampling"
>
<SystemMetricsSettingsCard />
</SettingsSection>
<Separator />
<SettingsSection
icon={<IconKeyboard className="h-5 w-5" />}
title="Chat Input"
@@ -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<SystemMetricsGlobalSettings>(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 (
<Card>
<CardHeader>
<CardTitle className="text-base">Resource Metrics</CardTitle>
</CardHeader>
<CardContent className="space-y-5">
<p className="max-w-3xl text-sm text-muted-foreground">
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.
</p>
<MetricsDisplayToggle
checked={userSettings.systemMetricsDisplay.showInTopbar}
onCheckedChange={toggleDisplay}
/>
<MetricsSamplerControls
settings={settings}
isSaving={isSaving}
onToggleMetric={toggleMetric}
onChangeSettings={(next) => void saveGlobal(next)}
onDraftSettings={setSettings}
/>
<ExecutionMetricsToggle
checked={settings.collect_execution}
disabled={isSaving}
onCheckedChange={(checked) =>
void saveGlobal({ ...settings, collect_execution: checked })
}
/>
</CardContent>
</Card>
);
}
function MetricsDisplayToggle({
checked,
onCheckedChange,
}: {
checked: boolean;
onCheckedChange: (checked: boolean) => void;
}) {
return (
<div className="flex items-center justify-between gap-4">
<div className="space-y-1">
<Label htmlFor="show-system-metrics">Show in desktop topbars</Label>
<p className="text-xs text-muted-foreground">
Collection starts only while at least one desktop or tablet client displays metrics.
</p>
</div>
<Switch id="show-system-metrics" checked={checked} onCheckedChange={onCheckedChange} />
</div>
);
}
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 (
<div className="grid gap-4 md:grid-cols-[1fr_180px_180px]">
<MetricCheckboxes settings={settings} isSaving={isSaving} onToggleMetric={onToggleMetric} />
<div className="space-y-2">
<Label htmlFor="metrics-interval">Frequency (seconds)</Label>
<Input
id="metrics-interval"
type="number"
min={1}
max={300}
value={settings.interval_seconds}
disabled={isSaving}
onChange={(event) =>
onChangeSettings({ ...settings, interval_seconds: clampInterval(event.target.value) })
}
/>
</div>
<div className="space-y-2">
<Label htmlFor="metrics-disk-path">Disk path</Label>
<Input
id="metrics-disk-path"
value={settings.backend_disk_path}
disabled={isSaving}
onChange={(event) =>
onDraftSettings({ ...settings, backend_disk_path: event.target.value })
}
onBlur={() => onChangeSettings(settings)}
/>
</div>
</div>
);
}
function MetricCheckboxes({
settings,
isSaving,
onToggleMetric,
}: {
settings: SystemMetricsGlobalSettings;
isSaving: boolean;
onToggleMetric: (metric: SystemMetricId, checked: boolean) => void;
}) {
return (
<div className="space-y-2">
<Label>Metrics</Label>
<div className="grid gap-2 sm:grid-cols-2">
{METRIC_OPTIONS.map((metric) => (
<label key={metric.id} className="flex items-center gap-2 text-sm">
<Checkbox
checked={settings.metrics.includes(metric.id)}
disabled={isSaving}
onCheckedChange={(checked) => onToggleMetric(metric.id, checked === true)}
/>
<span>{metric.label}</span>
</label>
))}
</div>
</div>
);
}
function ExecutionMetricsToggle({
checked,
disabled,
onCheckedChange,
}: {
checked: boolean;
disabled: boolean;
onCheckedChange: (checked: boolean) => void;
}) {
return (
<div className="flex items-center justify-between gap-4">
<div className="space-y-1">
<Label htmlFor="collect-execution-metrics">Collect execution environment metrics</Label>
<p className="text-xs text-muted-foreground">
Adds agentctl values for Docker, SSH, Sprites, and remote executors.
</p>
</div>
<Switch
id="collect-execution-metrics"
checked={checked}
disabled={disabled}
onCheckedChange={onCheckedChange}
/>
</div>
);
}
function clampInterval(raw: string) {
return Math.min(300, Math.max(1, Number(raw) || 5));
}
@@ -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 (
<div>
{label}:{enabled ? "on" : "off"}
</div>
);
}
function EnableMetricsFromNestedProvider() {
const setUserSettings = useAppStore((state) => state.setUserSettings);
const userSettings = useAppStore((state) => state.userSettings);
return (
<button
type="button"
onClick={() =>
setUserSettings({
...userSettings,
systemMetricsDisplay: { showInTopbar: true },
loaded: true,
})
}
>
Enable metrics
</button>
);
}
describe("StateProvider", () => {
it("reuses the parent store for nested route providers", async () => {
render(
<StateProvider>
<ShowMetricsPreference label="root" />
<StateProvider
initialState={{
userSettings: {
...defaultSettingsState.userSettings,
systemMetricsDisplay: { showInTopbar: false },
loaded: true,
},
}}
>
<EnableMetricsFromNestedProvider />
</StateProvider>
</StateProvider>,
);
expect(screen.getByText("root:off")).toBeTruthy();
fireEvent.click(screen.getByRole("button", { name: "Enable metrics" }));
expect(await screen.findByText("root:on")).toBeTruthy();
});
});
+9 -2
View File
@@ -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;
@@ -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 (
<div className="hidden md:flex h-7 items-center gap-1 rounded border border-border px-2 text-xs text-muted-foreground">
<IconActivity className="h-3.5 w-3.5" />
<span>Metrics</span>
</div>
);
}
return (
<div className="hidden md:flex max-w-[42vw] items-center gap-1 overflow-hidden">
{sources.map((source) => (
<SourceMetrics
key={source.id}
source={source}
updatedAt={snapshot?.timestamp}
showSource={sources.length > 1}
/>
))}
</div>
);
}
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 (
<div className="flex h-7 max-w-[220px] items-center gap-1 overflow-hidden rounded border border-border px-1.5 text-xs">
{showSource ? <SourceBadge source={source} updatedAt={updatedAt} /> : null}
{metrics.length > 0 ? (
metrics.map((metric) => (
<MetricChip key={metric.id} metric={metric} source={source} updatedAt={updatedAt} />
))
) : (
<span className="px-1 text-muted-foreground">-</span>
)}
</div>
);
}
function SourceBadge({ source, updatedAt }: { source: SystemMetricsSource; updatedAt?: string }) {
const isHost = source.kind === "backend";
const label = isHost ? "Host" : "Executor";
return (
<Tooltip>
<TooltipTrigger asChild>
<span
className="flex h-5 w-5 shrink-0 items-center justify-center rounded text-muted-foreground"
aria-label={`${label} metrics`}
>
{isHost ? (
<IconServer className="h-3.5 w-3.5" />
) : (
<IconDeviceDesktopAnalytics className="h-3.5 w-3.5" />
)}
</span>
</TooltipTrigger>
<TooltipContent>
<div className="space-y-1">
<div className="font-medium">{label}</div>
<div className="text-xs text-muted-foreground">{source.label}</div>
<div className="text-xs text-muted-foreground">{lastUpdatedText(updatedAt)}</div>
</div>
</TooltipContent>
</Tooltip>
);
}
function MetricChip({
metric,
source,
updatedAt,
}: {
metric: SystemMetricSample;
source: SystemMetricsSource;
updatedAt?: string;
}) {
return (
<Tooltip>
<TooltipTrigger asChild>
<span
className={`flex h-5 shrink-0 items-center gap-0.5 rounded px-1 tabular-nums ${metricColor(metric)}`}
aria-label={`${metricLabel(metric.id)} ${formatMetric(metric)}`}
>
{metricIcon(metric.id)}
<span>{formatMetric(metric)}</span>
</span>
</TooltipTrigger>
<TooltipContent>
<div className="space-y-1">
<div className="font-medium">{metricLabel(metric.id)}</div>
<div className="text-xs text-muted-foreground">
{source.kind === "backend" ? "Host" : "Executor"}: {source.label}
</div>
<div className="text-xs tabular-nums">{formatMetric(metric)}</div>
{metric.error ? (
<div className="text-xs text-muted-foreground">{metric.error}</div>
) : null}
<div className="text-xs text-muted-foreground">{lastUpdatedText(updatedAt)}</div>
</div>
</TooltipContent>
</Tooltip>
);
}
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 <IconCpu className="h-3.5 w-3.5" />;
case "memory_percent":
return <IconDatabase className="h-3.5 w-3.5" />;
case "disk_percent":
return <IconDisc className="h-3.5 w-3.5" />;
case "cpu_temp":
return <IconFlame className="h-3.5 w-3.5" />;
case "io_load":
return <IconGauge className="h-3.5 w-3.5" />;
default:
return <IconActivity className="h-3.5 w-3.5" />;
}
}
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 })}`;
}
@@ -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(<TaskTopBar taskId="task-1" remoteExecutorType="worktree" />);
renderTopBar(<TaskTopBar taskId="task-1" remoteExecutorType="worktree" />);
expect(screen.queryByTestId("executor-settings-button")).toBeNull();
});
it("shows the executor environment button for Docker executors", () => {
render(<TaskTopBar taskId="task-1" remoteExecutorType="local_docker" />);
renderTopBar(<TaskTopBar taskId="task-1" remoteExecutorType="local_docker" />);
expect(screen.getByTestId("executor-settings-button")).toBeTruthy();
});
});
function renderTopBar(ui: React.ReactNode) {
return render(<StateProvider>{ui}</StateProvider>);
}
@@ -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 (
<div className="flex items-center justify-self-end gap-2 [&_button]:whitespace-nowrap">
<TopbarMetrics activeSessionId={activeSessionId} />
{officeTaskHref && (
<TopbarCluster label="Open in office view" className="[&_a]:h-7 [&_a]:text-xs">
<Button asChild size="sm" variant="outline" className="h-7 cursor-pointer px-2">
@@ -0,0 +1,13 @@
"use client";
import { useEffect } from "react";
import { getWebSocketClient } from "@/lib/ws/connection";
export function useSystemMetricsSubscription(enabled: boolean) {
useEffect(() => {
if (!enabled) return;
const client = getWebSocketClient();
if (!client) return;
return client.subscribeSystemMetrics();
}, [enabled]);
}
@@ -60,6 +60,7 @@ function carryForwardCoreSettings(current: DisplaySettings) {
defaultUtilityAgentId: current.defaultUtilityAgentId ?? null,
keyboardShortcuts: current.keyboardShortcuts ?? {},
changesPanelLayout: current.changesPanelLayout ?? "tree",
systemMetricsDisplay: current.systemMetricsDisplay ?? { showInTopbar: false },
voiceMode: current.voiceMode ?? { ...DEFAULT_VOICE_MODE_STATE },
};
}
+22
View File
@@ -22,6 +22,10 @@ import type {
DynamicModelsResponse,
} from "@/lib/types/http";
import type { VoiceModeSettings } from "@/lib/types/http-voice";
import type {
SystemMetricsGlobalSettings,
SystemMetricsSettingsResponse,
} from "@/lib/types/system";
// User settings
export async function fetchUserSettings(options?: ApiRequestOptions) {
@@ -53,6 +57,7 @@ export async function updateUserSettings(
terminal_font_family?: string;
terminal_font_size?: number;
changes_panel_layout?: "flat" | "tree";
system_metrics_display?: { show_in_topbar?: boolean };
voice_mode?: VoiceModeSettings;
},
options?: ApiRequestOptions,
@@ -63,6 +68,23 @@ export async function updateUserSettings(
});
}
export async function fetchSystemMetricsSettings(options?: ApiRequestOptions) {
return fetchJsonWithRetry<SystemMetricsSettingsResponse>(
"/api/v1/system/metrics/settings",
options,
);
}
export async function updateSystemMetricsSettings(
payload: SystemMetricsGlobalSettings,
options?: ApiRequestOptions,
) {
return fetchJson<SystemMetricsSettingsResponse>("/api/v1/system/metrics/settings", {
...options,
init: { ...(options?.init ?? {}), method: "PATCH", body: JSON.stringify(payload) },
});
}
// Executors
export async function listExecutors(options?: ApiRequestOptions): Promise<ListExecutorsResponse> {
return fetchJson<ListExecutorsResponse>("/api/v1/executors", options);
+12
View File
@@ -14,6 +14,10 @@ export function parseChangesPanelLayout(value: string | undefined): "flat" | "tr
return value === "flat" ? "flat" : "tree";
}
export function parseSystemMetricsDisplay(value: UserSettingsData["system_metrics_display"]) {
return { showInTopbar: value?.show_in_topbar ?? false };
}
/**
* Maps the backend's snake_case VoiceMode payload into the camelCase shape
* the store and UI use. Missing or partial payloads fall back to the defaults
@@ -46,6 +50,12 @@ function buildVoiceModeFields(s: UserSettingsData) {
return { voiceMode: parseVoiceMode(s.voice_mode) };
}
function buildSystemMetricsDisplayFields(s: UserSettingsData | undefined) {
return {
systemMetricsDisplay: parseSystemMetricsDisplay(s?.system_metrics_display),
};
}
function buildIdentityFields(s: UserSettingsData) {
return {
workspaceId: s.workspace_id || null,
@@ -76,6 +86,7 @@ export function buildCoreFields(s: UserSettingsData) {
savedLayouts: s.saved_layouts ?? [],
sidebarViews: (s.sidebar_views ?? []).map(fromApiSidebarView) as SidebarView[],
...buildTerminalFields(s),
...buildSystemMetricsDisplayFields(s),
...buildVoiceModeFields(s),
};
}
@@ -117,6 +128,7 @@ export function mapUserSettingsResponse(response: UserSettingsResponse | null) {
terminalFontFamily: null,
terminalFontSize: null,
changesPanelLayout: "tree" as const,
...buildSystemMetricsDisplayFields(undefined),
voiceMode: { ...DEFAULT_VOICE_MODE_STATE },
...buildLspFields(undefined),
loaded: false,
@@ -44,6 +44,7 @@ export const defaultSettingsState: SettingsSliceState = {
terminalFontFamily: null,
terminalFontSize: null,
changesPanelLayout: "tree",
systemMetricsDisplay: { showInTopbar: false },
voiceMode: { ...DEFAULT_VOICE_MODE_STATE },
loaded: false,
},
@@ -161,6 +161,7 @@ export type UserSettingsState = {
terminalFontFamily: string | null;
terminalFontSize: number | null;
changesPanelLayout: "flat" | "tree";
systemMetricsDisplay: { showInTopbar: boolean };
voiceMode: VoiceModeState;
loaded: boolean;
};
@@ -10,6 +10,7 @@ export const defaultSystemState: SystemSliceState = {
logs: { files: [], tail: [], tailLoaded: false },
updates: null,
jobs: {},
metrics: null,
},
};
@@ -61,4 +62,8 @@ export const createSystemSlice: StateCreator<
set((draft) => {
delete draft.system.jobs[jobId];
}),
setSystemMetricsSnapshot: (snapshot) =>
set((draft) => {
draft.system.metrics = snapshot;
}),
});
@@ -6,6 +6,7 @@ import type {
LogFileInfo,
UpdatesResponse,
SystemJob,
SystemMetricsSnapshot,
} from "@/lib/types/system";
export type SystemBackupsState = {
@@ -30,6 +31,7 @@ export type SystemSliceState = {
logs: SystemLogsState;
updates: UpdatesResponse | null;
jobs: SystemJobsMap;
metrics: SystemMetricsSnapshot | null;
};
};
@@ -43,6 +45,7 @@ export type SystemSliceActions = {
setSystemUpdates: (updates: UpdatesResponse) => void;
upsertSystemJob: (job: SystemJob) => void;
clearSystemJob: (jobId: string) => void;
setSystemMetricsSnapshot: (snapshot: SystemMetricsSnapshot) => void;
};
export type SystemSlice = SystemSliceState & SystemSliceActions;
+6
View File
@@ -19,6 +19,7 @@ export type BackendMessageType =
| "session.git.event"
| "system.error"
| "system.job.update"
| "system.metrics.updated"
| "workspace.created"
| "workspace.updated"
| "workspace.deleted"
@@ -394,6 +395,7 @@ export type UserSettingsUpdatedPayload = {
keyboard_shortcuts?: Record<string, { key: string; modifiers?: Record<string, boolean> }>;
terminal_link_behavior?: string;
changes_panel_layout?: "flat" | "tree";
system_metrics_display?: { show_in_topbar?: boolean };
voice_mode?: import("@/lib/types/http-voice").VoiceModeSettings;
updated_at?: string;
};
@@ -523,6 +525,10 @@ export type BackendMessageMap = OfficeBackendMessageMap & {
"session.git.event": BackendMessage<"session.git.event", GitEventPayload>;
"system.error": BackendMessage<"system.error", SystemErrorPayload>;
"system.job.update": BackendMessage<"system.job.update", import("./system").SystemJob>;
"system.metrics.updated": BackendMessage<
"system.metrics.updated",
import("./system").SystemMetricsSnapshot
>;
"workspace.created": BackendMessage<"workspace.created", WorkspacePayload>;
"workspace.updated": BackendMessage<"workspace.updated", WorkspacePayload>;
"workspace.deleted": BackendMessage<"workspace.deleted", WorkspacePayload>;
+2 -3
View File
@@ -414,8 +414,6 @@ export type SidebarViewApi = {
collapsed_groups: string[];
};
import type { VoiceModeSettings } from "./http-voice";
export type UserSettings = {
user_id: string;
workspace_id: WorkspaceId;
@@ -442,7 +440,8 @@ export type UserSettings = {
terminal_font_family?: string;
terminal_font_size?: number;
changes_panel_layout?: "flat" | "tree";
voice_mode?: VoiceModeSettings;
system_metrics_display?: { show_in_topbar?: boolean };
voice_mode?: import("./http-voice").VoiceModeSettings;
updated_at: string;
};
+43
View File
@@ -108,6 +108,49 @@ export interface SystemJob {
ended_at?: string;
}
export type SystemMetricId =
| "cpu_percent"
| "memory_percent"
| "disk_percent"
| "cpu_temp"
| "io_load";
export interface SystemMetricsGlobalSettings {
metrics: SystemMetricId[];
interval_seconds: number;
backend_disk_path: string;
collect_execution: boolean;
}
export interface SystemMetricsSettingsResponse {
settings: SystemMetricsGlobalSettings;
}
export interface SystemMetricSample {
id: SystemMetricId | string;
label: string;
unit?: string;
value?: number;
available: boolean;
error?: string;
}
export interface SystemMetricsSource {
id: string;
label: string;
kind: "backend" | "execution" | string;
executor_type?: string;
session_id?: string;
task_id?: string;
metrics: SystemMetricSample[];
}
export interface SystemMetricsSnapshot {
timestamp: string;
interval_seconds: number;
sources: SystemMetricsSource[];
}
export interface JobAcceptResponse {
job_id: string;
}
+34
View File
@@ -64,6 +64,7 @@ export class WebSocketClient {
private sessionFocusCounts = new Map<string, number>();
private userSubscriptionCount = 0;
private runSubscriptions = new Map<string, number>();
private systemMetricsSubscriptionCount = 0;
constructor(
private url: string,
@@ -332,6 +333,31 @@ export class WebSocketClient {
return () => this.unsubscribeRun(runId);
}
subscribeSystemMetrics() {
this.systemMetricsSubscriptionCount += 1;
if (this.status === "connected" && this.systemMetricsSubscriptionCount === 1) {
this.send({
id: generateUUID(),
type: "request",
action: "system.metrics.subscribe",
payload: {},
});
}
return () => this.unsubscribeSystemMetrics();
}
unsubscribeSystemMetrics() {
this.systemMetricsSubscriptionCount = Math.max(0, this.systemMetricsSubscriptionCount - 1);
if (this.status === "connected" && this.systemMetricsSubscriptionCount === 0) {
this.send({
id: generateUUID(),
type: "request",
action: "system.metrics.unsubscribe",
payload: {},
});
}
}
unsubscribeRun(runId: string) {
const currentCount = this.runSubscriptions.get(runId);
if (!currentCount) return;
@@ -534,6 +560,14 @@ export class WebSocketClient {
payload: {},
});
}
if (this.systemMetricsSubscriptionCount > 0) {
this.send({
id: generateUUID(),
type: "request",
action: "system.metrics.subscribe",
payload: {},
});
}
}
private flushQueue() {
@@ -13,5 +13,8 @@ export function registerSystemEventsHandlers(store: StoreApi<AppState>): WsHandl
// jobs map mirrors the latest queued/running/succeeded/failed state.
store.getState().upsertSystemJob(message.payload);
},
"system.metrics.updated": (message) => {
store.getState().setSystemMetricsSnapshot(message.payload);
},
};
}
+6 -1
View File
@@ -1,7 +1,11 @@
import type { StoreApi } from "zustand";
import type { AppState } from "@/lib/state/store";
import type { WsHandlers } from "@/lib/ws/handlers/types";
import { parseChangesPanelLayout, parseVoiceMode } from "@/lib/ssr/user-settings";
import {
parseChangesPanelLayout,
parseSystemMetricsDisplay,
parseVoiceMode,
} from "@/lib/ssr/user-settings";
export function registerUsersHandlers(store: StoreApi<AppState>): WsHandlers {
return {
@@ -32,6 +36,7 @@ export function registerUsersHandlers(store: StoreApi<AppState>): WsHandlers {
? "browser_panel"
: "new_tab",
changesPanelLayout: parseChangesPanelLayout(message.payload.changes_panel_layout),
systemMetricsDisplay: parseSystemMetricsDisplay(message.payload.system_metrics_display),
voiceMode: parseVoiceMode(message.payload.voice_mode),
loaded: true,
},
@@ -0,0 +1,31 @@
# 0017: Resource Metrics Sampling
**Status:** accepted
**Date:** 2026-06-14
**Area:** backend, frontend, protocol
## Context
Kandev needs lightweight CPU, memory, disk, load, temperature, and IO pressure visibility for users running the app on a remote server or using isolated execution environments. The values should appear in desktop task and kanban topbars only when a user opts in. Kandev also runs agentctl in multiple runtime shapes: some share the backend host, while Docker, remote, and cloud executors have their own resource boundary.
## Decision
Use global system metrics settings for collection policy and per-user settings only for display preference. Global Kandev settings are persisted in the shared `settings` key/value table owned by `internal/system/settings`; resource metrics store their sampler policy under the `system_metrics` key instead of owning a feature-specific settings table. The backend sampler starts only while at least one desktop/tablet WebSocket connection has explicitly subscribed to metrics display, and stops when the last interested connection unsubscribes or disconnects.
The backend process samples the backend host. Agentctl exposes an execution-environment metrics endpoint for runtimes with a distinct boundary: local Docker, remote Docker, SSH/remote VPS, and cloud/Sprites. Local process and worktree agentctl instances are not sampled separately in v1 because they duplicate the backend host. Containerized agentctl collectors prefer cgroup CPU and memory accounting when available, falling back to procfs only when no meaningful cgroup limit exists.
Metrics updates are delivered over a dedicated WebSocket stream to subscribed connections only, not via global broadcast. Mobile clients do not subscribe and do not render the metrics strip.
## Consequences
The backend avoids background procfs/cgroup/agentctl polling when no visible UI needs metrics. Multi-user behavior stays clear: global settings control what is sampled and how often, while each user controls whether they see it. Docker and remote users get the execution-environment values that matter instead of duplicated backend-host values.
This adds a small connection-interest registry, a reusable install-wide settings table, and a new agentctl API surface. It also means homepage/list views intentionally show backend metrics only in v1; execution metrics are scoped to task detail where the active session is unambiguous.
## Alternatives Considered
- **Per-user sampler settings:** Rejected because metric collection is process-wide and would cause conflicting intervals and duplicated polling in multi-user sessions.
- **Always-on backend sampler:** Rejected because the feature is optional and should have zero steady-state cost when nobody displays it.
- **Sample every agentctl instance:** Rejected because local process/worktree instances share the backend host and would duplicate values. V1 samples only execution boundaries that add distinct resource context.
- **Global broadcast:** Rejected because this is a high-frequency stream; only interested connections should receive it.
- **Metrics-owned `system_settings` table:** Rejected because future install-wide Kandev settings should share one clearly owned backend store instead of each feature inventing its own generic table.
+1
View File
@@ -22,3 +22,4 @@ Read individual ADRs for full context. Create new ones via `/record decision` or
| 0014 | [Per-CLI MCP server injection for passthrough mode](0014-passthrough-mcp-injection-strategies.md) | accepted | backend | 2026-05-29 |
| 0015 | [Explicit completion signal for auto-advance](0015-explicit-completion-signal-for-auto-advance.md) | proposed | backend, frontend | 2026-06-04 |
| 0016 | [Read-only absolute file paths](0016-observed-external-file-reads.md) | accepted | backend | 2026-06-14 |
| 0017 | [Resource metrics sampling](0017-resource-metrics-sampling.md) | accepted | backend, frontend, protocol | 2026-06-14 |