feat(heartbeat): cron wakeMode, queue-aware scheduling, lightContext

- CronPayload.WakeHeartbeat triggers heartbeat immediately after cron job completes
- Cron tool supports wake_heartbeat param on add/update actions
- Scheduler.HasActiveSessionsForAgent() detects busy agents for heartbeat skip
- RunRequest.LightContext skips loading context files during heartbeat runs
This commit is contained in:
viettranx committed 2026-03-18 13:11:58 +07:00
1 parent 08a2d95c0c
commit 29816db0ab
6 files changed
+45 -7

No files matched your search

+9
View File
@@ -17,6 +17,10 @@ import (
// makeCronJobHandler creates a cron job handler that routes through the scheduler's cron lane.
// This ensures per-session concurrency control (same job can't run concurrently)
// and integration with /stop, /stopall commands.
// cronHeartbeatWakeFn holds the heartbeat wake function, set after ticker creation.
// Safe because cron jobs only fire after Start(), well after this is set.
var cronHeartbeatWakeFn func(agentID string)
func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg *config.Config, channelMgr *channels.Manager) func(job *store.CronJob) (*store.CronJobResult, error) {
return func(job *store.CronJob) (*store.CronJobResult, error) {
agentID := job.AgentID
@@ -102,6 +106,11 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
cronResult.OutputTokens = result.Usage.CompletionTokens
}
// wakeMode: trigger heartbeat after cron job completes
if job.Payload.WakeHeartbeat && cronHeartbeatWakeFn != nil {
cronHeartbeatWakeFn(agentID)
}
return cronResult, nil
}
}
+1
View File
@@ -363,6 +363,7 @@ type RunRequest struct {
TraceName string // override trace name (default: "chat <agentID>")
TraceTags []string // additional tags for the trace (e.g. "cron")
MaxIterations int // per-request override (0 = use agent default, must be lower)
LightContext bool // skip loading context files (only inject ExtraSystemPrompt)
// Run classification
RunKind string // "delegation", "announce" — empty for user-initiated runs
+15
View File
@@ -3,6 +3,7 @@ package scheduler
import (
"context"
"log/slog"
"strings"
"sync"
"sync/atomic"
@@ -143,6 +144,20 @@ func (s *Scheduler) Stop() {
s.lanes.StopAll()
}
// HasActiveSessionsForAgent checks if any session queue for the given agent has active runs.
// Used by heartbeat ticker to skip agents currently processing other requests.
func (s *Scheduler) HasActiveSessionsForAgent(agentID string) bool {
prefix := "agent:" + agentID + ":"
s.mu.RLock()
defer s.mu.RUnlock()
for key, sq := range s.sessions {
if strings.HasPrefix(key, prefix) && sq.IsActive() {
return true
}
}
return false
}
// LaneStats returns utilization metrics for all lanes.
func (s *Scheduler) LaneStats() []LaneStats {
return s.lanes.AllStats()
+8 -6
View File
@@ -28,12 +28,13 @@ type CronSchedule struct {
// CronPayload describes what a job does when triggered.
type CronPayload struct {
Kind string `json:"kind"`
Message string `json:"message"`
Command string `json:"command,omitempty"`
Deliver bool `json:"deliver"`
Channel string `json:"channel,omitempty"`
To string `json:"to,omitempty"`
Kind string `json:"kind"`
Message string `json:"message"`
Command string `json:"command,omitempty"`
Deliver bool `json:"deliver"`
Channel string `json:"channel,omitempty"`
To string `json:"to,omitempty"`
WakeHeartbeat bool `json:"wake_heartbeat,omitempty"` // trigger heartbeat after job completes
}
// CronJobState tracks runtime state for a job.
@@ -75,6 +76,7 @@ type CronJobPatch struct {
Channel *string `json:"channel,omitempty"`
To *string `json:"to,omitempty"`
DeleteAfterRun *bool `json:"deleteAfterRun,omitempty"`
WakeHeartbeat *bool `json:"wakeHeartbeat,omitempty"`
}
// CronEvent represents a job lifecycle event sent to subscribers.
+4 -1
View File
@@ -148,7 +148,7 @@ func (s *PGCronStore) UpdateJob(jobID string, patch store.CronJobPatch) (*store.
}
// Update payload JSONB — fetch current, merge patch fields, re-serialize
needsPayloadUpdate := patch.Message != "" || patch.Deliver != nil || patch.Channel != nil || patch.To != nil
needsPayloadUpdate := patch.Message != "" || patch.Deliver != nil || patch.Channel != nil || patch.To != nil || patch.WakeHeartbeat != nil
if needsPayloadUpdate {
var payloadJSON []byte
if scanErr := s.db.QueryRow("SELECT payload FROM cron_jobs WHERE id = $1", id).Scan(&payloadJSON); scanErr == nil {
@@ -167,6 +167,9 @@ func (s *PGCronStore) UpdateJob(jobID string, patch store.CronJobPatch) (*store.
if patch.To != nil {
payload.To = *patch.To
}
if patch.WakeHeartbeat != nil {
payload.WakeHeartbeat = *patch.WakeHeartbeat
}
merged, _ := json.Marshal(payload)
updates["payload"] = merged
+8
View File
@@ -267,6 +267,14 @@ func (t *CronTool) handleAdd(ctx context.Context, args map[string]any, agentID,
return ErrorResult(fmt.Sprintf("failed to create cron job: %v", err))
}
// Set wake_heartbeat if requested (triggers heartbeat after cron job completes)
if wh, _ := jobObj["wake_heartbeat"].(bool); wh {
wakeTrue := true
if updated, uErr := t.cronStore.UpdateJob(job.ID, store.CronJobPatch{WakeHeartbeat: &wakeTrue}); uErr == nil {
job = updated
}
}
data, _ := json.MarshalIndent(map[string]any{"job": job}, "", " ")
return NewResult(string(data))
}