From c79d6e1e26cc529cf9710d433c56573744c96194 Mon Sep 17 00:00:00 2001 From: ZanzyTHEbar Date: Thu, 5 Mar 2026 19:43:52 +0000 Subject: [PATCH] refactor(cortex): update bulletin, drift, prioritize, rl tasks --- pkg/cortex/tasks_bulletin.go | 41 +++++++++++++++++--- pkg/cortex/tasks_drift.go | 66 ++++++++++++++++++++++++++++----- pkg/cortex/tasks_prioritize.go | 68 ++++++++++++++++++++++++++++------ pkg/cortex/tasks_rl.go | 14 +++++-- pkg/cortex/tasks_rl_test.go | 6 ++- 5 files changed, 163 insertions(+), 32 deletions(-) diff --git a/pkg/cortex/tasks_bulletin.go b/pkg/cortex/tasks_bulletin.go index 2001ad75d..6d9e772cb 100644 --- a/pkg/cortex/tasks_bulletin.go +++ b/pkg/cortex/tasks_bulletin.go @@ -27,6 +27,9 @@ type BulletinStore interface { // GetLastBulletin retrieves the most recent bulletin GetLastBulletin(ctx context.Context, agentID string) (*DailyBulletin, error) + + // ListActiveAgents returns all agent IDs with recent activity + ListActiveAgents(ctx context.Context, since time.Time) ([]string, error) } // LLMClient provides LLM generation capabilities. @@ -97,17 +100,42 @@ func (t *BulletinTask) Interval() time.Duration { return t.cfg.GenerateInterval } -// Execute generates the daily bulletin. +// Execute generates the daily bulletin for all active agents. func (t *BulletinTask) Execute(ctx context.Context) error { logger.InfoCF("cortex", "Generating daily bulletin", map[string]interface{}{"task": "bulletin"}) - // For now, process a single agent (in production, iterate over all agents) - agentID := "default" // TODO: Get from context or iterate + // Get all active agents + since := time.Now().Add(-t.cfg.LookbackWindow) + agents, err := t.store.ListActiveAgents(ctx, since) + if err != nil { + logger.WarnCF("cortex", "Failed to list active agents, falling back to default", + map[string]interface{}{"error": err.Error()}) + agents = []string{"default"} + } + if len(agents) == 0 { + logger.DebugCF("cortex", "No active agents found, skipping bulletin generation", nil) + return nil + } + // Process bulletin for each agent + for _, agentID := range agents { + if err := t.processAgentBulletin(ctx, agentID); err != nil { + logger.WarnCF("cortex", "Failed to generate bulletin for agent", + map[string]interface{}{"agent_id": agentID, "error": err.Error()}) + // Continue with other agents - don't let one failure stop the batch + } + } + + return nil +} + +// processAgentBulletin generates and stores a bulletin for a single agent. +func (t *BulletinTask) processAgentBulletin(ctx context.Context, agentID string) error { // Check if we need to generate last, err := t.store.GetLastBulletin(ctx, agentID) if err == nil && last != nil && time.Since(last.GeneratedAt) < t.cfg.GenerateInterval { - logger.DebugCF("cortex", "Bulletin still fresh, skipping", map[string]interface{}{"last_generated": last.GeneratedAt}) + logger.DebugCF("cortex", "Bulletin still fresh, skipping", + map[string]interface{}{"agent_id": agentID, "last_generated": last.GeneratedAt}) return nil } @@ -123,8 +151,9 @@ func (t *BulletinTask) Execute(ctx context.Context) error { } logger.DebugCF("cortex", "Bulletin generated successfully", map[string]interface{}{ - "task": "bulletin", - "tokens": bulletin.Tokens, + "task": "bulletin", + "agent_id": agentID, + "tokens": bulletin.Tokens, }) return nil diff --git a/pkg/cortex/tasks_drift.go b/pkg/cortex/tasks_drift.go index d6bdac1fe..44f462652 100644 --- a/pkg/cortex/tasks_drift.go +++ b/pkg/cortex/tasks_drift.go @@ -60,6 +60,9 @@ type DriftStore interface { // GetDriftStatus retrieves the current drift status for a domain GetDriftStatus(ctx context.Context, agentID string, domain string) (*DomainDrift, error) + + // ListActiveAgents returns all agent IDs with recent activity + ListActiveAgents(ctx context.Context, since time.Time) ([]string, error) } // DomainActivity holds raw activity counts. @@ -129,20 +132,55 @@ func (t *DriftTask) Interval() time.Duration { return t.cfg.CheckInterval } -// Execute performs drift detection across all domains. +// Execute performs drift detection across all agents and their domains. func (t *DriftTask) Execute(ctx context.Context) error { logger.InfoCF("cortex", "Running drift detection", map[string]interface{}{"task": "drift"}) - agentID := "default" // TODO: Iterate over all agents + // Get all active agents + since := time.Now().Add(-t.cfg.ActivityWindow) + agents, err := t.store.ListActiveAgents(ctx, since) + if err != nil { + logger.WarnCF("cortex", "Failed to list active agents, falling back to default", + map[string]interface{}{"error": err.Error()}) + agents = []string{"default"} + } + if len(agents) == 0 { + logger.DebugCF("cortex", "No active agents found, skipping drift detection", nil) + return nil + } - // Get all domains + // Process drift detection for each agent + var totalDriftDetected int + for _, agentID := range agents { + count, err := t.processAgentDrift(ctx, agentID) + if err != nil { + logger.WarnCF("cortex", "Failed to process drift for agent", + map[string]interface{}{"agent_id": agentID, "error": err.Error()}) + // Continue with other agents + } + totalDriftDetected += count + } + + logger.DebugCF("cortex", "Drift detection complete", map[string]interface{}{ + "task": "drift", + "agents_checked": len(agents), + "drift_detected": totalDriftDetected, + }) + + return nil +} + +// processAgentDrift performs drift detection for a single agent. +func (t *DriftTask) processAgentDrift(ctx context.Context, agentID string) (int, error) { + // Get all domains for this agent domains, err := t.store.GetDomains(ctx, agentID) if err != nil { - return fmt.Errorf("get domains: %w", err) + return 0, fmt.Errorf("get domains for agent %s: %w", agentID, err) } logger.DebugCF("cortex", "Checking drift for domains", map[string]interface{}{ "task": "drift", + "agent_id": agentID, "domain_count": len(domains), }) @@ -151,16 +189,18 @@ func (t *DriftTask) Execute(ctx context.Context) error { drift, err := t.analyzeDomain(ctx, agentID, domain) if err != nil { logger.DebugCF("cortex", "Failed to analyze domain", map[string]interface{}{ - "error": err, - "domain": domain, + "error": err, + "agent_id": agentID, + "domain": domain, }) continue } if err := t.store.StoreDriftStatus(ctx, drift); err != nil { logger.DebugCF("cortex", "Failed to store drift status", map[string]interface{}{ - "error": err, - "domain": domain, + "error": err, + "agent_id": agentID, + "domain": domain, }) continue } @@ -177,13 +217,14 @@ func (t *DriftTask) Execute(ctx context.Context) error { } } - logger.DebugCF("cortex", "Drift detection complete", map[string]interface{}{ + logger.DebugCF("cortex", "Drift detection complete for agent", map[string]interface{}{ "task": "drift", + "agent_id": agentID, "domains_checked": len(domains), "drift_detected": driftDetected, }) - return nil + return driftDetected, nil } func (t *DriftTask) analyzeDomain(ctx context.Context, agentID, domain string) (*DomainDrift, error) { @@ -468,3 +509,8 @@ func (m *MemoryDriftAdapter) GetDriftStatus(ctx context.Context, agentID string, Score: 0.5, }, nil } + +func (m *MemoryDriftAdapter) ListActiveAgents(ctx context.Context, since time.Time) ([]string, error) { + // TODO: Query memory store for actual active agents + return []string{"default"}, nil +} diff --git a/pkg/cortex/tasks_prioritize.go b/pkg/cortex/tasks_prioritize.go index 6798cc20e..fe17252e9 100644 --- a/pkg/cortex/tasks_prioritize.go +++ b/pkg/cortex/tasks_prioritize.go @@ -29,6 +29,9 @@ type PrioritizeStore interface { // UpdateActionableStatus updates the status of an actionable item UpdateActionableStatus(ctx context.Context, itemID ids.UUID, status ActionableStatus) error + + // ListActiveAgents returns all agent IDs with recent activity + ListActiveAgents(ctx context.Context, since time.Time) ([]string, error) } // ActionableStatus represents the state of an actionable item. @@ -114,28 +117,66 @@ func (t *PrioritizeTask) Interval() time.Duration { return t.cfg.ProcessInterval } -// Execute performs prioritization of memory items. +// Execute performs prioritization of memory items across all agents. func (t *PrioritizeTask) Execute(ctx context.Context) error { logger.InfoCF("cortex", "Running prioritization scan", map[string]interface{}{"task": "prioritize"}) - agentID := "default" // TODO: Iterate over all agents + // Get all active agents + since := time.Now().Add(-t.cfg.LookbackWindow) + agents, err := t.store.ListActiveAgents(ctx, since) + if err != nil { + logger.WarnCF("cortex", "Failed to list active agents, falling back to default", + map[string]interface{}{"error": err.Error()}) + agents = []string{"default"} + } + if len(agents) == 0 { + logger.DebugCF("cortex", "No active agents found, skipping prioritization", nil) + return nil + } + // Process prioritization for each agent + var totalProcessed, totalExtracted int + for _, agentID := range agents { + processed, extracted, err := t.processAgentItems(ctx, agentID) + if err != nil { + logger.WarnCF("cortex", "Failed to process items for agent", + map[string]interface{}{"agent_id": agentID, "error": err.Error()}) + // Continue with other agents + } + totalProcessed += processed + totalExtracted += extracted + } + + logger.DebugCF("cortex", "Prioritization complete", map[string]interface{}{ + "task": "prioritize", + "agents_processed": len(agents), + "items_processed": totalProcessed, + "items_extracted": totalExtracted, + }) + + return nil +} + +// processAgentItems processes prioritization for a single agent. +func (t *PrioritizeTask) processAgentItems(ctx context.Context, agentID string) (int, int, error) { since := time.Now().Add(-t.cfg.LookbackWindow) // Get unprocessed items items, err := t.store.GetUnprocessedItems(ctx, agentID, since, t.cfg.MaxItemsPerRun) if err != nil { - return fmt.Errorf("get unprocessed items: %w", err) + return 0, 0, fmt.Errorf("get unprocessed items for agent %s: %w", agentID, err) } if len(items) == 0 { - logger.DebugCF("cortex", "No new items to prioritize", map[string]interface{}{"task": "prioritize"}) - return nil + logger.DebugCF("cortex", "No new items to prioritize", + map[string]interface{}{"task": "prioritize", "agent_id": agentID}) + return 0, 0, nil } logger.DebugCF("cortex", "Processing items for actionability", map[string]interface{}{ - "task": "prioritize", - "count": len(items), + "task": "prioritize", + "agent_id": agentID, + "count": len(items), }) // Process each item @@ -147,8 +188,9 @@ func (t *PrioritizeTask) Execute(ctx context.Context) error { if isActionable { if err := t.store.StoreActionableItem(ctx, actionable); err != nil { logger.DebugCF("cortex", "Failed to store actionable item", map[string]interface{}{ - "error": err, - "item_id": item.ID, + "error": err, + "item_id": item.ID, + "agent_id": agentID, }) continue } @@ -159,16 +201,18 @@ func (t *PrioritizeTask) Execute(ctx context.Context) error { // Mark items as processed if err := t.store.MarkAsProcessed(ctx, processed); err != nil { - logger.DebugCF("cortex", "Failed to mark items as processed", map[string]interface{}{"error": err}) + logger.DebugCF("cortex", "Failed to mark items as processed", + map[string]interface{}{"error": err, "agent_id": agentID}) } - logger.DebugCF("cortex", "Prioritization complete", map[string]interface{}{ + logger.DebugCF("cortex", "Prioritization complete for agent", map[string]interface{}{ "task": "prioritize", + "agent_id": agentID, "processed": len(processed), "extracted": extracted, }) - return nil + return len(processed), extracted, nil } // analyzeItem determines if a memory item contains an actionable task. diff --git a/pkg/cortex/tasks_rl.go b/pkg/cortex/tasks_rl.go index e79d3ee83..f61cc8a21 100644 --- a/pkg/cortex/tasks_rl.go +++ b/pkg/cortex/tasks_rl.go @@ -13,12 +13,13 @@ import ( // RLStore is the minimal interface for reinforcement learning weight updates. // Implemented by the memory delegate via hand-written SQL. type RLStore interface { - GetCompletedTasks(ctx context.Context, since time.Time) ([]TaskRecord, error) + GetCompletedTasks(ctx context.Context, agentID string, since time.Time) ([]TaskRecord, error) GetRetrievedMemories(ctx context.Context, taskID string) ([]RetrievedMemoryRecord, error) GetTaskBaseline(ctx context.Context, agentID string) (*TaskBaseline, error) UpdateTaskBaseline(ctx context.Context, agentID string, baseline *TaskBaseline) error UpdateMemoryWeight(ctx context.Context, memoryID ids.UUID, weight, credit float64) error UpdateMemorySelfReport(ctx context.Context, memoryID ids.UUID, score int) error + ListActiveAgents(ctx context.Context, since time.Time) ([]string, error) } // TaskRecord represents a completed task with performance metrics. @@ -84,7 +85,7 @@ func (t *RLTask) Execute(ctx context.Context) error { } // Get tasks completed since last run - tasks, err := t.store.GetCompletedTasks(ctx, t.lastRun) + tasks, err := t.store.GetCompletedTasks(ctx, t.agentID, t.lastRun) if err != nil { return fmt.Errorf("failed to get completed tasks: %w", err) } @@ -115,7 +116,14 @@ func (t *RLTask) Execute(ctx context.Context) error { // Get memory stats for logging memories, err := t.store.GetRetrievedMemories(ctx, task.ID) - if err == nil { + if err != nil { + logger.WarnCF("cortex", "Failed to get retrieved memories for RL processing", + map[string]interface{}{ + "task_id": task.ID, + "error": err.Error(), + }) + // Continue processing - don't let memory retrieval failure stop the batch + } else { totalMemoriesUpdated += len(memories) for _, m := range memories { if m.SelfReportScore != nil { diff --git a/pkg/cortex/tasks_rl_test.go b/pkg/cortex/tasks_rl_test.go index cbb92aff3..a49080434 100644 --- a/pkg/cortex/tasks_rl_test.go +++ b/pkg/cortex/tasks_rl_test.go @@ -46,7 +46,7 @@ func (m *mockRLStore) UpdateTaskBaseline(ctx context.Context, agentID string, ba return nil } -func (m *mockRLStore) GetCompletedTasks(ctx context.Context, since time.Time) ([]TaskRecord, error) { +func (m *mockRLStore) GetCompletedTasks(ctx context.Context, agentID string, since time.Time) ([]TaskRecord, error) { if m.getCompletedTasksErr != nil { return nil, m.getCompletedTasksErr } @@ -83,6 +83,10 @@ func (m *mockRLStore) UpdateMemorySelfReport(ctx context.Context, memoryID ids.U return nil } +func (m *mockRLStore) ListActiveAgents(ctx context.Context, since time.Time) ([]string, error) { + return []string{"test-agent"}, nil +} + func TestRLTask_Name(t *testing.T) { store := &mockRLStore{} task := NewRLTask(store, "test-agent")