From 08a2d95c0c05c5135ceac811eaad2b240fd64f52 Mon Sep 17 00:00:00 2001 From: viettranx Date: Wed, 18 Mar 2026 13:11:44 +0700 Subject: [PATCH] =?UTF-8?q?feat:=20agent=20heartbeat=20system=20=E2=80=94?= =?UTF-8?q?=20periodic=20proactive=20check-ins=20(#245)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 (Core): - Migration 000022: agent_heartbeats, heartbeat_run_logs, agent_config_permissions tables - HeartbeatStore + ConfigPermissionStore interfaces with PG implementations - HeartbeatTicker: background poll → active hours filter → queue-aware skip → run → smart suppression → deliver/log - Heartbeat tool: status/get/set/toggle/set_checklist/get_checklist/test/logs actions - Permission check with wildcard scope matching + TTL cache (60s) - RPC methods: heartbeat.get/set/toggle/test/logs/checklist.get/checklist.set - HEARTBEAT.md routed via context file interceptor (read/write for both open + predefined agents) - Session keys: agent:{id}:heartbeat or agent:{id}:heartbeat:{ts} (isolated) - PromptMinimal for heartbeat sessions (like cron/subagent) - Event broadcasting + cache invalidation via bus (heartbeat + config_perms) - Gateway wiring: ticker init, event wiring, graceful shutdown Phase 2 (Integration): - wakeMode: CronPayload.WakeHeartbeat triggers heartbeat after cron job completes - Queue-aware: Scheduler.HasActiveSessionsForAgent() skips busy agents - Stagger: deterministic FNV offset spreads heartbeats across interval - lightContext: RunRequest.LightContext skips context files, only injects checklist - System prompt distinguishes cron (user-scheduled tasks) vs heartbeat (autonomous monitoring) --- cmd/gateway.go | 35 +- cmd/gateway_heartbeat.go | 15 + cmd/gateway_managed.go | 28 ++ cmd/gateway_methods.go | 10 +- internal/agent/loop_history.go | 10 +- internal/agent/systemprompt.go | 3 +- internal/bootstrap/files.go | 9 +- internal/bus/types.go | 4 + internal/gateway/methods/agents_files.go | 1 + internal/gateway/methods/heartbeat.go | 400 +++++++++++++++++ internal/heartbeat/ticker.go | 454 ++++++++++++++++++++ internal/sessions/key.go | 18 + internal/store/config_permission_store.go | 34 ++ internal/store/heartbeat_store.go | 109 +++++ internal/store/pg/config_permissions.go | 191 ++++++++ internal/store/pg/factory.go | 4 +- internal/store/pg/heartbeat.go | 271 ++++++++++++ internal/store/stores.go | 4 +- internal/tools/context_file_interceptor.go | 7 +- internal/tools/heartbeat.go | 412 ++++++++++++++++++ internal/upgrade/version.go | 2 +- migrations/000022_agent_heartbeats.down.sql | 3 + migrations/000022_agent_heartbeats.up.sql | 68 +++ pkg/protocol/events.go | 1 + pkg/protocol/methods.go | 11 + 25 files changed, 2088 insertions(+), 16 deletions(-) create mode 100644 cmd/gateway_heartbeat.go create mode 100644 internal/gateway/methods/heartbeat.go create mode 100644 internal/heartbeat/ticker.go create mode 100644 internal/store/config_permission_store.go create mode 100644 internal/store/heartbeat_store.go create mode 100644 internal/store/pg/config_permissions.go create mode 100644 internal/store/pg/heartbeat.go create mode 100644 internal/tools/heartbeat.go create mode 100644 migrations/000022_agent_heartbeats.down.sql create mode 100644 migrations/000022_agent_heartbeats.up.sql diff --git a/cmd/gateway.go b/cmd/gateway.go index c49ead7e..ac011d42 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -24,6 +24,7 @@ import ( zalopersonal "github.com/nextlevelbuilder/goclaw/internal/channels/zalo/personal" "github.com/nextlevelbuilder/goclaw/internal/config" "github.com/nextlevelbuilder/goclaw/internal/gateway" + "github.com/nextlevelbuilder/goclaw/internal/heartbeat" "github.com/nextlevelbuilder/goclaw/internal/gateway/methods" httpapi "github.com/nextlevelbuilder/goclaw/internal/http" mcpbridge "github.com/nextlevelbuilder/goclaw/internal/mcp" @@ -172,6 +173,12 @@ func runGateway() { toolsReg.Register(tools.NewCronTool(pgStores.Cron)) slog.Info("cron tool registered") + // Heartbeat tool (agent-facing) + heartbeatTool := tools.NewHeartbeatTool(pgStores.Heartbeats, pgStores.ConfigPermissions) + heartbeatTool.SetAgentStore(pgStores.Agents) + toolsReg.Register(heartbeatTool) + slog.Info("heartbeat tool registered") + // Session tools (list, status, history, send) toolsReg.Register(tools.NewSessionsListTool()) toolsReg.Register(tools.NewSessionStatusTool()) @@ -398,7 +405,7 @@ func runGateway() { // Register all RPC methods server.SetLogTee(logTee) - pairingMethods := registerAllMethods(server, agentRouter, pgStores.Sessions, pgStores.Cron, pgStores.Pairing, cfg, cfgPath, workspace, dataDir, msgBus, execApprovalMgr, pgStores.Agents, pgStores.Skills, pgStores.ConfigSecrets, pgStores.Teams, contextFileInterceptor, logTee) + pairingMethods, heartbeatMethods := registerAllMethods(server, agentRouter, pgStores.Sessions, pgStores.Cron, pgStores.Pairing, cfg, cfgPath, workspace, dataDir, msgBus, execApprovalMgr, pgStores.Agents, pgStores.Skills, pgStores.ConfigSecrets, pgStores.Teams, contextFileInterceptor, logTee, pgStores.Heartbeats) // Wire pairing event broadcasts to all WS clients. pairingMethods.SetBroadcaster(server.BroadcastEvent) @@ -700,6 +707,29 @@ func runGateway() { slog.Warn("cron service failed to start", "error", err) } + // Start heartbeat ticker (routes through scheduler's cron lane) + heartbeatTicker := heartbeat.NewTicker(heartbeat.TickerConfig{ + Store: pgStores.Heartbeats, + Agents: pgStores.Agents, + MsgBus: msgBus, + Sched: sched, + RunAgent: makeHeartbeatRunFn(sched), + }) + heartbeatTicker.SetOnEvent(func(event store.HeartbeatEvent) { + server.BroadcastEvent(*protocol.NewEvent(protocol.EventHeartbeat, event)) + }) + heartbeatTicker.Start() + + // Wire heartbeat wake function to tool + RPC + cron wakeMode + heartbeatTool.SetWakeFn(heartbeatTicker.Wake) + heartbeatMethods.SetWakeFn(heartbeatTicker.Wake) + heartbeatMethods.SetAgentStore(pgStores.Agents) + cronHeartbeatWakeFn = func(agentID string) { + if id, err := uuid.Parse(agentID); err == nil { + heartbeatTicker.Wake(id) + } + } + // Adaptive throttle: reduce per-session concurrency when nearing the summary threshold. // This prevents concurrent runs from racing with summarization. // Uses calibrated token estimation (actual prompt tokens from last LLM call) @@ -842,9 +872,10 @@ func runGateway() { // Broadcast shutdown event server.BroadcastEvent(*protocol.NewEvent(protocol.EventShutdown, nil)) - // Stop channels, cron, and task ticker + // Stop channels, cron, heartbeat, and task ticker channelMgr.StopAll(context.Background()) pgStores.Cron.Stop() + heartbeatTicker.Stop() if taskTicker != nil { taskTicker.Stop() } diff --git a/cmd/gateway_heartbeat.go b/cmd/gateway_heartbeat.go new file mode 100644 index 00000000..a0b91c5b --- /dev/null +++ b/cmd/gateway_heartbeat.go @@ -0,0 +1,15 @@ +package cmd + +import ( + "context" + + "github.com/nextlevelbuilder/goclaw/internal/agent" + "github.com/nextlevelbuilder/goclaw/internal/scheduler" +) + +// makeHeartbeatRunFn creates a function that routes a heartbeat run through the scheduler's cron lane. +func makeHeartbeatRunFn(sched *scheduler.Scheduler) func(ctx context.Context, req agent.RunRequest) <-chan scheduler.RunOutcome { + return func(ctx context.Context, req agent.RunRequest) <-chan scheduler.RunOutcome { + return sched.Schedule(ctx, scheduler.LaneCron, req) + } +} diff --git a/cmd/gateway_managed.go b/cmd/gateway_managed.go index 49b73291..9da01165 100644 --- a/cmd/gateway_managed.go +++ b/cmd/gateway_managed.go @@ -358,6 +358,34 @@ func wireExtras( }) } + // Heartbeat cache: invalidate due cache on config changes + if hi, ok := stores.Heartbeats.(store.CacheInvalidatable); ok { + msgBus.Subscribe(bus.TopicCacheHeartbeat, func(event bus.Event) { + if event.Name != protocol.EventCacheInvalidate { + return + } + payload, ok := event.Payload.(bus.CacheInvalidatePayload) + if !ok || payload.Kind != bus.CacheKindHeartbeat { + return + } + hi.InvalidateCache() + }) + } + + // Config permissions cache: invalidate on grant/revoke changes + if pi, ok := stores.ConfigPermissions.(store.CacheInvalidatable); ok { + msgBus.Subscribe(bus.TopicCacheConfigPerms, func(event bus.Event) { + if event.Name != protocol.EventCacheInvalidate { + return + } + payload, ok := event.Payload.(bus.CacheInvalidatePayload) + if !ok || payload.Kind != bus.CacheKindConfigPerms { + return + } + pi.InvalidateCache() + }) + } + // Custom tools cache: reload global tools on create/update/delete if dynamicLoader != nil { msgBus.Subscribe(bus.TopicCacheCustomTools, func(event bus.Event) { diff --git a/cmd/gateway_methods.go b/cmd/gateway_methods.go index 4cf71cee..9dab82b7 100644 --- a/cmd/gateway_methods.go +++ b/cmd/gateway_methods.go @@ -12,7 +12,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/tools" ) -func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore store.SessionStore, cronStore store.CronStore, pairingStore store.PairingStore, cfg *config.Config, cfgPath, workspace, dataDir string, msgBus *bus.MessageBus, execApprovalMgr *tools.ExecApprovalManager, agentStore store.AgentStore, skillStore store.SkillStore, configSecretsStore store.ConfigSecretsStore, teamStore store.TeamStore, contextFileInterceptor *tools.ContextFileInterceptor, logTee *gateway.LogTee) *methods.PairingMethods { +func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore store.SessionStore, cronStore store.CronStore, pairingStore store.PairingStore, cfg *config.Config, cfgPath, workspace, dataDir string, msgBus *bus.MessageBus, execApprovalMgr *tools.ExecApprovalManager, agentStore store.AgentStore, skillStore store.SkillStore, configSecretsStore store.ConfigSecretsStore, teamStore store.TeamStore, contextFileInterceptor *tools.ContextFileInterceptor, logTee *gateway.LogTee, heartbeatStore store.HeartbeatStore) (*methods.PairingMethods, *methods.HeartbeatMethods) { router := server.Router() // Phase 1: Core methods @@ -27,6 +27,10 @@ func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore // Phase 2: Cron (store created externally, shared with gateway) methods.NewCronMethods(cronStore, msgBus).Register(router) + // Phase 2: Heartbeat + heartbeatMethods := methods.NewHeartbeatMethods(heartbeatStore, msgBus) + heartbeatMethods.Register(router) + // Phase 2: Pairing (store created externally, shared with channel manager). // OnApprove callback is set later by the caller after channel manager is created. pairingMethods := methods.NewPairingMethods(pairingStore, msgBus, server.RateLimiter()) @@ -51,8 +55,8 @@ func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore slog.Info("registered all RPC methods", "phase1", []string{"chat", "agents", "sessions", "config"}, - "phase2", []string{"skills", "cron", "pairing", "usage", "exec_approval", "send"}, + "phase2", []string{"skills", "cron", "heartbeat", "pairing", "usage", "exec_approval", "send"}, ) - return pairingMethods + return pairingMethods, heartbeatMethods } diff --git a/internal/agent/loop_history.go b/internal/agent/loop_history.go index f664b385..10406c70 100644 --- a/internal/agent/loop_history.go +++ b/internal/agent/loop_history.go @@ -64,12 +64,12 @@ func (l *Loop) buildMCPToolDescs(toolNames []string) map[string]string { // buildMessages constructs the full message list for an LLM request. // Returns the messages and whether BOOTSTRAP.md was present in context files // (used by the caller for auto-cleanup without an extra DB roundtrip). -func (l *Loop) buildMessages(ctx context.Context, history []providers.Message, summary, userMessage, extraSystemPrompt, sessionKey, channel, channelType, peerKind, userID string, historyLimit int, skillFilter []string) ([]providers.Message, bool) { +func (l *Loop) buildMessages(ctx context.Context, history []providers.Message, summary, userMessage, extraSystemPrompt, sessionKey, channel, channelType, peerKind, userID string, historyLimit int, skillFilter []string, lightContext bool) ([]providers.Message, bool) { var messages []providers.Message // Build full system prompt using the new builder (matching TS buildAgentSystemPrompt) mode := PromptFull - if bootstrap.IsSubagentSession(sessionKey) || bootstrap.IsCronSession(sessionKey) { + if bootstrap.IsSubagentSession(sessionKey) || bootstrap.IsCronSession(sessionKey) || bootstrap.IsHeartbeatSession(sessionKey) { mode = PromptMinimal } @@ -96,7 +96,11 @@ func (l *Loop) buildMessages(ctx context.Context, history []providers.Message, s } // Resolve context files once — also detect BOOTSTRAP.md presence. - contextFiles := l.resolveContextFiles(ctx, userID) + // lightContext: skip loading context files, only inject ExtraSystemPrompt (heartbeat checklist). + var contextFiles []bootstrap.ContextFile + if !lightContext { + contextFiles = l.resolveContextFiles(ctx, userID) + } hadBootstrap := false for _, cf := range contextFiles { if cf.Path == bootstrap.BootstrapFile { diff --git a/internal/agent/systemprompt.go b/internal/agent/systemprompt.go index bb876827..b12879f6 100644 --- a/internal/agent/systemprompt.go +++ b/internal/agent/systemprompt.go @@ -79,7 +79,8 @@ var coreToolSummaries = map[string]string{ "web_search": "Search the web", "web_fetch": "Fetch and extract content from a URL", "datetime": "Get current date/time with timezone support — use before creating cron jobs or time-sensitive operations", - "cron": "Manage scheduled jobs and reminders", + "cron": "Manage scheduled jobs and reminders — use for user-requested tasks at specific times or intervals (e.g. 'remind me at 9am', 'check weather every morning')", + "heartbeat": "Manage agent heartbeat — periodic background monitoring with HEARTBEAT.md checklist. Use for autonomous proactive check-ins (e.g. 'monitor server status every 30 min'). Unlike cron, heartbeat auto-suppresses 'all OK' responses via HEARTBEAT_OK", "skill_search": "Search available skills by keyword (weather, translate, github, etc.)", "skill_manage": "Create, patch, or delete skills from conversation experience", "publish_skill": "Register a skill directory in the system database, making it discoverable", diff --git a/internal/bootstrap/files.go b/internal/bootstrap/files.go index 586dfb42..617c6e8f 100644 --- a/internal/bootstrap/files.go +++ b/internal/bootstrap/files.go @@ -30,6 +30,7 @@ const ( DelegationFile = "DELEGATION.md" TeamFile = "TEAM.md" AvailabilityFile = "AVAILABILITY.md" + HeartbeatFile = "HEARTBEAT.md" MemoryFile = "MEMORY.md" MemoryAltFile = "memory.md" MemoryJSONFile = "MEMORY.json" @@ -93,7 +94,7 @@ func LoadWorkspaceFiles(workspaceDir string) []File { // Normal sessions get all files. Subagent and cron sessions get only // AGENTS.md and TOOLS.md (minimal mode), matching TS filterBootstrapFilesForSession(). func FilterForSession(files []File, sessionKey string) []File { - if !IsSubagentSession(sessionKey) && !IsCronSession(sessionKey) { + if !IsSubagentSession(sessionKey) && !IsCronSession(sessionKey) && !IsHeartbeatSession(sessionKey) { return files } @@ -122,6 +123,12 @@ func IsCronSession(sessionKey string) bool { return strings.HasPrefix(strings.ToLower(rest), "cron:") } +// IsHeartbeatSession checks if a session key indicates a heartbeat session. +func IsHeartbeatSession(sessionKey string) bool { + rest := sessionRest(sessionKey) + return strings.HasPrefix(rest, "heartbeat") +} + // IsTeamSession checks if a session key indicates a team-dispatched task session. // Session key format: agent:{agentId}:team:{teamID}:{chatID} // Team sessions have "team:" in the rest part. diff --git a/internal/bus/types.go b/internal/bus/types.go index 5ff9f51c..20ad0dda 100644 --- a/internal/bus/types.go +++ b/internal/bus/types.go @@ -66,6 +66,8 @@ const ( CacheKindMCP = "mcp" CacheKindProvider = "provider" CacheKindAPIKeys = "api_keys" + CacheKindHeartbeat = "heartbeat" + CacheKindConfigPerms = "config_perms" ) // Topic constants for msgBus.Subscribe() / Broadcast(). @@ -83,6 +85,8 @@ const ( TopicCacheSkillGrants = "cache:skill_grants" TopicCacheMCP = "cache:mcp" TopicCacheProvider = "cache:provider" + TopicCacheHeartbeat = "cache:heartbeat" + TopicCacheConfigPerms = "cache:config_perms" TopicAudit = "audit" TopicTeamTaskAudit = "team-task-audit" TopicChannelStreaming = "channel-streaming" diff --git a/internal/gateway/methods/agents_files.go b/internal/gateway/methods/agents_files.go index a5d771f9..c6058209 100644 --- a/internal/gateway/methods/agents_files.go +++ b/internal/gateway/methods/agents_files.go @@ -20,6 +20,7 @@ import ( var allowedAgentFiles = []string{ bootstrap.AgentsFile, bootstrap.SoulFile, bootstrap.IdentityFile, bootstrap.UserFile, bootstrap.UserPredefinedFile, bootstrap.BootstrapFile, bootstrap.MemoryJSONFile, + bootstrap.HeartbeatFile, } // --- agents.files.list --- diff --git a/internal/gateway/methods/heartbeat.go b/internal/gateway/methods/heartbeat.go new file mode 100644 index 00000000..aaf387bc --- /dev/null +++ b/internal/gateway/methods/heartbeat.go @@ -0,0 +1,400 @@ +package methods + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "log/slog" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/gateway" + "github.com/nextlevelbuilder/goclaw/internal/i18n" + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +// HeartbeatMethods handles heartbeat.get/set/toggle/test/logs/checklist RPC methods. +type HeartbeatMethods struct { + hbStore store.HeartbeatStore + agentStore store.AgentStore + eventBus bus.EventPublisher + wakeFn func(uuid.UUID) // triggers immediate heartbeat run +} + +func NewHeartbeatMethods(hb store.HeartbeatStore, eventBus bus.EventPublisher) *HeartbeatMethods { + return &HeartbeatMethods{hbStore: hb, eventBus: eventBus} +} + +// SetAgentStore sets the agent store for HEARTBEAT.md read/write via RPC. +func (m *HeartbeatMethods) SetAgentStore(as store.AgentStore) { + m.agentStore = as +} + +// SetWakeFn sets the function called when "heartbeat.test" triggers an immediate run. +func (m *HeartbeatMethods) SetWakeFn(fn func(uuid.UUID)) { + m.wakeFn = fn +} + +func (m *HeartbeatMethods) Register(router *gateway.MethodRouter) { + router.Register(protocol.MethodHeartbeatGet, m.handleGet) + router.Register(protocol.MethodHeartbeatSet, m.handleSet) + router.Register(protocol.MethodHeartbeatToggle, m.handleToggle) + router.Register(protocol.MethodHeartbeatTest, m.handleTest) + router.Register(protocol.MethodHeartbeatLogs, m.handleLogs) + router.Register(protocol.MethodHeartbeatChecklistGet, m.handleChecklistGet) + router.Register(protocol.MethodHeartbeatChecklistSet, m.handleChecklistSet) +} + +func (m *HeartbeatMethods) handleGet(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { + locale := store.LocaleFromContext(ctx) + var params struct { + AgentID string `json:"agentId"` + } + if req.Params != nil { + json.Unmarshal(req.Params, ¶ms) + } + if params.AgentID == "" { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "agentId"))) + return + } + + agentUUID, err := uuid.Parse(params.AgentID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "invalid agentId")) + return + } + + hb, err := m.hbStore.Get(ctx, agentUUID) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"heartbeat": nil})) + return + } + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"heartbeat": hb})) +} + +func (m *HeartbeatMethods) handleSet(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { + locale := store.LocaleFromContext(ctx) + var params struct { + AgentID string `json:"agentId"` + Enabled *bool `json:"enabled"` + IntervalSec *int `json:"intervalSec"` + Prompt *string `json:"prompt"` + Model *string `json:"model"` + IsolatedSession *bool `json:"isolatedSession"` + LightContext *bool `json:"lightContext"` + AckMaxChars *int `json:"ackMaxChars"` + MaxRetries *int `json:"maxRetries"` + ActiveHoursStart *string `json:"activeHoursStart"` + ActiveHoursEnd *string `json:"activeHoursEnd"` + Timezone *string `json:"timezone"` + Channel *string `json:"channel"` + ChatID *string `json:"chatId"` + } + if req.Params != nil { + json.Unmarshal(req.Params, ¶ms) + } + if params.AgentID == "" { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "agentId"))) + return + } + + agentUUID, err := uuid.Parse(params.AgentID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "invalid agentId")) + return + } + + // Load existing or create new. + hb, err := m.hbStore.Get(ctx, agentUUID) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + if hb == nil { + hb = &store.AgentHeartbeat{ + AgentID: agentUUID, + IntervalSec: 1800, + IsolatedSession: true, + AckMaxChars: 300, + MaxRetries: 2, + } + } + + if params.Enabled != nil { + hb.Enabled = *params.Enabled + } + if params.IntervalSec != nil { + if *params.IntervalSec < 300 { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "minimum interval is 300 seconds")) + return + } + hb.IntervalSec = *params.IntervalSec + } + if params.Prompt != nil { + hb.Prompt = params.Prompt + } + if params.Model != nil { + hb.Model = params.Model + } + if params.IsolatedSession != nil { + hb.IsolatedSession = *params.IsolatedSession + } + if params.LightContext != nil { + hb.LightContext = *params.LightContext + } + if params.AckMaxChars != nil { + if *params.AckMaxChars < 0 { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "ackMaxChars must be >= 0")) + return + } + hb.AckMaxChars = *params.AckMaxChars + } + if params.MaxRetries != nil { + if *params.MaxRetries < 0 || *params.MaxRetries > 10 { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "maxRetries must be 0-10")) + return + } + hb.MaxRetries = *params.MaxRetries + } + if params.ActiveHoursStart != nil { + hb.ActiveHoursStart = params.ActiveHoursStart + } + if params.ActiveHoursEnd != nil { + hb.ActiveHoursEnd = params.ActiveHoursEnd + } + if params.Timezone != nil { + hb.Timezone = params.Timezone + } + if params.Channel != nil { + hb.Channel = params.Channel + } + if params.ChatID != nil { + hb.ChatID = params.ChatID + } + + if hb.Enabled && hb.NextRunAt == nil { + nextRun := time.Now().Add(time.Duration(hb.IntervalSec)*time.Second + store.StaggerOffset(hb.AgentID, hb.IntervalSec)) + hb.NextRunAt = &nextRun + } + + if err := m.hbStore.Upsert(ctx, hb); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"heartbeat": hb})) + m.emitCacheInvalidate(hb.AgentID.String()) + emitAudit(m.eventBus, client, "heartbeat.set", "heartbeat", hb.AgentID.String()) +} + +func (m *HeartbeatMethods) handleToggle(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { + locale := store.LocaleFromContext(ctx) + var params struct { + AgentID string `json:"agentId"` + Enabled bool `json:"enabled"` + } + if req.Params != nil { + json.Unmarshal(req.Params, ¶ms) + } + if params.AgentID == "" { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "agentId"))) + return + } + + agentUUID, err := uuid.Parse(params.AgentID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "invalid agentId")) + return + } + + hb, err := m.hbStore.Get(ctx, agentUUID) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrNotFound, "heartbeat not configured")) + return + } + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + + hb.Enabled = params.Enabled + if params.Enabled && hb.NextRunAt == nil { + nextRun := time.Now().Add(time.Duration(hb.IntervalSec) * time.Second) + hb.NextRunAt = &nextRun + } + + if err := m.hbStore.Upsert(ctx, hb); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ + "agentId": params.AgentID, + "enabled": params.Enabled, + })) + m.emitCacheInvalidate(params.AgentID) + emitAudit(m.eventBus, client, "heartbeat.toggled", "heartbeat", params.AgentID) +} + +func (m *HeartbeatMethods) handleTest(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { + locale := store.LocaleFromContext(ctx) + var params struct { + AgentID string `json:"agentId"` + } + if req.Params != nil { + json.Unmarshal(req.Params, ¶ms) + } + if params.AgentID == "" { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "agentId"))) + return + } + + agentUUID, err := uuid.Parse(params.AgentID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "invalid agentId")) + return + } + + if m.wakeFn == nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, "heartbeat ticker not available")) + return + } + + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{"ok": true})) + emitAudit(m.eventBus, client, "heartbeat.test", "heartbeat", params.AgentID) + + go func() { + m.wakeFn(agentUUID) + slog.Info("heartbeat.test triggered", "agent_id", params.AgentID) + }() +} + +func (m *HeartbeatMethods) handleLogs(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { + locale := store.LocaleFromContext(ctx) + var params struct { + AgentID string `json:"agentId"` + Limit int `json:"limit"` + Offset int `json:"offset"` + } + if req.Params != nil { + json.Unmarshal(req.Params, ¶ms) + } + if params.AgentID == "" { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "agentId"))) + return + } + + agentUUID, err := uuid.Parse(params.AgentID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "invalid agentId")) + return + } + + logs, total, err := m.hbStore.ListLogs(ctx, agentUUID, params.Limit, params.Offset) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ + "logs": logs, + "total": total, + })) +} + +func (m *HeartbeatMethods) handleChecklistGet(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { + locale := store.LocaleFromContext(ctx) + var params struct { + AgentID string `json:"agentId"` + } + if req.Params != nil { + json.Unmarshal(req.Params, ¶ms) + } + if params.AgentID == "" { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "agentId"))) + return + } + if m.agentStore == nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, "agent store not configured")) + return + } + + agentUUID, err := uuid.Parse(params.AgentID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "invalid agentId")) + return + } + + files, err := m.agentStore.GetAgentContextFiles(ctx, agentUUID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + + var content string + for _, f := range files { + if f.FileName == "HEARTBEAT.md" { + content = f.Content + break + } + } + + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ + "content": content, + })) +} + +func (m *HeartbeatMethods) handleChecklistSet(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { + locale := store.LocaleFromContext(ctx) + var params struct { + AgentID string `json:"agentId"` + Content string `json:"content"` + } + if req.Params != nil { + json.Unmarshal(req.Params, ¶ms) + } + if params.AgentID == "" { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "agentId"))) + return + } + if m.agentStore == nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, "agent store not configured")) + return + } + + agentUUID, err := uuid.Parse(params.AgentID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "invalid agentId")) + return + } + + if err := m.agentStore.SetAgentContextFile(ctx, agentUUID, "HEARTBEAT.md", params.Content); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ + "ok": true, + "length": len([]rune(params.Content)), + })) + emitAudit(m.eventBus, client, "heartbeat.checklist.set", "heartbeat", params.AgentID) +} + +func (m *HeartbeatMethods) emitCacheInvalidate(agentID string) { + m.eventBus.Broadcast(bus.Event{ + Name: protocol.EventCacheInvalidate, + Payload: bus.CacheInvalidatePayload{ + Kind: bus.CacheKindHeartbeat, + Key: agentID, + }, + }) +} diff --git a/internal/heartbeat/ticker.go b/internal/heartbeat/ticker.go new file mode 100644 index 00000000..b8cd3276 --- /dev/null +++ b/internal/heartbeat/ticker.go @@ -0,0 +1,454 @@ +// Package heartbeat implements a background ticker that runs periodic +// agent heartbeat check-ins. Each enabled agent gets a scheduled turn +// where it reads HEARTBEAT.md and reports if anything needs attention. +package heartbeat + +import ( + "context" + "fmt" + "log/slog" + "strings" + "sync" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/agent" + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/scheduler" + "github.com/nextlevelbuilder/goclaw/internal/sessions" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +const ( + minIntervalSec = 300 // 5 minutes minimum + pollInterval = 30 * time.Second + maxSummaryLen = 500 // truncate summary in logs +) + +// TickerConfig holds dependencies for the heartbeat ticker. +type TickerConfig struct { + Store store.HeartbeatStore + Agents store.AgentStore + MsgBus *bus.MessageBus + Sched *scheduler.Scheduler + RunAgent func(ctx context.Context, req agent.RunRequest) <-chan scheduler.RunOutcome +} + +// Ticker polls for due heartbeats and runs them through the agent loop. +type Ticker struct { + store store.HeartbeatStore + agents store.AgentStore + msgBus *bus.MessageBus + sched *scheduler.Scheduler + runAgent func(ctx context.Context, req agent.RunRequest) <-chan scheduler.RunOutcome + onEvent func(store.HeartbeatEvent) + + wakeCh chan uuid.UUID + stopCh chan struct{} + wg sync.WaitGroup +} + +// NewTicker creates a new heartbeat ticker. +func NewTicker(cfg TickerConfig) *Ticker { + return &Ticker{ + store: cfg.Store, + agents: cfg.Agents, + msgBus: cfg.MsgBus, + sched: cfg.Sched, + runAgent: cfg.RunAgent, + wakeCh: make(chan uuid.UUID, 16), + stopCh: make(chan struct{}), + } +} + +// Start begins the background poll loop. +func (t *Ticker) Start() { + t.wg.Add(1) + go t.loop() + slog.Info("heartbeat ticker started") +} + +// Stop signals the poll loop to exit and waits for completion. +func (t *Ticker) Stop() { + close(t.stopCh) + t.wg.Wait() + slog.Info("heartbeat ticker stopped") +} + +// SetOnEvent sets the event callback (called for lifecycle events like running/completed/error). +func (t *Ticker) SetOnEvent(fn func(store.HeartbeatEvent)) { + t.onEvent = fn +} + +func (t *Ticker) emitEvent(event store.HeartbeatEvent) { + if t.onEvent != nil { + t.onEvent(event) + } +} + +// Wake triggers an immediate heartbeat run for a specific agent (wakeMode). +func (t *Ticker) Wake(agentID uuid.UUID) { + select { + case t.wakeCh <- agentID: + default: // channel full, skip + } +} + +func (t *Ticker) loop() { + defer t.wg.Done() + ticker := time.NewTicker(pollInterval) + defer ticker.Stop() + + for { + select { + case <-t.stopCh: + return + case <-ticker.C: + t.runDueHeartbeats() + case agentID := <-t.wakeCh: + go t.runOneByAgentID(agentID) + } + } +} + +func (t *Ticker) runDueHeartbeats() { + ctx := context.Background() + now := time.Now() + due, err := t.store.ListDue(ctx, now) + if err != nil { + slog.Warn("heartbeat.list_due_failed", "error", err) + return + } + if len(due) == 0 { + return + } + + var wg sync.WaitGroup + for _, hb := range due { + wg.Add(1) + go func(hb store.AgentHeartbeat) { + defer wg.Done() + t.runOne(ctx, hb) + }(hb) + } + wg.Wait() +} + +func (t *Ticker) runOneByAgentID(agentID uuid.UUID) { + ctx := context.Background() + hb, err := t.store.Get(ctx, agentID) + if err != nil { + slog.Warn("heartbeat.wake_get_failed", "agent_id", agentID, "error", err) + return + } + t.runOne(ctx, *hb) +} + +func (t *Ticker) runOne(ctx context.Context, hb store.AgentHeartbeat) { + start := time.Now() + agentIDStr := hb.AgentID.String() + + // Resolve agent key for events. + agentKey := agentIDStr + if ag, err := t.agents.GetByID(ctx, hb.AgentID); err == nil { + agentKey = ag.AgentKey + } + + // [1] Active hours filter. + if !isWithinActiveHours(hb) { + t.logSkipped(ctx, hb, "active_hours", agentKey) + t.advanceNextRun(ctx, hb) + return + } + + // [2] Queue-aware: skip if agent is busy (active runs in scheduler). + if t.sched != nil && t.sched.HasActiveSessionsForAgent(agentIDStr) { + t.logSkipped(ctx, hb, "queue_busy", agentKey) + // Don't advance next_run_at — retry on next poll. + return + } + + // [3] Read HEARTBEAT.md from agent context files. + checklistContent := t.readChecklist(ctx, hb.AgentID) + if checklistContent == "" { + t.logSkipped(ctx, hb, "empty_checklist", agentKey) + t.advanceNextRun(ctx, hb) + return + } + + // [4] Emit running event. + t.emitEvent(store.HeartbeatEvent{ + Action: "running", AgentID: agentIDStr, AgentKey: agentKey, + }) + + // [4] Build prompt. + prompt := "Read HEARTBEAT.md if it exists. Follow it strictly.\nIf nothing needs attention, reply HEARTBEAT_OK." + if hb.Prompt != nil && *hb.Prompt != "" { + prompt = *hb.Prompt + } + + extraSystem := fmt.Sprintf( + "[Heartbeat Check-in]\nThis is a periodic heartbeat run for agent %s.\n"+ + "Your HEARTBEAT.md checklist:\n---\n%s\n---\n"+ + "Follow the checklist. If everything is OK, reply with HEARTBEAT_OK.\n"+ + "If something needs attention, describe it clearly — your response will be delivered to the configured channel.", + agentKey, checklistContent, + ) + + sessionKey := sessions.BuildHeartbeatSessionKey(agentIDStr, hb.IsolatedSession) + + channel := "heartbeat" + if hb.Channel != nil && *hb.Channel != "" { + channel = *hb.Channel + } + chatID := "" + if hb.ChatID != nil { + chatID = *hb.ChatID + } + + // [5] Run through agent loop via scheduler. + var lastErr error + var result *agent.RunResult + maxAttempts := hb.MaxRetries + 1 + + for attempt := range maxAttempts { + outCh := t.runAgent(ctx, agent.RunRequest{ + SessionKey: sessionKey, + Message: prompt, + Channel: channel, + ChatID: chatID, + RunID: fmt.Sprintf("heartbeat:%s", agentIDStr), + Stream: false, + ExtraSystemPrompt: extraSystem, + LightContext: hb.LightContext, + TraceName: fmt.Sprintf("Heartbeat [%s]", agentKey), + TraceTags: []string{"heartbeat"}, + }) + + outcome := <-outCh + if outcome.Err == nil { + result = outcome.Result + lastErr = nil + break + } + lastErr = outcome.Err + if attempt < maxAttempts-1 { + // Exponential backoff between retries. + time.Sleep(time.Duration(1<= startMin && nowMin < endMin + } + // Wraps midnight (e.g. 22:00 - 06:00). + return nowMin >= startMin || nowMin < endMin +} + +func parseHHMM(s string) int { + parts := strings.SplitN(s, ":", 2) + if len(parts) != 2 { + return 0 + } + h, m := 0, 0 + fmt.Sscanf(parts[0], "%d", &h) + fmt.Sscanf(parts[1], "%d", &m) + return h*60 + m +} + +func truncate(s string, maxLen int) string { + runes := []rune(s) + if len(runes) <= maxLen { + return s + } + return string(runes[:maxLen]) + "…" +} + +func deref(s *string) string { + if s == nil { + return "" + } + return *s +} diff --git a/internal/sessions/key.go b/internal/sessions/key.go index fe986d0f..ccd17833 100644 --- a/internal/sessions/key.go +++ b/internal/sessions/key.go @@ -24,6 +24,7 @@ package sessions import ( "fmt" "strings" + "time" ) // PeerKind distinguishes DM from group conversations. @@ -165,6 +166,23 @@ func IsCronSession(key string) bool { return strings.HasPrefix(strings.ToLower(rest), "cron:") } +// BuildHeartbeatSessionKey builds the session key for a heartbeat run. +// +// isolated=true: agent:{agentId}:heartbeat:{unix_ms} +// isolated=false: agent:{agentId}:heartbeat +func BuildHeartbeatSessionKey(agentID string, isolated bool) string { + if isolated { + return fmt.Sprintf("agent:%s:heartbeat:%d", agentID, time.Now().UnixMilli()) + } + return fmt.Sprintf("agent:%s:heartbeat", agentID) +} + +// IsHeartbeatSession checks if a session key indicates a heartbeat session. +func IsHeartbeatSession(key string) bool { + _, rest := ParseSessionKey(key) + return strings.HasPrefix(rest, "heartbeat") +} + // PeerKindFromGroup returns PeerGroup if isGroup is true, PeerDirect otherwise. func PeerKindFromGroup(isGroup bool) PeerKind { if isGroup { diff --git a/internal/store/config_permission_store.go b/internal/store/config_permission_store.go new file mode 100644 index 00000000..be4c9ca3 --- /dev/null +++ b/internal/store/config_permission_store.go @@ -0,0 +1,34 @@ +package store + +import ( + "context" + "encoding/json" + "time" + + "github.com/google/uuid" +) + +// ConfigPermission represents an allow/deny rule for agent configuration. +type ConfigPermission struct { + ID uuid.UUID `json:"id"` + AgentID uuid.UUID `json:"agentId"` + Scope string `json:"scope"` // "agent" | "group:telegram:-100456" | "group:*" | "*" + ConfigType string `json:"configType"` // "heartbeat" | "cron" | "context_files" | "*" + UserID string `json:"userId"` + Permission string `json:"permission"` // "allow" | "deny" + GrantedBy *string `json:"grantedBy,omitempty"` + Metadata json.RawMessage `json:"metadata,omitempty"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// ConfigPermissionStore manages agent configuration permissions with wildcard scope matching. +type ConfigPermissionStore interface { + // CheckPermission checks if a user has permission for a given config action. + // Evaluates deny rules first, then allow rules, using Go-level wildcard matching. + CheckPermission(ctx context.Context, agentID uuid.UUID, scope, configType, userID string) (bool, error) + + Grant(ctx context.Context, perm *ConfigPermission) error + Revoke(ctx context.Context, agentID uuid.UUID, scope, configType, userID string) error + List(ctx context.Context, agentID uuid.UUID, configType string) ([]ConfigPermission, error) +} diff --git a/internal/store/heartbeat_store.go b/internal/store/heartbeat_store.go new file mode 100644 index 00000000..067f5ab2 --- /dev/null +++ b/internal/store/heartbeat_store.go @@ -0,0 +1,109 @@ +package store + +import ( + "context" + "encoding/json" + "time" + + "github.com/google/uuid" +) + +// AgentHeartbeat represents the heartbeat configuration for an agent. +type AgentHeartbeat struct { + ID uuid.UUID `json:"id"` + AgentID uuid.UUID `json:"agentId"` + Enabled bool `json:"enabled"` + IntervalSec int `json:"intervalSec"` + Prompt *string `json:"prompt,omitempty"` + ProviderID *uuid.UUID `json:"providerId,omitempty"` + Model *string `json:"model,omitempty"` + IsolatedSession bool `json:"isolatedSession"` + LightContext bool `json:"lightContext"` + AckMaxChars int `json:"ackMaxChars"` + MaxRetries int `json:"maxRetries"` + ActiveHoursStart *string `json:"activeHoursStart,omitempty"` + ActiveHoursEnd *string `json:"activeHoursEnd,omitempty"` + Timezone *string `json:"timezone,omitempty"` + Channel *string `json:"channel,omitempty"` + ChatID *string `json:"chatId,omitempty"` + NextRunAt *time.Time `json:"nextRunAt,omitempty"` + LastRunAt *time.Time `json:"lastRunAt,omitempty"` + LastStatus *string `json:"lastStatus,omitempty"` + LastError *string `json:"lastError,omitempty"` + RunCount int `json:"runCount"` + SuppressCount int `json:"suppressCount"` + Metadata json.RawMessage `json:"metadata,omitempty"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// HeartbeatState holds runtime state updates for a heartbeat run. +type HeartbeatState struct { + NextRunAt *time.Time + LastRunAt *time.Time + LastStatus string + LastError string + RunCount int + SuppressCount int +} + +// HeartbeatRunLog records a single heartbeat execution. +type HeartbeatRunLog struct { + ID uuid.UUID `json:"id"` + HeartbeatID uuid.UUID `json:"heartbeatId"` + AgentID uuid.UUID `json:"agentId"` + Status string `json:"status"` + Summary *string `json:"summary,omitempty"` + Error *string `json:"error,omitempty"` + DurationMS *int `json:"durationMs,omitempty"` + InputTokens int `json:"inputTokens"` + OutputTokens int `json:"outputTokens"` + SkipReason *string `json:"skipReason,omitempty"` + Metadata json.RawMessage `json:"metadata,omitempty"` + RanAt time.Time `json:"ranAt"` + CreatedAt time.Time `json:"createdAt"` +} + +// StaggerOffset returns a deterministic offset for spreading heartbeats evenly. +// Uses FNV-1a hash of agent ID to produce a value in [0, intervalSec). +func StaggerOffset(agentID uuid.UUID, intervalSec int) time.Duration { + if intervalSec <= 0 { + return 0 + } + h := uint32(2166136261) // FNV offset basis + for _, b := range agentID { + h ^= uint32(b) + h *= 16777619 // FNV prime + } + offset := int(h) % intervalSec + if offset < 0 { + offset = -offset + } + return time.Duration(offset) * time.Second +} + +// HeartbeatEvent represents a heartbeat lifecycle event sent to subscribers. +type HeartbeatEvent struct { + Action string `json:"action"` // "running", "completed", "suppressed", "error", "skipped" + AgentID string `json:"agentId"` + AgentKey string `json:"agentKey,omitempty"` + Status string `json:"status,omitempty"` + Error string `json:"error,omitempty"` + Reason string `json:"reason,omitempty"` // skip reason +} + +// HeartbeatStore manages agent heartbeat configurations and run logs. +type HeartbeatStore interface { + Get(ctx context.Context, agentID uuid.UUID) (*AgentHeartbeat, error) + Upsert(ctx context.Context, hb *AgentHeartbeat) error + ListDue(ctx context.Context, now time.Time) ([]AgentHeartbeat, error) + UpdateState(ctx context.Context, id uuid.UUID, state HeartbeatState) error + Delete(ctx context.Context, agentID uuid.UUID) error + + // Logs + InsertLog(ctx context.Context, log *HeartbeatRunLog) error + ListLogs(ctx context.Context, agentID uuid.UUID, limit, offset int) ([]HeartbeatRunLog, int, error) + + // Events + SetOnEvent(fn func(HeartbeatEvent)) +} diff --git a/internal/store/pg/config_permissions.go b/internal/store/pg/config_permissions.go new file mode 100644 index 00000000..8b5c5173 --- /dev/null +++ b/internal/store/pg/config_permissions.go @@ -0,0 +1,191 @@ +package pg + +import ( + "context" + "database/sql" + "encoding/json" + "strings" + "sync" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +const permCacheTTL = 60 * time.Second + +// permCacheEntry holds cached permission rows for an agent+user pair. +type permCacheEntry struct { + rows []permRow + fetched time.Time +} + +type permRow struct { + Scope string + ConfigType string + Permission string +} + +// PGConfigPermissionStore implements store.ConfigPermissionStore backed by Postgres. +// Includes a TTL cache for CheckPermission to avoid per-request DB queries. +type PGConfigPermissionStore struct { + db *sql.DB + mu sync.RWMutex + cache map[string]permCacheEntry // key: "agentID:userID" +} + +func NewPGConfigPermissionStore(db *sql.DB) *PGConfigPermissionStore { + return &PGConfigPermissionStore{db: db, cache: make(map[string]permCacheEntry)} +} + +// InvalidateCache clears all cached permission entries. +func (s *PGConfigPermissionStore) InvalidateCache() { + s.mu.Lock() + defer s.mu.Unlock() + s.cache = make(map[string]permCacheEntry) +} + +// CheckPermission evaluates deny-first, allow-second permission with Go-level wildcard matching. +func (s *PGConfigPermissionStore) CheckPermission(ctx context.Context, agentID uuid.UUID, scope, configType, userID string) (bool, error) { + cacheKey := agentID.String() + ":" + userID + + // Check cache. + s.mu.RLock() + if entry, ok := s.cache[cacheKey]; ok && time.Since(entry.fetched) < permCacheTTL { + s.mu.RUnlock() + return evalPermRows(entry.rows, scope, configType), nil + } + s.mu.RUnlock() + + // Fetch from DB. + rows, err := s.db.QueryContext(ctx, + `SELECT scope, config_type, permission FROM agent_config_permissions + WHERE agent_id = $1 AND user_id = $2`, + agentID, userID, + ) + if err != nil { + return false, err + } + defer rows.Close() + + var permRows []permRow + for rows.Next() { + var r permRow + if err := rows.Scan(&r.Scope, &r.ConfigType, &r.Permission); err != nil { + return false, err + } + permRows = append(permRows, r) + } + if err := rows.Err(); err != nil { + return false, err + } + + // Update cache. + s.mu.Lock() + s.cache[cacheKey] = permCacheEntry{rows: permRows, fetched: time.Now()} + s.mu.Unlock() + + return evalPermRows(permRows, scope, configType), nil +} + +// evalPermRows evaluates cached permission rows against scope and configType. +func evalPermRows(rows []permRow, scope, configType string) bool { + var hasDeny, hasAllow bool + for _, r := range rows { + if !matchWildcard(r.Scope, scope) || !matchWildcard(r.ConfigType, configType) { + continue + } + switch r.Permission { + case "deny": + hasDeny = true + case "allow": + hasAllow = true + } + } + if hasDeny { + return false + } + return hasAllow +} + +// matchWildcard performs simple wildcard matching for scope/config_type. +// Pattern examples: "*", "group:*", "group:telegram:*", "group:telegram:-100456" +func matchWildcard(pattern, value string) bool { + if pattern == "*" { + return true + } + if strings.HasSuffix(pattern, ":*") { + prefix := strings.TrimSuffix(pattern, "*") + return strings.HasPrefix(value, prefix) || value == strings.TrimSuffix(prefix, ":") + } + return pattern == value +} + +func (s *PGConfigPermissionStore) Grant(ctx context.Context, perm *store.ConfigPermission) error { + meta := perm.Metadata + if meta == nil { + meta = json.RawMessage("{}") + } + now := time.Now() + _, err := s.db.ExecContext(ctx, + `INSERT INTO agent_config_permissions (agent_id, scope, config_type, user_id, permission, granted_by, metadata, created_at, updated_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$8) + ON CONFLICT (agent_id, scope, config_type, user_id) DO UPDATE SET + permission = EXCLUDED.permission, + granted_by = EXCLUDED.granted_by, + metadata = EXCLUDED.metadata, + updated_at = EXCLUDED.updated_at`, + perm.AgentID, perm.Scope, perm.ConfigType, perm.UserID, perm.Permission, perm.GrantedBy, meta, now, + ) + if err == nil { + s.InvalidateCache() + } + return err +} + +func (s *PGConfigPermissionStore) Revoke(ctx context.Context, agentID uuid.UUID, scope, configType, userID string) error { + _, err := s.db.ExecContext(ctx, + `DELETE FROM agent_config_permissions WHERE agent_id = $1 AND scope = $2 AND config_type = $3 AND user_id = $4`, + agentID, scope, configType, userID, + ) + if err == nil { + s.InvalidateCache() + } + return err +} + +func (s *PGConfigPermissionStore) List(ctx context.Context, agentID uuid.UUID, configType string) ([]store.ConfigPermission, error) { + query := `SELECT id, agent_id, scope, config_type, user_id, permission, granted_by, metadata, created_at, updated_at + FROM agent_config_permissions WHERE agent_id = $1` + args := []any{agentID} + + if configType != "" { + query += ` AND config_type = $2` + args = append(args, configType) + } + query += ` ORDER BY created_at` + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var perms []store.ConfigPermission + for rows.Next() { + var p store.ConfigPermission + var metadata []byte + if err := rows.Scan( + &p.ID, &p.AgentID, &p.Scope, &p.ConfigType, &p.UserID, &p.Permission, &p.GrantedBy, &metadata, &p.CreatedAt, &p.UpdatedAt, + ); err != nil { + return nil, err + } + p.Metadata = metadata + perms = append(perms, p) + } + if err := rows.Err(); err != nil { + return nil, err + } + return perms, nil +} diff --git a/internal/store/pg/factory.go b/internal/store/pg/factory.go index 002fdcf6..dd777c1e 100644 --- a/internal/store/pg/factory.go +++ b/internal/store/pg/factory.go @@ -44,6 +44,8 @@ func NewPGStores(cfg store.StoreConfig) (*store.Stores, error) { Activity: NewPGActivityStore(db), Snapshots: NewPGSnapshotStore(db), SecureCLI: NewPGSecureCLIStore(db, cfg.EncryptionKey), - APIKeys: NewPGAPIKeyStore(db), + APIKeys: NewPGAPIKeyStore(db), + Heartbeats: NewPGHeartbeatStore(db), + ConfigPermissions: NewPGConfigPermissionStore(db), }, nil } diff --git a/internal/store/pg/heartbeat.go b/internal/store/pg/heartbeat.go new file mode 100644 index 00000000..e2505637 --- /dev/null +++ b/internal/store/pg/heartbeat.go @@ -0,0 +1,271 @@ +package pg + +import ( + "context" + "database/sql" + "encoding/json" + "sync" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// PGHeartbeatStore implements store.HeartbeatStore backed by Postgres. +type PGHeartbeatStore struct { + db *sql.DB + mu sync.Mutex + onEvent func(store.HeartbeatEvent) + + // Due cache: reduces ListDue polling. + dueCache []store.AgentHeartbeat + cacheLoaded bool + cacheTime time.Time + cacheTTL time.Duration +} + +const defaultHeartbeatCacheTTL = 30 * time.Second + +func NewPGHeartbeatStore(db *sql.DB) *PGHeartbeatStore { + return &PGHeartbeatStore{db: db, cacheTTL: defaultHeartbeatCacheTTL} +} + +func (s *PGHeartbeatStore) SetOnEvent(fn func(store.HeartbeatEvent)) { + s.mu.Lock() + defer s.mu.Unlock() + s.onEvent = fn +} + +func (s *PGHeartbeatStore) emitEvent(event store.HeartbeatEvent) { + s.mu.Lock() + fn := s.onEvent + s.mu.Unlock() + if fn != nil { + fn(event) + } +} + +// InvalidateCache clears the due cache (called on config changes). +func (s *PGHeartbeatStore) InvalidateCache() { + s.mu.Lock() + defer s.mu.Unlock() + s.cacheLoaded = false +} + +func (s *PGHeartbeatStore) Get(ctx context.Context, agentID uuid.UUID) (*store.AgentHeartbeat, error) { + var hb store.AgentHeartbeat + var metadata []byte + err := s.db.QueryRowContext(ctx, + `SELECT id, agent_id, enabled, interval_sec, prompt, provider_id, model, + isolated_session, light_context, ack_max_chars, max_retries, + active_hours_start, active_hours_end, timezone, + channel, chat_id, + next_run_at, last_run_at, last_status, last_error, + run_count, suppress_count, metadata, created_at, updated_at + FROM agent_heartbeats WHERE agent_id = $1`, agentID, + ).Scan( + &hb.ID, &hb.AgentID, &hb.Enabled, &hb.IntervalSec, &hb.Prompt, &hb.ProviderID, &hb.Model, + &hb.IsolatedSession, &hb.LightContext, &hb.AckMaxChars, &hb.MaxRetries, + &hb.ActiveHoursStart, &hb.ActiveHoursEnd, &hb.Timezone, + &hb.Channel, &hb.ChatID, + &hb.NextRunAt, &hb.LastRunAt, &hb.LastStatus, &hb.LastError, + &hb.RunCount, &hb.SuppressCount, &metadata, &hb.CreatedAt, &hb.UpdatedAt, + ) + if err != nil { + return nil, err + } + hb.Metadata = metadata + return &hb, nil +} + +func (s *PGHeartbeatStore) Upsert(ctx context.Context, hb *store.AgentHeartbeat) error { + meta := hb.Metadata + if meta == nil { + meta = json.RawMessage("{}") + } + now := time.Now() + err := s.db.QueryRowContext(ctx, + `INSERT INTO agent_heartbeats (agent_id, enabled, interval_sec, prompt, provider_id, model, + isolated_session, light_context, ack_max_chars, max_retries, + active_hours_start, active_hours_end, timezone, + channel, chat_id, metadata, created_at, updated_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$17) + ON CONFLICT (agent_id) DO UPDATE SET + enabled = EXCLUDED.enabled, + interval_sec = EXCLUDED.interval_sec, + prompt = EXCLUDED.prompt, + provider_id = EXCLUDED.provider_id, + model = EXCLUDED.model, + isolated_session = EXCLUDED.isolated_session, + light_context = EXCLUDED.light_context, + ack_max_chars = EXCLUDED.ack_max_chars, + max_retries = EXCLUDED.max_retries, + active_hours_start = EXCLUDED.active_hours_start, + active_hours_end = EXCLUDED.active_hours_end, + timezone = EXCLUDED.timezone, + channel = EXCLUDED.channel, + chat_id = EXCLUDED.chat_id, + metadata = EXCLUDED.metadata, + updated_at = EXCLUDED.updated_at + RETURNING id, created_at, updated_at`, + hb.AgentID, hb.Enabled, hb.IntervalSec, hb.Prompt, hb.ProviderID, hb.Model, + hb.IsolatedSession, hb.LightContext, hb.AckMaxChars, hb.MaxRetries, + hb.ActiveHoursStart, hb.ActiveHoursEnd, hb.Timezone, + hb.Channel, hb.ChatID, meta, now, + ).Scan(&hb.ID, &hb.CreatedAt, &hb.UpdatedAt) + if err != nil { + return err + } + s.InvalidateCache() + return nil +} + +func (s *PGHeartbeatStore) ListDue(ctx context.Context, now time.Time) ([]store.AgentHeartbeat, error) { + s.mu.Lock() + if s.cacheLoaded && time.Since(s.cacheTime) < s.cacheTTL { + cached := s.dueCache + s.mu.Unlock() + // Filter in-memory for due items. + var due []store.AgentHeartbeat + for _, hb := range cached { + if hb.NextRunAt != nil && !hb.NextRunAt.After(now) { + due = append(due, hb) + } + } + return due, nil + } + s.mu.Unlock() + + rows, err := s.db.QueryContext(ctx, + `SELECT id, agent_id, enabled, interval_sec, prompt, provider_id, model, + isolated_session, light_context, ack_max_chars, max_retries, + active_hours_start, active_hours_end, timezone, + channel, chat_id, + next_run_at, last_run_at, last_status, last_error, + run_count, suppress_count, metadata, created_at, updated_at + FROM agent_heartbeats + WHERE enabled = true AND next_run_at IS NOT NULL`) + if err != nil { + return nil, err + } + defer rows.Close() + + var all []store.AgentHeartbeat + for rows.Next() { + var hb store.AgentHeartbeat + var metadata []byte + if err := rows.Scan( + &hb.ID, &hb.AgentID, &hb.Enabled, &hb.IntervalSec, &hb.Prompt, &hb.ProviderID, &hb.Model, + &hb.IsolatedSession, &hb.LightContext, &hb.AckMaxChars, &hb.MaxRetries, + &hb.ActiveHoursStart, &hb.ActiveHoursEnd, &hb.Timezone, + &hb.Channel, &hb.ChatID, + &hb.NextRunAt, &hb.LastRunAt, &hb.LastStatus, &hb.LastError, + &hb.RunCount, &hb.SuppressCount, &metadata, &hb.CreatedAt, &hb.UpdatedAt, + ); err != nil { + return nil, err + } + hb.Metadata = metadata + all = append(all, hb) + } + if err := rows.Err(); err != nil { + return nil, err + } + + // Update cache. + s.mu.Lock() + s.dueCache = all + s.cacheLoaded = true + s.cacheTime = time.Now() + s.mu.Unlock() + + // Filter for due. + var due []store.AgentHeartbeat + for _, hb := range all { + if hb.NextRunAt != nil && !hb.NextRunAt.After(now) { + due = append(due, hb) + } + } + return due, nil +} + +func (s *PGHeartbeatStore) UpdateState(ctx context.Context, id uuid.UUID, state store.HeartbeatState) error { + _, err := s.db.ExecContext(ctx, + `UPDATE agent_heartbeats SET + next_run_at = $2, last_run_at = $3, last_status = $4, last_error = $5, + run_count = $6, suppress_count = $7, updated_at = NOW() + WHERE id = $1`, + id, state.NextRunAt, state.LastRunAt, state.LastStatus, state.LastError, + state.RunCount, state.SuppressCount, + ) + return err +} + +func (s *PGHeartbeatStore) Delete(ctx context.Context, agentID uuid.UUID) error { + _, err := s.db.ExecContext(ctx, + `DELETE FROM agent_heartbeats WHERE agent_id = $1`, agentID) + if err == nil { + s.InvalidateCache() + } + return err +} + +// InsertLog records a heartbeat run execution. +func (s *PGHeartbeatStore) InsertLog(ctx context.Context, log *store.HeartbeatRunLog) error { + meta := log.Metadata + if meta == nil { + meta = json.RawMessage("{}") + } + _, err := s.db.ExecContext(ctx, + `INSERT INTO heartbeat_run_logs (heartbeat_id, agent_id, status, summary, error, + duration_ms, input_tokens, output_tokens, skip_reason, metadata, ran_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`, + log.HeartbeatID, log.AgentID, log.Status, log.Summary, log.Error, + log.DurationMS, log.InputTokens, log.OutputTokens, log.SkipReason, meta, log.RanAt, + ) + return err +} + +func (s *PGHeartbeatStore) ListLogs(ctx context.Context, agentID uuid.UUID, limit, offset int) ([]store.HeartbeatRunLog, int, error) { + if limit <= 0 { + limit = 20 + } + + var total int + err := s.db.QueryRowContext(ctx, + `SELECT COUNT(*) FROM heartbeat_run_logs WHERE agent_id = $1`, agentID, + ).Scan(&total) + if err != nil { + return nil, 0, err + } + + rows, err := s.db.QueryContext(ctx, + `SELECT id, heartbeat_id, agent_id, status, summary, error, + duration_ms, input_tokens, output_tokens, skip_reason, metadata, ran_at, created_at + FROM heartbeat_run_logs WHERE agent_id = $1 + ORDER BY ran_at DESC LIMIT $2 OFFSET $3`, + agentID, limit, offset, + ) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + var logs []store.HeartbeatRunLog + for rows.Next() { + var l store.HeartbeatRunLog + var metadata []byte + if err := rows.Scan( + &l.ID, &l.HeartbeatID, &l.AgentID, &l.Status, &l.Summary, &l.Error, + &l.DurationMS, &l.InputTokens, &l.OutputTokens, &l.SkipReason, &metadata, &l.RanAt, &l.CreatedAt, + ); err != nil { + return nil, 0, err + } + l.Metadata = metadata + logs = append(logs, l) + } + if err := rows.Err(); err != nil { + return nil, 0, err + } + return logs, total, nil +} diff --git a/internal/store/stores.go b/internal/store/stores.go index d039f904..58c0909a 100644 --- a/internal/store/stores.go +++ b/internal/store/stores.go @@ -26,5 +26,7 @@ type Stores struct { Activity ActivityStore Snapshots SnapshotStore SecureCLI SecureCLIStore - APIKeys APIKeyStore + APIKeys APIKeyStore + Heartbeats HeartbeatStore + ConfigPermissions ConfigPermissionStore } diff --git a/internal/tools/context_file_interceptor.go b/internal/tools/context_file_interceptor.go index 6c3249d4..0cb5cbd7 100644 --- a/internal/tools/context_file_interceptor.go +++ b/internal/tools/context_file_interceptor.go @@ -34,6 +34,7 @@ var contextFileSet = map[string]bool{ bootstrap.UserFile: true, bootstrap.UserPredefinedFile: true, bootstrap.BootstrapFile: true, // first-run file (deleted after completion) + bootstrap.HeartbeatFile: true, // agent-level heartbeat checklist } // isContextFile checks if a path refers to a workspace-root context file. @@ -149,7 +150,7 @@ func (b *ContextFileInterceptor) ReadFile(ctx context.Context, path string) (str // Predefined agent: block reads of shared identity files (SOUL.md, IDENTITY.md, AGENTS.md). // These are already injected into the system prompt — allowing read_file would let the // agent echo their full contents to users, leaking persona configuration. - if agentType == store.AgentTypePredefined && fileName != bootstrap.UserFile && fileName != bootstrap.BootstrapFile { + if agentType == store.AgentTypePredefined && fileName != bootstrap.UserFile && fileName != bootstrap.BootstrapFile && fileName != bootstrap.HeartbeatFile { return "", true, fmt.Errorf( "this file (%s) is already loaded into your context. You don't need to read it again — refer to your system instructions instead.", fileName, @@ -241,9 +242,9 @@ func (b *ContextFileInterceptor) WriteFile(ctx context.Context, path, content st return true, err } - // Predefined agent: block writes to shared files (only USER.md allowed per-user). + // Predefined agent: block writes to shared files (only USER.md + HEARTBEAT.md allowed). // Exception: SOUL.md is allowed when self_evolve is enabled (style/tone evolution). - if agentType == store.AgentTypePredefined && fileName != bootstrap.UserFile { + if agentType == store.AgentTypePredefined && fileName != bootstrap.UserFile && fileName != bootstrap.HeartbeatFile { allowSoulEvolve := fileName == bootstrap.SoulFile && store.SelfEvolveFromContext(ctx) if !allowSoulEvolve { return true, fmt.Errorf( diff --git a/internal/tools/heartbeat.go b/internal/tools/heartbeat.go new file mode 100644 index 00000000..5852e5bf --- /dev/null +++ b/internal/tools/heartbeat.go @@ -0,0 +1,412 @@ +package tools + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// HeartbeatTool lets agents manage their own heartbeat configuration. +type HeartbeatTool struct { + hbStore store.HeartbeatStore + permStore store.ConfigPermissionStore + agentStore store.AgentStore + wakeFn func(uuid.UUID) // triggers immediate heartbeat run +} + +// NewHeartbeatTool creates a heartbeat management tool. +func NewHeartbeatTool(hb store.HeartbeatStore, perms store.ConfigPermissionStore) *HeartbeatTool { + return &HeartbeatTool{hbStore: hb, permStore: perms} +} + +// SetAgentStore sets the agent store for HEARTBEAT.md read/write. +func (t *HeartbeatTool) SetAgentStore(as store.AgentStore) { + t.agentStore = as +} + +// SetWakeFn sets the function called when the "test" action triggers an immediate run. +func (t *HeartbeatTool) SetWakeFn(fn func(uuid.UUID)) { + t.wakeFn = fn +} + +func (t *HeartbeatTool) Name() string { return "heartbeat" } + +func (t *HeartbeatTool) Description() string { + return `Manage agent heartbeat — periodic proactive check-in. + +ACTIONS: +- status: Show heartbeat status (enabled, last/next run, counts) +- get: Get full heartbeat configuration +- set: Create or update heartbeat config (interval, active_hours, ack_max_chars, etc.) +- toggle: Enable or disable heartbeat (enabled: true/false) +- set_checklist: Set HEARTBEAT.md content (the checklist the agent follows each run) +- get_checklist: Read current HEARTBEAT.md content +- test: Trigger an immediate heartbeat run (background) +- logs: View heartbeat run history (limit, offset) + +EXAMPLES: + {"action":"status"} + {"action":"set","interval":1800,"channel":"telegram","chat_id":"-100123456"} + {"action":"toggle","enabled":true} + {"action":"set_checklist","content":"# Heartbeat Checklist\n- Check server status\n- Report any alerts"} + {"action":"test"} + {"action":"logs","limit":10}` +} + +func (t *HeartbeatTool) Parameters() map[string]any { + return map[string]any{ + "type": "object", + "properties": map[string]any{ + "action": map[string]any{ + "type": "string", + "enum": []string{"status", "get", "set", "toggle", "set_checklist", "get_checklist", "test", "logs"}, + "description": "Action to perform", + }, + "enabled": map[string]any{"type": "boolean", "description": "For toggle: enable/disable heartbeat"}, + "interval": map[string]any{"type": "integer", "description": "Heartbeat interval in seconds (min 300)"}, + "prompt": map[string]any{"type": "string", "description": "Custom heartbeat prompt (empty = default)"}, + "ack_max_chars": map[string]any{"type": "integer", "description": "Max chars for suppression threshold (default 300)"}, + "max_retries": map[string]any{"type": "integer", "description": "Max retry attempts on failure (default 2)"}, + "isolated_session": map[string]any{"type": "boolean", "description": "Use isolated session per run (default true)"}, + "light_context": map[string]any{"type": "boolean", "description": "Skip loading context files, only inject heartbeat checklist (default false)"}, + "active_hours": map[string]any{"type": "string", "description": "Active hours range, e.g. '08:00-22:00'"}, + "timezone": map[string]any{"type": "string", "description": "IANA timezone for active hours, e.g. 'Asia/Ho_Chi_Minh'"}, + "channel": map[string]any{"type": "string", "description": "Delivery channel name (auto-filled from current context if empty)"}, + "chat_id": map[string]any{"type": "string", "description": "Delivery target chat ID (auto-filled from current context if empty)"}, + "content": map[string]any{"type": "string", "description": "For set_checklist: HEARTBEAT.md content (the checklist the agent follows each run)"}, + "limit": map[string]any{"type": "integer", "description": "For logs: max entries to return"}, + "offset": map[string]any{"type": "integer", "description": "For logs: pagination offset"}, + }, + "required": []string{"action"}, + } +} + +func (t *HeartbeatTool) Execute(ctx context.Context, args map[string]any) *Result { + action, _ := args["action"].(string) + if action == "" { + return ErrorResult("action parameter is required") + } + + agentID := store.AgentIDFromContext(ctx) + if agentID == uuid.Nil { + return ErrorResult("no agent context") + } + + // Permission check for mutation actions. + switch action { + case "set", "toggle", "set_checklist": + if err := t.checkPermission(ctx, agentID); err != nil { + return ErrorResult(err.Error()) + } + } + + switch action { + case "status": + return t.handleStatus(ctx, agentID) + case "get": + return t.handleGet(ctx, agentID) + case "set": + return t.handleSet(ctx, agentID, args) + case "toggle": + enabled, ok := args["enabled"].(bool) + if !ok { + return ErrorResult("toggle requires 'enabled' parameter (boolean)") + } + return t.handleToggle(ctx, agentID, enabled) + case "set_checklist": + content, _ := args["content"].(string) + return t.handleSetChecklist(ctx, agentID, content) + case "get_checklist": + return t.handleGetChecklist(ctx, agentID) + case "test": + return t.handleTest(agentID) + case "logs": + limit := intArg(args, "limit", 10) + offset := intArg(args, "offset", 0) + return t.handleLogs(ctx, agentID, limit, offset) + default: + return ErrorResult(fmt.Sprintf("unknown action %q — use status/get/set/toggle/set_checklist/get_checklist/test/logs", action)) + } +} + +func (t *HeartbeatTool) handleStatus(ctx context.Context, agentID uuid.UUID) *Result { + hb, err := t.hbStore.Get(ctx, agentID) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return NewResult("Heartbeat not configured for this agent.") + } + return ErrorResult(err.Error()) + } + return NewResult(fmt.Sprintf( + "Heartbeat: %s | interval: %ds | runs: %d | suppressed: %d | last: %s (%s) | next: %s", + boolLabel(hb.Enabled, "enabled", "disabled"), + hb.IntervalSec, hb.RunCount, hb.SuppressCount, + fmtTimePtr(hb.LastRunAt), derefStr(hb.LastStatus, "never"), + fmtTimePtr(hb.NextRunAt), + )) +} + +func (t *HeartbeatTool) handleGet(ctx context.Context, agentID uuid.UUID) *Result { + hb, err := t.hbStore.Get(ctx, agentID) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return NewResult("Heartbeat not configured for this agent.") + } + return ErrorResult(err.Error()) + } + data, _ := json.MarshalIndent(hb, "", " ") + return NewResult(string(data)) +} + +func (t *HeartbeatTool) handleSet(ctx context.Context, agentID uuid.UUID, args map[string]any) *Result { + hb, err := t.hbStore.Get(ctx, agentID) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return ErrorResult(err.Error()) + } + if hb == nil { + hb = &store.AgentHeartbeat{ + AgentID: agentID, + IntervalSec: 1800, + IsolatedSession: true, + AckMaxChars: 300, + MaxRetries: 2, + } + } + + if v, ok := args["enabled"].(bool); ok { + hb.Enabled = v + } + if v := intArg(args, "interval", 0); v > 0 { + if v < 300 { + return ErrorResult("minimum interval is 300 seconds (5 minutes)") + } + hb.IntervalSec = v + } + if v, ok := args["prompt"].(string); ok { + hb.Prompt = &v + } + if v := intArg(args, "ack_max_chars", 0); v > 0 { + hb.AckMaxChars = v + } + if v := intArg(args, "max_retries", -1); v >= 0 { + hb.MaxRetries = v + } + if v, ok := args["isolated_session"].(bool); ok { + hb.IsolatedSession = v + } + if v, ok := args["light_context"].(bool); ok { + hb.LightContext = v + } + if v, ok := args["timezone"].(string); ok { + hb.Timezone = &v + } + if v, ok := args["active_hours"].(string); ok { + start, end := parseActiveHoursRange(v) + hb.ActiveHoursStart = &start + hb.ActiveHoursEnd = &end + } + + // Auto-fill delivery from context. + if v, ok := args["channel"].(string); ok { + hb.Channel = &v + } else if hb.Channel == nil { + if ch := ToolChannelFromCtx(ctx); ch != "" { + hb.Channel = &ch + } + } + if v, ok := args["chat_id"].(string); ok { + hb.ChatID = &v + } else if hb.ChatID == nil { + if cid := ToolChatIDFromCtx(ctx); cid != "" { + hb.ChatID = &cid + } + } + + if hb.Enabled && hb.NextRunAt == nil { + nextRun := time.Now().Add(time.Duration(hb.IntervalSec)*time.Second + store.StaggerOffset(hb.AgentID, hb.IntervalSec)) + hb.NextRunAt = &nextRun + } + + if err := t.hbStore.Upsert(ctx, hb); err != nil { + return ErrorResult(fmt.Sprintf("failed to save heartbeat config: %v", err)) + } + + return NewResult(fmt.Sprintf("Heartbeat config saved. Enabled: %v, interval: %ds, delivery: %s/%s", + hb.Enabled, hb.IntervalSec, derefStr(hb.Channel, "none"), derefStr(hb.ChatID, "none"))) +} + +func (t *HeartbeatTool) handleToggle(ctx context.Context, agentID uuid.UUID, enabled bool) *Result { + hb, err := t.hbStore.Get(ctx, agentID) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return ErrorResult("heartbeat not configured — use set action first") + } + return ErrorResult(err.Error()) + } + hb.Enabled = enabled + if enabled && hb.NextRunAt == nil { + nextRun := time.Now().Add(time.Duration(hb.IntervalSec) * time.Second) + hb.NextRunAt = &nextRun + } + if err := t.hbStore.Upsert(ctx, hb); err != nil { + return ErrorResult(err.Error()) + } + return NewResult(fmt.Sprintf("Heartbeat %s.", boolLabel(enabled, "enabled", "disabled"))) +} + +// checkPermission verifies the calling user has permission to modify heartbeat config. +// Flow: deny list → allow list → agent owner/shares fallback. +// Read-only actions (status, get, get_checklist, logs) skip this check. +func (t *HeartbeatTool) checkPermission(ctx context.Context, agentID uuid.UUID) error { + if t.permStore == nil { + return nil // no permission store = allow all (backward compat) + } + userID := store.UserIDFromContext(ctx) + if userID == "" { + return nil // system context (cron, subagent) = allow + } + + // Determine scope from context: "agent" for DM, "group:{channel}:{chatId}" for groups. + scope := "agent" + if ch := ToolChannelFromCtx(ctx); ch != "" { + if cid := ToolChatIDFromCtx(ctx); cid != "" { + scope = "group:" + ch + ":" + cid + } + } + + allowed, err := t.permStore.CheckPermission(ctx, agentID, scope, "heartbeat", userID) + if err != nil { + return fmt.Errorf("permission check failed: %w", err) + } + if allowed { + return nil + } + + // Fallback: check if user is agent owner (via agent store). + if t.agentStore != nil { + ag, agErr := t.agentStore.GetByID(ctx, agentID) + if agErr == nil { + senderID := store.SenderIDFromContext(ctx) + if senderID == "" { + senderID = userID + } + if ag.OwnerID != "" && ag.OwnerID == senderID { + return nil // agent owner = allow + } + } + } + + return fmt.Errorf("permission denied: you are not authorized to modify heartbeat config for this agent") +} + +func (t *HeartbeatTool) handleSetChecklist(ctx context.Context, agentID uuid.UUID, content string) *Result { + if t.agentStore == nil { + return ErrorResult("agent store not configured") + } + if content == "" { + return ErrorResult("content parameter is required — provide the HEARTBEAT.md checklist") + } + if err := t.agentStore.SetAgentContextFile(ctx, agentID, "HEARTBEAT.md", content); err != nil { + return ErrorResult(fmt.Sprintf("failed to save HEARTBEAT.md: %v", err)) + } + return NewResult(fmt.Sprintf("HEARTBEAT.md saved (%d chars). The heartbeat ticker will use this checklist on each run.", len([]rune(content)))) +} + +func (t *HeartbeatTool) handleGetChecklist(ctx context.Context, agentID uuid.UUID) *Result { + if t.agentStore == nil { + return ErrorResult("agent store not configured") + } + files, err := t.agentStore.GetAgentContextFiles(ctx, agentID) + if err != nil { + return ErrorResult(err.Error()) + } + for _, f := range files { + if f.FileName == "HEARTBEAT.md" && f.Content != "" { + return NewResult(f.Content) + } + } + return NewResult("HEARTBEAT.md not found. Use set_checklist to create one.") +} + +func (t *HeartbeatTool) handleTest(agentID uuid.UUID) *Result { + if t.wakeFn == nil { + return ErrorResult("heartbeat ticker not available") + } + t.wakeFn(agentID) + return NewResult("Heartbeat test triggered (running in background).") +} + +func (t *HeartbeatTool) handleLogs(ctx context.Context, agentID uuid.UUID, limit, offset int) *Result { + logs, total, err := t.hbStore.ListLogs(ctx, agentID, limit, offset) + if err != nil { + return ErrorResult(err.Error()) + } + if len(logs) == 0 { + return NewResult("No heartbeat run logs found.") + } + data, _ := json.MarshalIndent(map[string]any{ + "total": total, + "logs": logs, + }, "", " ") + return NewResult(string(data)) +} + +// parseActiveHoursRange parses "HH:MM-HH:MM" into start and end strings. +func parseActiveHoursRange(s string) (string, string) { + for _, sep := range []string{"-", " - ", "~"} { + parts := splitOnce(s, sep) + if len(parts) == 2 { + return parts[0], parts[1] + } + } + return s, "" +} + +func splitOnce(s, sep string) []string { + idx := -1 + for i := range len(s) - len(sep) + 1 { + if s[i:i+len(sep)] == sep { + idx = i + break + } + } + if idx < 0 { + return []string{s} + } + return []string{s[:idx], s[idx+len(sep):]} +} + +func intArg(args map[string]any, key string, fallback int) int { + if v, ok := args[key].(float64); ok { + return int(v) + } + return fallback +} + +func boolLabel(v bool, trueStr, falseStr string) string { + if v { + return trueStr + } + return falseStr +} + +func derefStr(s *string, fallback string) string { + if s == nil || *s == "" { + return fallback + } + return *s +} + +func fmtTimePtr(t *time.Time) string { + if t == nil { + return "never" + } + return t.Format(time.RFC3339) +} diff --git a/internal/upgrade/version.go b/internal/upgrade/version.go index fb56bf40..148ee80e 100644 --- a/internal/upgrade/version.go +++ b/internal/upgrade/version.go @@ -2,4 +2,4 @@ package upgrade // RequiredSchemaVersion is the schema migration version this binary requires. // Bump this whenever adding a new SQL migration file. -const RequiredSchemaVersion uint = 21 +const RequiredSchemaVersion uint = 22 diff --git a/migrations/000022_agent_heartbeats.down.sql b/migrations/000022_agent_heartbeats.down.sql new file mode 100644 index 00000000..77b9c414 --- /dev/null +++ b/migrations/000022_agent_heartbeats.down.sql @@ -0,0 +1,3 @@ +DROP TABLE IF EXISTS agent_config_permissions; +DROP TABLE IF EXISTS heartbeat_run_logs; +DROP TABLE IF EXISTS agent_heartbeats; diff --git a/migrations/000022_agent_heartbeats.up.sql b/migrations/000022_agent_heartbeats.up.sql new file mode 100644 index 00000000..d48491ae --- /dev/null +++ b/migrations/000022_agent_heartbeats.up.sql @@ -0,0 +1,68 @@ +-- Agent heartbeat configuration (per-agent, not per-user). +CREATE TABLE agent_heartbeats ( + id UUID PRIMARY KEY DEFAULT uuid_generate_v7(), + agent_id UUID NOT NULL UNIQUE REFERENCES agents(id) ON DELETE CASCADE, + enabled BOOLEAN NOT NULL DEFAULT false, + interval_sec INT NOT NULL DEFAULT 1800, + prompt TEXT, + provider_id UUID REFERENCES llm_providers(id), + model VARCHAR(200), + isolated_session BOOLEAN NOT NULL DEFAULT true, + light_context BOOLEAN NOT NULL DEFAULT false, + ack_max_chars INT NOT NULL DEFAULT 300, + max_retries INT NOT NULL DEFAULT 2, + active_hours_start VARCHAR(5), + active_hours_end VARCHAR(5), + timezone TEXT, + channel VARCHAR(50), + chat_id TEXT, + next_run_at TIMESTAMPTZ, + last_run_at TIMESTAMPTZ, + last_status VARCHAR(20), + last_error TEXT, + run_count INT NOT NULL DEFAULT 0, + suppress_count INT NOT NULL DEFAULT 0, + metadata JSONB DEFAULT '{}', + created_at TIMESTAMPTZ DEFAULT NOW(), + updated_at TIMESTAMPTZ DEFAULT NOW() +); + +CREATE INDEX idx_heartbeats_due ON agent_heartbeats (next_run_at) + WHERE enabled = true AND next_run_at IS NOT NULL; + +-- Heartbeat execution logs. +CREATE TABLE heartbeat_run_logs ( + id UUID PRIMARY KEY DEFAULT uuid_generate_v7(), + heartbeat_id UUID NOT NULL REFERENCES agent_heartbeats(id) ON DELETE CASCADE, + agent_id UUID NOT NULL REFERENCES agents(id), + status VARCHAR(20) NOT NULL, + summary TEXT, + error TEXT, + duration_ms INT, + input_tokens INT DEFAULT 0, + output_tokens INT DEFAULT 0, + skip_reason VARCHAR(50), + metadata JSONB DEFAULT '{}', + ran_at TIMESTAMPTZ DEFAULT NOW(), + created_at TIMESTAMPTZ DEFAULT NOW() +); + +CREATE INDEX idx_hb_logs_heartbeat ON heartbeat_run_logs (heartbeat_id, ran_at DESC); +CREATE INDEX idx_hb_logs_agent ON heartbeat_run_logs (agent_id, ran_at DESC); + +-- Generic agent config permissions (heartbeat, cron, context_files, etc.) +CREATE TABLE agent_config_permissions ( + id UUID PRIMARY KEY DEFAULT uuid_generate_v7(), + agent_id UUID NOT NULL REFERENCES agents(id) ON DELETE CASCADE, + scope VARCHAR(100) NOT NULL, + config_type VARCHAR(50) NOT NULL, + user_id VARCHAR(255) NOT NULL, + permission VARCHAR(10) NOT NULL, + granted_by VARCHAR(255), + metadata JSONB DEFAULT '{}', + created_at TIMESTAMPTZ DEFAULT NOW(), + updated_at TIMESTAMPTZ DEFAULT NOW(), + UNIQUE(agent_id, scope, config_type, user_id) +); + +CREATE INDEX idx_acp_lookup ON agent_config_permissions (agent_id, scope, config_type); diff --git a/pkg/protocol/events.go b/pkg/protocol/events.go index 043efa88..953615bc 100644 --- a/pkg/protocol/events.go +++ b/pkg/protocol/events.go @@ -6,6 +6,7 @@ const ( EventChat = "chat" EventHealth = "health" EventCron = "cron" + EventHeartbeat = "heartbeat" EventExecApprovalReq = "exec.approval.requested" EventExecApprovalRes = "exec.approval.resolved" EventPresence = "presence" diff --git a/pkg/protocol/methods.go b/pkg/protocol/methods.go index 6f160bed..fbbc7531 100644 --- a/pkg/protocol/methods.go +++ b/pkg/protocol/methods.go @@ -83,6 +83,17 @@ const ( MethodSend = "send" ) +// Agent heartbeat +const ( + MethodHeartbeatGet = "heartbeat.get" + MethodHeartbeatSet = "heartbeat.set" + MethodHeartbeatToggle = "heartbeat.toggle" + MethodHeartbeatTest = "heartbeat.test" + MethodHeartbeatLogs = "heartbeat.logs" + MethodHeartbeatChecklistGet = "heartbeat.checklist.get" + MethodHeartbeatChecklistSet = "heartbeat.checklist.set" +) + // Channel instances management const ( MethodChannelInstancesList = "channels.instances.list"