refactor(cortex): update bulletin, drift, prioritize, rl tasks

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)
+	// Get all active agents
-	agentID := "default" // TODO: Get from context or iterate
+	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",
+		"task":     "bulletin",
-		"tokens": bulletin.Tokens,
+		"agent_id": agentID,
+		"tokens":   bulletin.Tokens,
 	})
 
 	return nil

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,
+				"error":    err,
-				"domain": domain,
+				"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,
+				"error":    err,
-				"domain": domain,
+				"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
+}

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"})
+		logger.DebugCF("cortex", "No new items to prioritize",
-		return nil
+			map[string]interface{}{"task": "prioritize", "agent_id": agentID})
+		return 0, 0, nil
 	}
 
 	logger.DebugCF("cortex", "Processing items for actionability", map[string]interface{}{
-		"task":  "prioritize",
+		"task":     "prioritize",
-		"count": len(items),
+		"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,
+					"error":    err,
-					"item_id": item.ID,
+					"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.

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 {

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")