fix(evolution): fix 7 bugs in evolution flow — guardrails, cron, dedup, dead types

- Remove broken delta comparison in CheckGuardrails (compared usage rate vs max delta)
- Replace hardcoded +0.05 threshold bump with configurable MaxDeltaPerCycle, cap at 0.95
- Remove dead MetricFeedback and SuggestMemoryPrune constants + UI references
- Make SuggestToolOrder approval actionable — disables tool via BuiltinToolTenantConfigStore
- Replace time.Ticker(24h) with wall-clock scheduling (3 AM daily, Sundays weekly eval)
- Add PG advisory lock on pinned connection for multi-instance safety
- Add EvolutionSuggest flag check in cron (metrics-only agents skip suggestion analysis)
- Change suggestion dedup from type-only to (type, metric_key) composite key
This commit is contained in:
viettranx committed 2026-04-10 10:13:58 +07:00
1 parent 4b7a2e6e58
commit 6d43de4169
10 files changed
+177 -67

No files matched your search

+80 -15
View File
@@ -2,6 +2,7 @@ package cmd
import (
"context"
"database/sql"
"log/slog"
"time"
@@ -9,36 +10,94 @@ import (
"github.com/nextlevelbuilder/goclaw/internal/store"
)
// runEvolutionCron runs the v3 evolution suggestion engine (daily) and
// evaluation/rollback check (weekly) as background goroutines.
// evolutionCronLockID is a PG advisory lock ID for preventing duplicate cron runs.
// Only one gateway instance should run evolution analysis at a time.
const evolutionCronLockID int64 = 0x65766F6C // "evol"
// runEvolutionCron runs the v3 evolution suggestion engine (daily at 3:00 AM)
// and evaluation/rollback check (weekly on Sundays at 3:00 AM) as a background goroutine.
// Designed to be called with `go runEvolutionCron(...)`.
func runEvolutionCron(stores *store.Stores, engine *agent.SuggestionEngine) {
dailyTicker := time.NewTicker(24 * time.Hour)
defer dailyTicker.Stop()
weeklyTicker := time.NewTicker(7 * 24 * time.Hour)
defer weeklyTicker.Stop()
// Run first analysis 1 minute after startup (warm-up).
// Wait 1 minute after startup for warm-up, then run first analysis.
time.Sleep(1 * time.Minute)
runSuggestionAnalysis(stores, engine)
for {
select {
case <-dailyTicker.C:
runSuggestionAnalysis(stores, engine)
case <-weeklyTicker.C:
nextDaily := nextDailyRun(3) // 3:00 AM server time
timer := time.NewTimer(time.Until(nextDaily))
<-timer.C
timer.Stop()
runSuggestionAnalysis(stores, engine)
// Weekly evaluation on Sundays.
if time.Now().Weekday() == time.Sunday {
runEvolutionEvaluation(stores)
}
}
}
// runSuggestionAnalysis lists agents with evolution metrics enabled and runs analysis.
// nextDailyRun returns the next occurrence of the given hour (0-23) today or tomorrow.
// Uses AddDate to handle DST transitions correctly (calendar day, not 24h).
func nextDailyRun(hour int) time.Time {
now := time.Now()
next := time.Date(now.Year(), now.Month(), now.Day(), hour, 0, 0, 0, now.Location())
if !now.Before(next) {
next = next.AddDate(0, 0, 1) // calendar day, not 24h — DST-safe
}
return next
}
// tryAdvisoryLock acquires a PG session-level advisory lock on a pinned connection.
// Returns the pinned connection (caller must close) and whether the lock was acquired.
// No-op (returns nil, true) when db is nil (SQLite/desktop edition runs single-instance).
func tryAdvisoryLock(ctx context.Context, db *sql.DB) (*sql.Conn, bool) {
if db == nil {
return nil, true
}
conn, err := db.Conn(ctx)
if err != nil {
slog.Debug("evolution.cron.conn_failed", "error", err)
return nil, false
}
var acquired bool
if err := conn.QueryRowContext(ctx, "SELECT pg_try_advisory_lock($1)", evolutionCronLockID).Scan(&acquired); err != nil {
conn.Close()
slog.Debug("evolution.cron.advisory_lock_failed", "error", err)
return nil, false
}
if !acquired {
conn.Close()
return nil, false
}
return conn, true
}
// releaseAdvisoryLock releases a PG advisory lock on the pinned connection and closes it.
func releaseAdvisoryLock(ctx context.Context, conn *sql.Conn) {
if conn == nil {
return
}
_, _ = conn.ExecContext(ctx, "SELECT pg_advisory_unlock($1)", evolutionCronLockID)
conn.Close()
}
// runSuggestionAnalysis lists agents with evolution enabled and runs analysis.
// Note: List(ctx, "") uses bare context (no tenant) to list ALL agents cross-tenant.
// Per-agent calls are scoped via WithTenantID.
func runSuggestionAnalysis(stores *store.Stores, engine *agent.SuggestionEngine) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
// List all agents (empty ownerID = all, tenant-scoped via context).
// Acquire advisory lock on a pinned connection to prevent duplicate runs across instances.
conn, acquired := tryAdvisoryLock(ctx, stores.DB)
if !acquired {
slog.Debug("evolution.cron.skipped_lock_held")
return
}
defer releaseAdvisoryLock(ctx, conn)
agents, err := stores.Agents.List(ctx, "")
if err != nil {
slog.Warn("evolution.cron.list_agents_failed", "error", err)
@@ -54,6 +113,9 @@ func runSuggestionAnalysis(stores *store.Stores, engine *agent.SuggestionEngine)
if !flags.EvolutionMetrics {
continue
}
if !flags.EvolutionSuggest {
continue // metrics enabled but suggestions disabled — skip analysis
}
agentCtx := store.WithTenantID(ctx, ag.TenantID)
if _, err := engine.Analyze(agentCtx, ag.ID); err != nil {
slog.Debug("evolution.cron.analyze_failed", "agent", ag.ID, "error", err)
@@ -86,6 +148,9 @@ func runEvolutionEvaluation(stores *store.Stores) {
if !flags.EvolutionMetrics {
continue
}
if !flags.EvolutionSuggest {
continue // skip evaluation for agents with suggestions disabled
}
agentCtx := store.WithTenantID(ctx, ag.TenantID)
if err := agent.EvaluateApplied(agentCtx, ag.ID, guardrails, stores.EvolutionMetrics, stores.EvolutionSuggestions, stores.Agents); err != nil {
slog.Debug("evolution.cron.eval_failed", "agent", ag.ID, "error", err)
+3
View File
@@ -167,6 +167,9 @@ func (d *gatewayDeps) wireHTTPHandlersOnServer(
if d.pgStores.Agents != nil {
evoOpts = append(evoOpts, httpapi.WithAgentStore(d.pgStores.Agents))
}
if d.pgStores.BuiltinToolTenantCfgs != nil {
evoOpts = append(evoOpts, httpapi.WithToolTenantCfgs(d.pgStores.BuiltinToolTenantCfgs))
}
d.server.SetEvolutionHandler(httpapi.NewEvolutionHandler(d.pgStores.EvolutionMetrics, d.pgStores.EvolutionSuggestions, evoOpts...))
}
+13 -20
View File
@@ -6,7 +6,6 @@ import (
"fmt"
"log/slog"
"maps"
"math"
"slices"
"github.com/google/uuid"
@@ -55,22 +54,8 @@ func CheckGuardrails(g AdaptationGuardrails, sg store.EvolutionSuggestion, dataP
}
}
// Check delta constraint for threshold-type suggestions.
if sg.SuggestionType == store.SuggestThreshold {
maxDelta := g.MaxDeltaPerCycle
if maxDelta <= 0 {
maxDelta = 0.1
}
var params map[string]any
if err := json.Unmarshal(sg.Parameters, &params); err == nil {
if currentRate, ok := params["current_usage_rate"].(float64); ok {
// Proposed change is bounded by max delta.
if math.Abs(currentRate) > maxDelta {
return fmt.Errorf("delta %.2f exceeds max %.2f per cycle", currentRate, maxDelta)
}
}
}
}
// Threshold suggestions: the applied delta is capped at MaxDeltaPerCycle in ApplySuggestion,
// so no additional delta check is needed here. MinDataPoints and LockedParams already guard safety.
return nil
}
@@ -78,7 +63,7 @@ func CheckGuardrails(g AdaptationGuardrails, sg store.EvolutionSuggestion, dataP
// ApplySuggestion applies an approved suggestion's parameters to the agent's other_config JSONB.
// Stores the previous values in the suggestion's parameters for rollback.
// Scope: only retrieval-related parameters. Never security or core settings.
func ApplySuggestion(ctx context.Context, agentStore store.AgentStore, sugStore store.EvolutionSuggestionStore, sg store.EvolutionSuggestion) error {
func ApplySuggestion(ctx context.Context, agentStore store.AgentStore, sugStore store.EvolutionSuggestionStore, sg store.EvolutionSuggestion, guardrails AdaptationGuardrails) error {
// Load current agent config.
agent, err := agentStore.GetByID(ctx, sg.AgentID)
if err != nil {
@@ -102,12 +87,20 @@ func ApplySuggestion(ctx context.Context, agentStore store.AgentStore, sugStore
// Apply suggestion-specific changes based on type.
switch sg.SuggestionType {
case store.SuggestThreshold:
// Raise retrieval threshold slightly (bounded by guardrails).
// Raise retrieval threshold by MaxDeltaPerCycle (bounded by guardrails).
current, _ := otherConfig["retrieval_threshold"].(float64)
if current == 0 {
current = 0.3 // default threshold
}
otherConfig["retrieval_threshold"] = current + 0.05 // conservative bump
delta := guardrails.MaxDeltaPerCycle
if delta <= 0 {
delta = 0.05 // fallback
}
newThreshold := current + delta
if newThreshold > 0.95 {
newThreshold = 0.95
}
otherConfig["retrieval_threshold"] = newThreshold
default:
// Non-threshold suggestions are informational only, no auto-apply.
return fmt.Errorf("suggestion type %q does not support auto-apply", sg.SuggestionType)
+12 -17
View File
@@ -2,6 +2,7 @@ package agent
import (
"encoding/json"
"strings"
"testing"
"github.com/nextlevelbuilder/goclaw/internal/store"
@@ -51,7 +52,7 @@ func TestCheckGuardrails(t *testing.T) {
dataPoints: 200,
params: map[string]any{"source": "mem"},
sgType: store.SuggestThreshold,
wantErr: `parameter "source" is locked`,
wantErr: "parameter \"source\" is locked",
},
{
name: "no locked params passes",
@@ -67,6 +68,14 @@ func TestCheckGuardrails(t *testing.T) {
sgType: store.SuggestThreshold,
wantErr: "insufficient data",
},
{
name: "threshold with usage rate passes (no broken delta check)",
guardrails: DefaultGuardrails(),
dataPoints: 200,
params: map[string]any{"current_usage_rate": 0.15, "source": "vault"},
sgType: store.SuggestThreshold,
// Previously this would fail with "delta 0.15 exceeds max 0.10" — now passes.
},
}
for _, tt := range tests {
@@ -84,24 +93,10 @@ func TestCheckGuardrails(t *testing.T) {
t.Errorf("expected error containing %q, got nil", tt.wantErr)
}
if tt.wantErr != "" && err != nil {
if got := err.Error(); !contains(got, tt.wantErr) {
t.Errorf("error %q does not contain %q", got, tt.wantErr)
if !strings.Contains(err.Error(), tt.wantErr) {
t.Errorf("error %q does not contain %q", err.Error(), tt.wantErr)
}
}
})
}
}
func contains(s, substr string) bool {
return len(s) >= len(substr) && (s == substr || len(substr) == 0 ||
(len(s) > 0 && len(substr) > 0 && searchStr(s, substr)))
}
func searchStr(s, sub string) bool {
for i := 0; i <= len(s)-len(sub); i++ {
if s[i:i+len(sub)] == sub {
return true
}
}
return false
}
+32 -6
View File
@@ -45,6 +45,30 @@ func NewSuggestionEngine(metrics store.EvolutionMetricsStore, suggestions store.
}
}
// dedupKey uniquely identifies a suggestion by type + metric key (e.g., tool name or source).
type dedupKey struct {
Type store.SuggestionType
Key string
}
// extractMetricKey returns the distinguishing key from suggestion parameters.
func extractMetricKey(params json.RawMessage, st store.SuggestionType) string {
var p map[string]any
if json.Unmarshal(params, &p) != nil {
return ""
}
switch st {
case store.SuggestThreshold:
s, _ := p["source"].(string)
return s
case store.SuggestToolOrder, store.SuggestSkillAdd:
s, _ := p["tool"].(string)
return s
default:
return ""
}
}
// Analyze runs all rules against a single agent's metrics (7-day window).
// Returns newly created suggestions. Skips rules that produce duplicates.
func (e *SuggestionEngine) Analyze(ctx context.Context, agentID uuid.UUID) ([]store.EvolutionSuggestion, error) {
@@ -65,11 +89,12 @@ func (e *SuggestionEngine) Analyze(ctx context.Context, agentID uuid.UUID) ([]st
Since: since,
}
// Load existing pending suggestions to avoid duplicates.
// Load existing pending suggestions to avoid duplicates (composite key: type + metric key).
existing, _ := e.suggestions.ListSuggestions(ctx, agentID, "pending", 100)
existingTypes := make(map[store.SuggestionType]bool, len(existing))
existingKeys := make(map[dedupKey]bool, len(existing))
for _, sg := range existing {
existingTypes[sg.SuggestionType] = true
mk := extractMetricKey(sg.Parameters, sg.SuggestionType)
existingKeys[dedupKey{sg.SuggestionType, mk}] = true
}
var created []store.EvolutionSuggestion
@@ -82,8 +107,9 @@ func (e *SuggestionEngine) Analyze(ctx context.Context, agentID uuid.UUID) ([]st
if sg == nil {
continue
}
// Skip if pending suggestion of same type already exists.
if existingTypes[sg.SuggestionType] {
// Skip if pending suggestion with same type + metric key already exists.
mk := extractMetricKey(sg.Parameters, sg.SuggestionType)
if existingKeys[dedupKey{sg.SuggestionType, mk}] {
continue
}
@@ -95,7 +121,7 @@ func (e *SuggestionEngine) Analyze(ctx context.Context, agentID uuid.UUID) ([]st
continue
}
created = append(created, *sg)
existingTypes[sg.SuggestionType] = true
existingKeys[dedupKey{sg.SuggestionType, mk}] = true
}
return created, nil
+33 -1
View File
@@ -1,6 +1,7 @@
package http
import (
"encoding/json"
"log/slog"
"net/http"
"strconv"
@@ -26,6 +27,9 @@ type EvolutionHandler struct {
// Optional: agent store for applying threshold suggestions.
agentStore store.AgentStore
// Optional: tenant tool config for disabling tools on SuggestToolOrder approval.
toolTenantCfgs store.BuiltinToolTenantConfigStore
}
// EvolutionHandlerOpt configures optional EvolutionHandler dependencies.
@@ -45,6 +49,11 @@ func WithAgentStore(as store.AgentStore) EvolutionHandlerOpt {
return func(h *EvolutionHandler) { h.agentStore = as }
}
// WithToolTenantCfgs enables tool disabling on SuggestToolOrder approval.
func WithToolTenantCfgs(tc store.BuiltinToolTenantConfigStore) EvolutionHandlerOpt {
return func(h *EvolutionHandler) { h.toolTenantCfgs = tc }
}
func NewEvolutionHandler(m store.EvolutionMetricsStore, s store.EvolutionSuggestionStore, opts ...EvolutionHandlerOpt) *EvolutionHandler {
h := &EvolutionHandler{metrics: m, suggestions: s}
for _, opt := range opts {
@@ -222,6 +231,29 @@ func (h *EvolutionHandler) handleUpdateSuggestion(w http.ResponseWriter, r *http
writeJSON(w, http.StatusOK, map[string]string{"status": "ok", "action": "skill_created"})
return
case store.SuggestToolOrder:
action := "tool_order_approved"
if h.toolTenantCfgs != nil {
// Extract tool name from suggestion parameters.
var params map[string]any
if err := json.Unmarshal(existing.Parameters, &params); err == nil {
if toolName, _ := params["tool"].(string); toolName != "" {
// Disable tool at tenant level using existing infrastructure.
if err := h.toolTenantCfgs.Set(r.Context(), existing.TenantID, toolName, false); err != nil {
slog.Warn("evolution.tool_order.disable_failed", "tool", toolName, "error", err)
} else {
action = "tool_disabled"
}
}
}
}
if err := h.suggestions.UpdateSuggestionStatus(r.Context(), suggestionID, "applied", reviewedBy); err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
return
}
writeJSON(w, http.StatusOK, map[string]string{"status": "ok", "action": action})
return
case store.SuggestThreshold:
if h.agentStore == nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "threshold auto-apply not available"})
@@ -240,7 +272,7 @@ func (h *EvolutionHandler) handleUpdateSuggestion(w http.ResponseWriter, r *http
writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
return
}
if err := agent.ApplySuggestion(r.Context(), h.agentStore, h.suggestions, *existing); err != nil {
if err := agent.ApplySuggestion(r.Context(), h.agentStore, h.suggestions, *existing, guardrails); err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
return
}
+3 -5
View File
@@ -14,7 +14,6 @@ type MetricType string
const (
MetricRetrieval MetricType = "retrieval"
MetricTool MetricType = "tool"
MetricFeedback MetricType = "feedback"
)
// EvolutionMetric is a single recorded metric data point.
@@ -58,10 +57,9 @@ type EvolutionMetricsStore interface {
type SuggestionType string
const (
SuggestThreshold SuggestionType = "threshold"
SuggestToolOrder SuggestionType = "tool_order"
SuggestSkillAdd SuggestionType = "skill_add"
SuggestMemoryPrune SuggestionType = "memory_prune"
SuggestThreshold SuggestionType = "threshold"
SuggestToolOrder SuggestionType = "tool_order"
SuggestSkillAdd SuggestionType = "skill_add"
)
// EvolutionSuggestion is a data-driven suggestion for agent improvement.
@@ -14,7 +14,6 @@ const TYPE_COLORS: Record<string, string> = {
threshold: 'bg-blue-500/10 text-blue-600',
tool_order: 'bg-orange-500/10 text-orange-600',
skill_add: 'bg-green-500/10 text-green-600',
memory_prune: 'bg-surface-tertiary text-text-muted',
}
const STATUS_COLORS: Record<string, string> = {
@@ -14,7 +14,6 @@ const TYPE_COLORS: Record<string, string> = {
threshold: "bg-blue-100 text-blue-700 dark:bg-blue-900 dark:text-blue-300",
tool_order: "bg-orange-100 text-orange-700 dark:bg-orange-900 dark:text-orange-300",
skill_add: "bg-green-100 text-green-700 dark:bg-green-900 dark:text-green-300",
memory_prune: "bg-gray-100 text-gray-700 dark:bg-gray-800 dark:text-gray-300",
};
const STATUS_COLORS: Record<string, string> = {
+1 -1
View File
@@ -18,7 +18,7 @@ export interface RetrievalAggregate {
export interface EvolutionSuggestion {
id: string;
agent_id: string;
suggestion_type: "threshold" | "tool_order" | "skill_add" | "memory_prune";
suggestion_type: "threshold" | "tool_order" | "skill_add";
suggestion: string;
rationale: string;
parameters: Record<string, unknown> | null;