feat: agent heartbeat system — periodic proactive check-ins (#245)

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}💓{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)
This commit is contained in:
viettranx committed 2026-03-18 13:11:44 +07:00
1 parent da59377387
commit 08a2d95c0c
25 files changed
+2088 -16

No files matched your search

+33 -2
View File
@@ -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()
}
+15
View File
@@ -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)
}
}
+28
View File
@@ -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) {
+7 -3
View File
@@ -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
}
+7 -3
View File
@@ -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 {
+2 -1
View File
@@ -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",
+8 -1
View File
@@ -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.
+4
View File
@@ -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"
+1
View File
@@ -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 ---
+400
View File
@@ -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, &params)
}
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, &params)
}
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, &params)
}
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, &params)
}
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, &params)
}
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, &params)
}
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, &params)
}
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,
},
})
}
+454
View File
@@ -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<<uint(attempt)) * time.Second)
}
}
duration := time.Since(start)
durationMS := int(duration.Milliseconds())
// [6] Process result.
if lastErr != nil {
t.finishRun(ctx, hb, agentKey, "error", lastErr.Error(), "", durationMS, 0, 0)
return
}
// [7] Smart suppression.
deliver, cleaned := processResponse(result.Content, hb.AckMaxChars)
var inputTokens, outputTokens int
if result.Usage != nil {
inputTokens = result.Usage.PromptTokens
outputTokens = result.Usage.CompletionTokens
}
if !deliver {
t.finishRun(ctx, hb, agentKey, "suppressed", "", truncate(result.Content, maxSummaryLen), durationMS, inputTokens, outputTokens)
return
}
// [8] Deliver to channel.
if hb.Channel != nil && *hb.Channel != "" && hb.ChatID != nil && *hb.ChatID != "" {
t.msgBus.PublishOutbound(bus.OutboundMessage{
Channel: *hb.Channel,
ChatID: *hb.ChatID,
Content: cleaned,
})
}
t.finishRun(ctx, hb, agentKey, "ok", "", truncate(cleaned, maxSummaryLen), durationMS, inputTokens, outputTokens)
}
func (t *Ticker) finishRun(ctx context.Context, hb store.AgentHeartbeat, agentKey, status, errMsg, summary string, durationMS, inputTokens, outputTokens int) {
agentIDStr := hb.AgentID.String()
now := time.Now()
// Insert log.
logEntry := &store.HeartbeatRunLog{
HeartbeatID: hb.ID,
AgentID: hb.AgentID,
Status: status,
DurationMS: &durationMS,
InputTokens: inputTokens,
OutputTokens: outputTokens,
RanAt: now,
}
if summary != "" {
logEntry.Summary = &summary
}
if errMsg != "" {
logEntry.Error = &errMsg
}
if err := t.store.InsertLog(ctx, logEntry); err != nil {
slog.Warn("heartbeat.insert_log_failed", "agent_id", agentIDStr, "error", err)
}
// Update state.
newState := store.HeartbeatState{
LastRunAt: &now,
LastStatus: status,
RunCount: hb.RunCount + 1,
}
if status == "suppressed" {
newState.SuppressCount = hb.SuppressCount + 1
} else {
newState.SuppressCount = hb.SuppressCount
}
if errMsg != "" {
newState.LastError = errMsg
}
nextRun := now.Add(time.Duration(hb.IntervalSec) * time.Second)
newState.NextRunAt = &nextRun
if err := t.store.UpdateState(ctx, hb.ID, newState); err != nil {
slog.Warn("heartbeat.update_state_failed", "agent_id", agentIDStr, "error", err)
}
// Emit event.
t.emitEvent(store.HeartbeatEvent{
Action: status,
AgentID: agentIDStr,
AgentKey: agentKey,
Status: status,
Error: errMsg,
})
}
func (t *Ticker) logSkipped(ctx context.Context, hb store.AgentHeartbeat, reason, agentKey string) {
now := time.Now()
logEntry := &store.HeartbeatRunLog{
HeartbeatID: hb.ID,
AgentID: hb.AgentID,
Status: "skipped",
SkipReason: &reason,
RanAt: now,
}
if err := t.store.InsertLog(ctx, logEntry); err != nil {
slog.Warn("heartbeat.insert_skip_log_failed", "agent_id", hb.AgentID, "error", err)
}
t.emitEvent(store.HeartbeatEvent{
Action: "skipped",
AgentID: hb.AgentID.String(),
AgentKey: agentKey,
Reason: reason,
})
}
func (t *Ticker) advanceNextRun(ctx context.Context, hb store.AgentHeartbeat) {
nextRun := time.Now().Add(time.Duration(hb.IntervalSec) * time.Second)
state := store.HeartbeatState{
NextRunAt: &nextRun,
LastStatus: deref(hb.LastStatus),
RunCount: hb.RunCount,
SuppressCount: hb.SuppressCount,
}
if hb.LastRunAt != nil {
state.LastRunAt = hb.LastRunAt
}
if err := t.store.UpdateState(ctx, hb.ID, state); err != nil {
slog.Warn("heartbeat.advance_next_run_failed", "agent_id", hb.AgentID, "error", err)
}
}
func (t *Ticker) readChecklist(ctx context.Context, agentID uuid.UUID) string {
if t.agents == nil {
return ""
}
files, err := t.agents.GetAgentContextFiles(ctx, agentID)
if err != nil {
return ""
}
for _, f := range files {
if f.FileName == "HEARTBEAT.md" && f.Content != "" {
return f.Content
}
}
return ""
}
// processResponse implements smart suppression.
// If response contains HEARTBEAT_OK and cleaned content is within threshold, suppress delivery.
func processResponse(response string, ackMaxChars int) (deliver bool, cleaned string) {
const ackToken = "HEARTBEAT_OK"
if !strings.Contains(response, ackToken) {
return true, response // has alert, deliver as-is
}
cleaned = strings.ReplaceAll(response, ackToken, "")
cleaned = strings.TrimSpace(cleaned)
if len([]rune(cleaned)) <= ackMaxChars {
return false, "" // suppress
}
return true, cleaned // over threshold, deliver cleaned version
}
// isWithinActiveHours checks if current time falls within the configured active hours.
func isWithinActiveHours(hb store.AgentHeartbeat) bool {
if hb.ActiveHoursStart == nil || hb.ActiveHoursEnd == nil {
return true // no active hours = 24/7
}
startStr := *hb.ActiveHoursStart
endStr := *hb.ActiveHoursEnd
if startStr == "" || endStr == "" {
return true
}
loc := time.UTC
if hb.Timezone != nil && *hb.Timezone != "" {
if parsed, err := time.LoadLocation(*hb.Timezone); err == nil {
loc = parsed
}
}
now := time.Now().In(loc)
startMin := parseHHMM(startStr)
endMin := parseHHMM(endStr)
nowMin := now.Hour()*60 + now.Minute()
if startMin <= endMin {
return nowMin >= 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
}
+18
View File
@@ -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 {
+34
View File
@@ -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)
}
+109
View File
@@ -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))
}
+191
View File
@@ -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
}
+3 -1
View File
@@ -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
}
+271
View File
@@ -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
}
+3 -1
View File
@@ -26,5 +26,7 @@ type Stores struct {
Activity ActivityStore
Snapshots SnapshotStore
SecureCLI SecureCLIStore
APIKeys APIKeyStore
APIKeys APIKeyStore
Heartbeats HeartbeatStore
ConfigPermissions ConfigPermissionStore
}
+4 -3
View File
@@ -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(
+412
View File
@@ -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)
}
+1 -1
View File
@@ -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
@@ -0,0 +1,3 @@
DROP TABLE IF EXISTS agent_config_permissions;
DROP TABLE IF EXISTS heartbeat_run_logs;
DROP TABLE IF EXISTS agent_heartbeats;
+68
View File
@@ -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);
+1
View File
@@ -6,6 +6,7 @@ const (
EventChat = "chat"
EventHealth = "health"
EventCron = "cron"
EventHeartbeat = "heartbeat"
EventExecApprovalReq = "exec.approval.requested"
EventExecApprovalRes = "exec.approval.resolved"
EventPresence = "presence"
+11
View File
@@ -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"