diff --git a/pkg/agent/agent_run.go b/pkg/agent/agent_run.go index c0ff95c31..20639205b 100644 --- a/pkg/agent/agent_run.go +++ b/pkg/agent/agent_run.go @@ -222,7 +222,18 @@ func (al *AgentLoop) loadSessionState(ctx context.Context, opts processOptions) summary = al.sessions.GetSummary(opts.SessionKey) } - return al.applyDAGCompression(ctx, opts.SessionKey, history), summary + // Zero-cost continuity (LCM ADR-002): skip DAG compression when context + // usage is below the soft compaction threshold. This eliminates overhead + // for ~80% of interactions where context is not under pressure. + softPct, _ := al.compactionThresholds() + softThreshold := al.contextWindow * softPct / 100 + tokenEstimate := al.estimateTokens(history) + if tokenEstimate <= softThreshold { + al.contextBuilder.SetContextTreeBlock("") + return history, summary + } + + return al.applyContextTreeSelection(ctx, opts.SessionKey, opts.UserMessage, history), summary } func (al *AgentLoop) buildPromptMessages(opts processOptions, history []messages.Message, summary string) []messages.Message { diff --git a/pkg/agent/context.go b/pkg/agent/context.go index 92ab75483..c695a90ed 100644 --- a/pkg/agent/context.go +++ b/pkg/agent/context.go @@ -29,7 +29,7 @@ type ContextBuilder struct { observationBlock string // Pre-rendered observation block for prompt injection focusBlock string // Pre-rendered active focus block knowledgeBlock string // Pre-rendered knowledge block from Focus completions - dagBlock string // Pre-rendered DAG compressed history + contextTreeBlock string // Pre-rendered Context-Tree selected history contextWindow int // Max tokens for context window (0 = no limit) cacheMu sync.Mutex @@ -99,9 +99,9 @@ func (cb *ContextBuilder) SetKnowledgeBlock(block string) { cb.knowledgeBlock = block } -// SetDAGBlock sets the pre-rendered DAG compressed history for prompt injection. -func (cb *ContextBuilder) SetDAGBlock(block string) { - cb.dagBlock = block +// SetContextTreeBlock sets the pre-rendered query-selected context history block for prompt injection. +func (cb *ContextBuilder) SetContextTreeBlock(block string) { + cb.contextTreeBlock = block } // SetContextWindow configures the token budget for the system prompt. @@ -226,9 +226,9 @@ Do NOT assume skill content — always load before applying. sections = append(sections, contextSection{"knowledge", cb.knowledgeBlock, 6}) } - // P7: DAG compressed history (lowest priority — can be reconstructed) - if cb.dagBlock != "" { - sections = append(sections, contextSection{"dag", "# Conversation History (Compressed)\n\n" + cb.dagBlock, 7}) + // P7: Context-Tree selected history (query-adaptive) + if cb.contextTreeBlock != "" { + sections = append(sections, contextSection{"context_tree", "# Conversation Context (Query-Selected)\n\n" + cb.contextTreeBlock, 7}) } // Proportional budget enforcement: each section gets a share of the @@ -287,7 +287,7 @@ var priorityWeight = [8]float64{ 4: 0.10, // observations 5: 0.08, // focus 6: 0.06, // knowledge - 7: 0.06, // DAG + 7: 0.06, // Context Tree } // applyProportionalBudget distributes tokens among sections using priority diff --git a/pkg/agent/fewshot.go b/pkg/agent/fewshot.go new file mode 100644 index 000000000..37e260caf --- /dev/null +++ b/pkg/agent/fewshot.go @@ -0,0 +1,235 @@ +package agent + +import ( + "context" + "fmt" + "sort" + "strings" + + "github.com/ZanzyTHEbar/dragonscale/pkg/logger" + "github.com/ZanzyTHEbar/dragonscale/pkg/memory/dag" +) + +// FewShotProvider provides few-shot examples for prompt injection. +type FewShotProvider struct { + store dag.AuditStore + formatter *dag.FewShotFormatter + maxExamples int +} + +// NewFewShotProvider creates a new few-shot provider. +func NewFewShotProvider(store dag.AuditStore, maxExamples int) *FewShotProvider { + return &FewShotProvider{ + store: store, + formatter: dag.NewFewShotFormatter(maxExamples, 5), + maxExamples: maxExamples, + } +} + +// GetExamples retrieves and formats few-shot examples for a given intent. +func (fsp *FewShotProvider) GetExamples(ctx context.Context, agentID, intent string) string { + chains, err := fsp.store.GetTopChains(ctx, agentID, intent, fsp.maxExamples) + if err != nil { + logger.DebugCF("agent", "Failed to get few-shot chains", map[string]interface{}{"error": err}) + return "" + } + + if len(chains) == 0 { + return "" + } + + return fsp.formatter.FormatForPrompt(chains, intent) +} + +// GetExamplesForQuery retrieves examples matching a natural language query. +func (fsp *FewShotProvider) GetExamplesForQuery(ctx context.Context, agentID, query string) string { + // Simple intent detection from query + intent := detectIntent(query) + return fsp.GetExamples(ctx, agentID, intent) +} + +// detectIntent performs simple keyword-based intent detection. +func detectIntent(query string) string { + queryLower := strings.ToLower(query) + + // Search-related + if strings.Contains(queryLower, "find") || + strings.Contains(queryLower, "search") || + strings.Contains(queryLower, "lookup") { + return "research" + } + + // File operations + if strings.Contains(queryLower, "file") || + strings.Contains(queryLower, "read") || + strings.Contains(queryLower, "write") || + strings.Contains(queryLower, "edit") { + return "file_research" + } + + // Code/development + if strings.Contains(queryLower, "code") || + strings.Contains(queryLower, "function") || + strings.Contains(queryLower, "test") || + strings.Contains(queryLower, "bug") || + strings.Contains(queryLower, "fix") { + return "development" + } + + // Git operations + if strings.Contains(queryLower, "git") || + strings.Contains(queryLower, "commit") || + strings.Contains(queryLower, "branch") || + strings.Contains(queryLower, "push") || + strings.Contains(queryLower, "pull") { + return "git_workflow" + } + + // Execution + if strings.Contains(queryLower, "run") || + strings.Contains(queryLower, "execute") || + strings.Contains(queryLower, "command") || + strings.Contains(queryLower, "shell") { + return "execution" + } + + return "" +} + +// BuildFewShotBlock creates a formatted few-shot block for prompt injection. +func BuildFewShotBlock(examples string) string { + if examples == "" { + return "" + } + + var b strings.Builder + b.WriteString("## Example Task Completion Patterns\n\n") + b.WriteString("When approaching similar tasks, consider these successful patterns:\n\n") + b.WriteString(examples) + b.WriteString("\nUse these patterns as guidance when completing your current task.\n") + + return b.String() +} + +// FewShotInjector handles injection of few-shot examples into prompts. +type FewShotInjector struct { + provider *FewShotProvider + enabled bool +} + +// NewFewShotInjector creates a few-shot injector. +func NewFewShotInjector(store dag.AuditStore, enabled bool, maxExamples int) *FewShotInjector { + return &FewShotInjector{ + provider: NewFewShotProvider(store, maxExamples), + enabled: enabled, + } +} + +// Inject adds few-shot examples to a system prompt based on the user's query. +func (fsi *FewShotInjector) Inject(ctx context.Context, agentID, query, systemPrompt string) string { + if !fsi.enabled { + return systemPrompt + } + examples := fsi.provider.GetExamplesForQuery(ctx, agentID, query) + return fsi.injectBlock(systemPrompt, examples) +} + +func (fsi *FewShotInjector) injectBlock(systemPrompt, examples string) string { + if examples == "" { + return systemPrompt + } + block := BuildFewShotBlock(examples) + if idx := strings.LastIndex(systemPrompt, "## Current Task"); idx > 0 { + return systemPrompt[:idx] + block + "\n\n" + systemPrompt[idx:] + } + return systemPrompt + "\n\n" + block +} + +// InjectByIntent adds few-shot examples for a specific intent category. +func (fsi *FewShotInjector) InjectByIntent(ctx context.Context, agentID, intent, systemPrompt string) string { + if !fsi.enabled { + return systemPrompt + } + examples := fsi.provider.GetExamples(ctx, agentID, intent) + return fsi.injectBlock(systemPrompt, examples) +} + +// InjectForTools adds examples showing successful usage of specific tools. +func (fsi *FewShotInjector) InjectForTools(ctx context.Context, agentID string, toolNames []string, systemPrompt string) string { + if !fsi.enabled || len(toolNames) == 0 { + return systemPrompt + } + examples := fsi.provider.GetExamples(ctx, agentID, "development") + return fsi.injectToolBlock(systemPrompt, toolNames, examples) +} + +func (fsi *FewShotInjector) injectToolBlock(systemPrompt string, toolNames []string, examples string) string { + if examples == "" { + return systemPrompt + } + var b strings.Builder + b.WriteString("## Tool Usage Patterns\n\n") + b.WriteString(fmt.Sprintf("When using %s, follow these patterns:\n\n", strings.Join(toolNames, ", "))) + b.WriteString(examples) + return systemPrompt + "\n\n" + b.String() +} + +// ChainSummary provides a summary of available action chains. +type ChainSummary struct { + TotalChains int + ByIntent map[string]int + ByCategory map[string]int + AverageScore float64 + TopIntents []string +} + +// GetSummary returns a summary of available chains for an agent. +func (fsp *FewShotProvider) GetSummary(ctx context.Context, agentID string) (*ChainSummary, error) { + // Get chains across all intents + chains, err := fsp.store.GetTopChains(ctx, agentID, "", 100) + if err != nil { + return nil, err + } + + summary := &ChainSummary{ + TotalChains: len(chains), + ByIntent: make(map[string]int), + ByCategory: make(map[string]int), + } + + var totalScore float64 + intentScores := make(map[string]float64) + + for _, chain := range chains { + summary.ByIntent[chain.Intent]++ + summary.ByCategory[chain.Category]++ + totalScore += chain.Score + intentScores[chain.Intent] += chain.Score + } + + if len(chains) > 0 { + summary.AverageScore = totalScore / float64(len(chains)) + } + + // Find top intents by count + type intentCount struct { + intent string + count int + score float64 + } + + var intents []intentCount + for intent, count := range summary.ByIntent { + intents = append(intents, intentCount{intent, count, intentScores[intent]}) + } + + sort.Slice(intents, func(i, j int) bool { + return intents[i].count > intents[j].count + }) + + for _, ic := range intents { + summary.TopIntents = append(summary.TopIntents, ic.intent) + } + + return summary, nil +} diff --git a/pkg/agent/integration_test.go b/pkg/agent/integration_test.go index 05f14aa98..667226c8a 100644 --- a/pkg/agent/integration_test.go +++ b/pkg/agent/integration_test.go @@ -402,31 +402,25 @@ func TestIntegration_Streaming_TextDeltas(t *testing.T) { ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) defer cancel() - // Collect stream deltas in background - var deltas []string - var deltaDone = make(chan struct{}) - go func() { - defer close(deltaDone) - for { - msg, ok := msgBus.SubscribeOutbound(ctx) - if !ok { - return - } - if msg.StreamDelta { - deltas = append(deltas, msg.Content) - } - } - }() - // Process with streaming response, err := al.ProcessDirectStreaming(ctx, "Stream me", "stream-session", "test", "chat-1") if err != nil { t.Fatalf("ProcessDirectStreaming failed: %v", err) } - // Cancel to stop delta collector - cancel() - <-deltaDone + // Drain outbound stream deltas after processing completes. + var deltas []string + for { + readCtx, readCancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + msg, ok := msgBus.SubscribeOutbound(readCtx) + readCancel() + if !ok { + break + } + if msg.StreamDelta { + deltas = append(deltas, msg.Content) + } + } // Verify complete response if response != "Hello from streaming agent response" { diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index a0188d7a7..d95cf4cff 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -21,6 +21,7 @@ import ( "github.com/ZanzyTHEbar/dragonscale/pkg/channels" "github.com/ZanzyTHEbar/dragonscale/pkg/config" "github.com/ZanzyTHEbar/dragonscale/pkg/constants" + "github.com/ZanzyTHEbar/dragonscale/pkg/cortex" "github.com/ZanzyTHEbar/dragonscale/pkg/ids" "github.com/ZanzyTHEbar/dragonscale/pkg/logger" "github.com/ZanzyTHEbar/dragonscale/pkg/memory" @@ -63,7 +64,7 @@ type AgentLoop struct { running atomic.Bool // Owner: loop.go — lifecycle gate controlled by Run/Stop only summarizing sync.Map // Owner: summarizer.go — intended for async summarization lockout, currently gated by TODO path summarizeFailures sync.Map // Owner: summarizer.go — write/read in forceCompression + summarizeSession error paths - dagCache sync.Map // Owner: summarizer.go — sessionKey → dagCacheEntry, skips recompression when compressible count unchanged + contextTreeCache sync.Map // Owner: summarizer.go — sessionKey → contextTreeCacheEntry keyed by query and history size auditChan chan *memory.AuditEntry // Buffered channel for async audit logging; drained by background worker auditDone chan struct{} // Closed when audit worker exits focusDirty sync.Map // sessionKey → struct{}: set by focus tool callbacks, cleared after context reload @@ -73,6 +74,7 @@ type AgentLoop struct { commandRegistry []SlashCommand outputOverride atomic.Value // Owner: command_handler.go — CLI output redirection target for internal messages toolResultSearch fantasy.AgentTool + cortex *cortex.Cortex } type outputTarget struct { @@ -389,11 +391,59 @@ func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBu al.SetupSecureBus(secretStore, securebus.DefaultBusConfig()) subagentManager.SetRunLoop(MakeUnifiedRunLoopFunc(al)) + // Initialize Cortex autonomous scheduler with foundation tasks. + // DecayStore and BackfillStore are satisfied by LibSQLDelegate via duck typing. + var decayStore cortex.DecayStore + if ds, ok := al.memDelegate.(cortex.DecayStore); ok { + decayStore = ds + } + var backfillStore cortex.BackfillStore + if bs, ok := al.memDelegate.(cortex.BackfillStore); ok { + backfillStore = bs + } + var embedFn func(ctx context.Context, text string) ([]float32, error) + if al.memoryStore != nil && al.memoryStore.Embedder() != nil { + embedder := al.memoryStore.Embedder() + embedFn = func(ctx context.Context, text string) ([]float32, error) { + return embedder.Embed(ctx, text) + } + } + // ConsolidationStore for memory graph maintenance + var consolidationStore cortex.ConsolidationStore + if cs, ok := al.memDelegate.(cortex.ConsolidationStore); ok { + consolidationStore = cs + } + // PruneStore for permanent deletion of quarantined items + var pruneStore cortex.PruneStore + if ps, ok := al.memDelegate.(cortex.PruneStore); ok { + pruneStore = ps + } + // RLStore for reinforcement learning weight updates + var rlStore cortex.RLStore + if rs, ok := al.memDelegate.(cortex.RLStore); ok { + rlStore = rs + } + // AuditAnalysisStore for audit log pattern detection + var auditStore cortex.AuditAnalysisStore + if aus, ok := al.memDelegate.(cortex.AuditAnalysisStore); ok { + auditStore = aus + } + cortexTasks := []cortex.Task{ + cortex.NewDecayTask(cortex.DefaultDecayConfig(), decayStore), + cortex.NewBackfillTask(cortex.DefaultBackfillConfig(), backfillStore, embedFn), + cortex.NewConsolidationTask(cortex.DefaultConsolidationConfig(), consolidationStore), + cortex.NewPruneTask(cortex.DefaultPruneConfig(), pruneStore), + cortex.NewRLTask(rlStore, pkg.NAME), + cortex.NewAuditAnalysisTask(auditStore), + } + al.cortex = cortex.New(cortexTasks, 60*time.Second) + return al, nil } func (al *AgentLoop) Run(ctx context.Context) error { al.running.Store(true) + go al.cortex.Start(ctx) for al.running.Load() { select { diff --git a/pkg/agent/offloading_tool_runtime.go b/pkg/agent/offloading_tool_runtime.go index d610392a6..4819891ba 100644 --- a/pkg/agent/offloading_tool_runtime.go +++ b/pkg/agent/offloading_tool_runtime.go @@ -116,9 +116,11 @@ func (r OffloadingToolRuntime) Execute(ctx context.Context, tools []fantasy.Agen } } - preview = truncateRunes(payloadText, threshold) - preview = strings.TrimSpace(preview) + "\n\n" + + // Type-aware summarization instead of blind truncation + summary := SummarizeContent(payloadText, threshold) + preview = summary.Summary + "\n\n" + "[TRUNCATED]\n" + + "[Type: " + string(summary.ContentType) + "]\n" + "- run_id: " + r.RunID.String() + "\n" + "- tool_call_id: " + tc.ToolCallID + "\n" + "- chunk_count: " + strconv.FormatInt(chunkCount, 10) + "\n" + diff --git a/pkg/agent/summarizer.go b/pkg/agent/summarizer.go index 332558534..da3a2c4c0 100644 --- a/pkg/agent/summarizer.go +++ b/pkg/agent/summarizer.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "sort" "strings" "time" @@ -12,6 +13,7 @@ import ( "github.com/ZanzyTHEbar/dragonscale/pkg/bus" "github.com/ZanzyTHEbar/dragonscale/pkg/config" "github.com/ZanzyTHEbar/dragonscale/pkg/constants" + "github.com/ZanzyTHEbar/dragonscale/pkg/contexttree" "github.com/ZanzyTHEbar/dragonscale/pkg/ids" "github.com/ZanzyTHEbar/dragonscale/pkg/logger" "github.com/ZanzyTHEbar/dragonscale/pkg/memory" @@ -19,18 +21,52 @@ import ( "github.com/ZanzyTHEbar/dragonscale/pkg/memory/observation" "github.com/ZanzyTHEbar/dragonscale/pkg/messages" "github.com/ZanzyTHEbar/dragonscale/pkg/tools" + "github.com/ZanzyTHEbar/dragonscale/pkg/utils" ) -// maybeSummarize only triggers emergency compression when hard limits are exceeded. -// Normal background compaction is intentionally disabled for the unified kernel. +// maybeSummarize implements LCM ADR-002 dual-threshold compaction. +// Below soft threshold: no-op (zero cost). +// Between soft and hard: async background compaction. +// Above hard: synchronous blocking compaction with 3-level escalation. func (al *AgentLoop) maybeSummarize(ctx context.Context, sessionKey, channel, chatID string) { newHistory := al.sessions.GetHistory(sessionKey) tokenEstimate := al.estimateTokens(newHistory) - criticalThreshold := al.contextWindow * 95 / 100 - if tokenEstimate > criticalThreshold { - al.forceCompression(ctx, sessionKey, channel, chatID) + softPct, hardPct := al.compactionThresholds() + softThreshold := al.contextWindow * softPct / 100 + hardThreshold := al.contextWindow * hardPct / 100 + + if tokenEstimate <= softThreshold { + return } + + if tokenEstimate > hardThreshold { + al.forceCompression(ctx, sessionKey, channel, chatID) + return + } + + go func() { + al.summarizeSession(context.WithoutCancel(ctx), sessionKey) + }() +} + +// compactionThresholds returns the soft and hard compaction percentages from config. +func (al *AgentLoop) compactionThresholds() (softPct, hardPct int) { + softPct = 70 + hardPct = 90 + if al.cfg != nil { + c := al.cfg.Agents.Defaults.Compaction + if c.SoftThresholdPct > 0 && c.SoftThresholdPct < 100 { + softPct = c.SoftThresholdPct + } + if c.HardThresholdPct > 0 && c.HardThresholdPct <= 100 { + hardPct = c.HardThresholdPct + } + } + if hardPct <= softPct { + hardPct = softPct + 10 + } + return softPct, hardPct } // EmergencyProvenance captures provenance metadata for postmortem when @@ -118,7 +154,7 @@ func (al *AgentLoop) forceCompression(ctx context.Context, sessionKey, channel, HistoryMsgCount: len(history), }) - al.summarizeSession(ctx, sessionKey) + al.summarizeSession(ctx, sessionKey, cycle) } } @@ -250,7 +286,12 @@ func (al *AgentLoop) persistOversizedRecoveryRefs(ctx context.Context, sessionKe } // summarizeSession summarizes the conversation history for a session. -func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey string) { +// An optional escalationLevel controls summarization aggressiveness (1=normal, 2=aggressive, 3=deterministic). +func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey string, escalationLevel ...int) { + level := 1 + if len(escalationLevel) > 0 && escalationLevel[0] > 0 { + level = escalationLevel[0] + } ctx, cancel := context.WithTimeout(parentCtx, 120*time.Second) defer cancel() @@ -304,8 +345,8 @@ func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey stri part1 := validMessages[:mid] part2 := validMessages[mid:] - s1, _ := al.summarizeBatch(ctx, part1, "") - s2, _ := al.summarizeBatch(ctx, part2, "") + s1, _ := al.summarizeBatchEscalated(ctx, part1, "", level) + s2, _ := al.summarizeBatchEscalated(ctx, part2, "", level) // Merge them mergePrompt := fmt.Sprintf("Merge these two conversation summaries into one cohesive summary:\n\n1: %s\n\n2: %s", s1, s2) @@ -316,7 +357,7 @@ func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey stri finalSummary = s1 + " " + s2 } } else { - finalSummary, _ = al.summarizeBatch(ctx, validMessages, summary) + finalSummary, _ = al.summarizeBatchEscalated(ctx, validMessages, summary, level) } if omitted && finalSummary != "" { @@ -393,6 +434,38 @@ func (al *AgentLoop) summarizeBatch(ctx context.Context, batch []messages.Messag return al.callModel(ctx, prompt.String()) } +// summarizeBatchEscalated applies escalation levels to summarization (LCM ADR-002). +// Level 1: normal summary (existing behavior) +// Level 2: aggressive bullet-point compression +// Level 3: deterministic truncation (no LLM call) +func (al *AgentLoop) summarizeBatchEscalated(ctx context.Context, batch []messages.Message, existingSummary string, level int) (string, error) { + switch level { + case 1: + return al.summarizeBatch(ctx, batch, existingSummary) + case 2: + var prompt strings.Builder + prompt.WriteString("Compress the following conversation into a VERY brief bullet-point list (max 5 bullets). Preserve only critical facts, decisions, and action items.\n") + if existingSummary != "" { + fmt.Fprintf(&prompt, "Prior context: %s\n", existingSummary) + } + prompt.WriteString("\nCONVERSATION:\n") + for _, m := range batch { + fmt.Fprintf(&prompt, "%s: %s\n", m.Role, m.Content) + } + return al.callModel(ctx, prompt.String()) + default: + if len(batch) == 0 { + return existingSummary, nil + } + first := batch[0] + last := batch[len(batch)-1] + return fmt.Sprintf("[Deterministic truncation of %d messages] First: %s: %s | Last: %s: %s", + len(batch), + first.Role, utils.Truncate(first.Content, 200), + last.Role, utils.Truncate(last.Content, 200)), nil + } +} + // callModel makes a direct call to the Fantasy LanguageModel (no tools, no agent loop). // Used for summarization and other simple generation tasks. func (al *AgentLoop) callModel(ctx context.Context, prompt string) (string, error) { @@ -423,136 +496,139 @@ func (al *AgentLoop) sessionsToMessagePairs(sessionKey string) []observation.Mes return pairs } -// dagCacheEntry holds the cached DAG compression output for a session, -// enabling skip of recompression and repersistence when the compressible -// portion hasn't grown since the last call. -type dagCacheEntry struct { - msgCount int - rendered string - dag *dag.DAG - persistFailed bool +// contextTreeCacheEntry caches rendered query-selected history blocks per session. +type contextTreeCacheEntry struct { + msgCount int + query string + rendered string } -// applyDAGCompression compresses old history into a DAG summary block and -// returns only the tail messages that should be passed as raw conversation. -// The compressed portion is injected into the system prompt via contextBuilder. -// When memDelegate implements dag.DAGPersister, the DAG is persisted for dag_expand/describe/grep. -// -// Incremental optimization: if the compressible message count matches the -// cached count, the previous DAG and rendered block are reused without -// recompression or repersistence. -func (al *AgentLoop) applyDAGCompression(ctx context.Context, sessionKey string, history []messages.Message) []messages.Message { - const minHistoryForDAG = 16 +// applyContextTreeSelection selects relevant historical context via query-adaptive +// Context Tree scoring and keeps only the raw tail as messages. +func (al *AgentLoop) applyContextTreeSelection(ctx context.Context, sessionKey, query string, history []messages.Message) []messages.Message { + const minHistoryForSelection = 16 - if len(history) < minHistoryForDAG { - al.contextBuilder.SetDAGBlock("") + if len(history) < minHistoryForSelection { + al.contextBuilder.SetContextTreeBlock("") return history } budget := dag.ComputeBudget(al.contextWindow, dag.DefaultBudgetConfig()) tailCount := dag.TailMessageCount(budget.RawTail) if tailCount >= len(history) { - al.contextBuilder.SetDAGBlock("") + al.contextBuilder.SetContextTreeBlock("") return history } - // Split: compress old, keep tail raw compressible := history[:len(history)-tailCount] tail := history[len(history)-tailCount:] - // Tool-call-aware: don't split on a "tool" message for len(tail) > 0 && tail[0].Role == "tool" && len(compressible) > 0 { tail = append([]messages.Message{compressible[len(compressible)-1]}, tail...) compressible = compressible[:len(compressible)-1] } if len(compressible) == 0 { - al.contextBuilder.SetDAGBlock("") + al.contextBuilder.SetContextTreeBlock("") return history } - // Check DAG cache: skip recompression if compressible count hasn't changed - if cached, ok := al.dagCache.Load(sessionKey); ok { - entry := cached.(dagCacheEntry) - if entry.msgCount == len(compressible) { - al.contextBuilder.SetDAGBlock(entry.rendered) - // Retry failed persistence from previous turn - if entry.persistFailed { - al.retryDAGPersist(ctx, sessionKey, entry) - } + cacheHit := false + if cached, ok := al.contextTreeCache.Load(sessionKey); ok { + entry := cached.(contextTreeCacheEntry) + if entry.msgCount == len(compressible) && entry.query == query { + al.contextBuilder.SetContextTreeBlock(entry.rendered) return tail } } - dagMsgs := make([]dag.Message, len(compressible)) - for i, m := range compressible { - dagMsgs[i] = dag.Message{Role: m.Role, Content: m.Content} + tree := contexttree.NewContextTree(contexttree.DefaultScoringConfig()) + rootID := tree.Root.ID + for _, m := range compressible { + tree.AddNode(rootID, contextNodeTypeForRole(m.Role), m.Content, nil, contexttree.ExtractTerms(m.Content)) } - compressor := dag.NewCompressor(dag.DefaultCompressorConfig()) - d := compressor.Compress(dagMsgs) + queryTerms := contexttree.ExtractTerms(query) + if len(queryTerms) == 0 && len(tail) > 0 { + queryTerms = contexttree.ExtractTerms(tail[len(tail)-1].Content) + } - rendered := dag.RenderDAGForBudget(d, budget.DAGSummaries) - al.contextBuilder.SetDAGBlock(rendered) + scores := tree.ScoreAll(nil, queryTerms) + nodes := make([]*contexttree.ContextNode, 0, len(tree.NodeIndex)-1) + for id, node := range tree.NodeIndex { + if node.Type == contexttree.NodeTypeRoot { + continue + } + node.TotalScore = scores[id] + nodes = append(nodes, node) + } - // Cache the result - al.dagCache.Store(sessionKey, dagCacheEntry{ - msgCount: len(compressible), - rendered: rendered, - dag: d, + sort.Slice(nodes, func(i, j int) bool { + if nodes[i].TotalScore == nodes[j].TotalScore { + return nodes[i].CreatedAt.After(nodes[j].CreatedAt) + } + return nodes[i].TotalScore > nodes[j].TotalScore }) - // Persist DAG for dag_expand, dag_describe, dag_grep (additive; in-memory behavior unchanged) - persistOK := al.tryDAGPersist(ctx, sessionKey, d, len(compressible)) - if !persistOK { - // Mark for retry on next cache hit - al.dagCache.Store(sessionKey, dagCacheEntry{ - msgCount: len(compressible), - rendered: rendered, - dag: d, - persistFailed: true, - }) + selectionBudget := budget.DAGSummaries + if selectionBudget <= 0 { + selectionBudget = 512 + } + selected := make([]*contexttree.ContextNode, 0, len(nodes)) + usedTokens := 0 + for _, node := range nodes { + nodeTokens := observation.EstimateTokens(node.Content) + if nodeTokens == 0 { + continue + } + if usedTokens+nodeTokens > selectionBudget { + continue + } + selected = append(selected, node) + usedTokens += nodeTokens } - logger.DebugCF("agent", "DAG compression applied", - map[string]interface{}{ - "total_msgs": len(history), - "compressed_msgs": len(compressible), - "tail_msgs": len(tail), - "dag_nodes": len(d.Nodes), - "cache_hit": false, - }) + rendered := renderContextTreeSelection(selected) + al.contextBuilder.SetContextTreeBlock(rendered) + al.contextTreeCache.Store(sessionKey, contextTreeCacheEntry{msgCount: len(compressible), query: query, rendered: rendered}) + + logger.DebugCF("agent", "Context-Tree selection applied", map[string]interface{}{ + "total_msgs": len(history), + "compressed_msgs": len(compressible), + "tail_msgs": len(tail), + "selected_nodes": len(selected), + "selected_tokens": usedTokens, + "cache_hit": cacheHit, + }) return tail } -func (al *AgentLoop) tryDAGPersist(ctx context.Context, sessionKey string, d *dag.DAG, msgCount int) bool { - dp, ok := al.memDelegate.(dag.DAGPersister) - if !ok { - return true +func contextNodeTypeForRole(role string) contexttree.NodeType { + switch role { + case "tool": + return contexttree.NodeTypeToolCall + case "assistant": + return contexttree.NodeTypeSummary + default: + return contexttree.NodeTypeMessage } - if err := dp.PersistDAG(ctx, pkg.NAME, sessionKey, &dag.PersistSnapshot{ - FromMsgIdx: 0, - ToMsgIdx: msgCount, - MsgCount: msgCount, - DAG: d, - }); err != nil { - logger.WarnCF("agent", "DAG persist failed (will retry next turn)", - map[string]interface{}{"error": err.Error(), "session_key": sessionKey}) - return false - } - return true } -func (al *AgentLoop) retryDAGPersist(ctx context.Context, sessionKey string, entry dagCacheEntry) { - if al.tryDAGPersist(ctx, sessionKey, entry.dag, entry.msgCount) { - al.dagCache.Store(sessionKey, dagCacheEntry{ - msgCount: entry.msgCount, - rendered: entry.rendered, - dag: entry.dag, - persistFailed: false, - }) +func renderContextTreeSelection(selected []*contexttree.ContextNode) string { + if len(selected) == 0 { + return "" } + var b strings.Builder + for _, node := range selected { + line := strings.ReplaceAll(node.Content, "\n", " ") + line = strings.TrimSpace(line) + if len(line) > 320 { + line = line[:317] + "..." + } + fmt.Fprintf(&b, "[%s score=%.3f] %s\n", node.Type, node.TotalScore, line) + } + return strings.TrimSpace(b.String()) } func (al *AgentLoop) estimateTokens(msgs []messages.Message) int { diff --git a/pkg/agent/summary.go b/pkg/agent/summary.go new file mode 100644 index 000000000..a1ffd6a01 --- /dev/null +++ b/pkg/agent/summary.go @@ -0,0 +1,577 @@ +package agent + +import ( + "encoding/csv" + "encoding/json" + "fmt" + "strings" +) + +// ContentType represents the detected type of content for summarization. +type ContentType string + +const ( + ContentTypeJSON ContentType = "json" + ContentTypeCode ContentType = "code" + ContentTypeCSV ContentType = "csv" + ContentTypeText ContentType = "text" + ContentTypeError ContentType = "error" +) + +// TypeSummary contains a type-aware summary of content. +type TypeSummary struct { + ContentType ContentType `json:"content_type"` + Summary string `json:"summary"` + Details any `json:"details,omitempty"` +} + +// SummarizeContent detects content type and generates an appropriate summary. +// This replaces blind truncation with meaningful previews based on content structure. +func SummarizeContent(content string, maxLen int) TypeSummary { + content = strings.TrimSpace(content) + if content == "" { + return TypeSummary{ContentType: ContentTypeText, Summary: "(empty)"} + } + + // Check for error patterns first (they can appear in any content type) + if isError(content) { + return summarizeError(content, maxLen) + } + + // Try JSON first (most specific) + if isJSON(content) { + return summarizeJSON(content, maxLen) + } + + // Try CSV + if isCSV(content) { + return summarizeCSV(content, maxLen) + } + + // Try code (heuristic-based) + if isCode(content) { + return summarizeCode(content, maxLen) + } + + // Default to text summarization + return summarizeText(content, maxLen) +} + +// isJSON checks if content appears to be valid JSON. +func isJSON(content string) bool { + trimmed := strings.TrimSpace(content) + if !strings.HasPrefix(trimmed, "{") && !strings.HasPrefix(trimmed, "[") { + return false + } + var js any + if err := json.Unmarshal([]byte(trimmed), &js); err != nil { + return false + } + return true +} + +// summarizeJSON creates a JSON schema summary (keys for objects, length for arrays). +func summarizeJSON(content string, maxLen int) TypeSummary { + trimmed := strings.TrimSpace(content) + + var data any + if err := json.Unmarshal([]byte(trimmed), &data); err != nil { + return TypeSummary{ContentType: ContentTypeJSON, Summary: truncateRunes(content, maxLen)} + } + + switch v := data.(type) { + case map[string]any: + keys := make([]string, 0, len(v)) + for k := range v { + keys = append(keys, k) + } + schema := fmt.Sprintf("JSON object with %d fields: %s", len(keys), strings.Join(keys, ", ")) + if len(schema) > maxLen { + schema = schema[:maxLen-3] + "..." + } + return TypeSummary{ + ContentType: ContentTypeJSON, + Summary: schema, + Details: map[string]any{"keys": keys, "type": "object", "field_count": len(v)}, + } + + case []any: + // Show first few items as sample + sampleCount := min(len(v), 3) + var sampleItems []string + for i := 0; i < sampleCount; i++ { + switch item := v[i].(type) { + case map[string]any: + itemKeys := make([]string, 0, len(item)) + for k := range item { + itemKeys = append(itemKeys, k) + } + sampleItems = append(sampleItems, fmt.Sprintf("{%d fields}", len(itemKeys))) + case string: + sampleItems = append(sampleItems, fmt.Sprintf("%q", truncateRunes(item, 30))) + default: + sampleItems = append(sampleItems, fmt.Sprintf("%v", item)) + } + } + + summary := fmt.Sprintf("JSON array with %d items", len(v)) + if len(sampleItems) > 0 { + summary += fmt.Sprintf(" [sample: %s]", strings.Join(sampleItems, ", ")) + } + summary = truncateSummary(summary, maxLen) + return TypeSummary{ + ContentType: ContentTypeJSON, + Summary: summary, + Details: map[string]any{"length": len(v), "type": "array", "sample_count": sampleCount}, + } + + default: + return TypeSummary{ContentType: ContentTypeJSON, Summary: fmt.Sprintf("JSON scalar: %v", v)} + } +} + +// isError checks if content contains error patterns like stack traces, +// exception messages, or failure indicators at the start of lines. +func isError(content string) bool { + errorPatterns := []string{ + "error:", "exception:", "stack trace", "panic:", "fail:", + "runtime error", "fatal:", "warning:", "syntax error", + "compilation error", "build failed", "test failed", + } + + lines := strings.Split(content, "\n") + errorCount := 0 + + for _, line := range lines { + trimmed := strings.TrimSpace(line) + lower := strings.ToLower(trimmed) + + // Check for error patterns at the start of lines + for _, pattern := range errorPatterns { + if strings.HasPrefix(lower, pattern) { + errorCount++ + break + } + } + + // Additional check for common error indicators anywhere in line + if strings.Contains(lower, "--- stack trace ---") || + strings.Contains(lower, "caused by:") || + strings.Contains(lower, "at line") || + strings.Contains(lower, "error code:") { + errorCount++ + } + } + + // Require at least 2 error indicators or 1 strong indicator + return errorCount >= 2 || + strings.Contains(strings.ToLower(content), "panic:") || + strings.Contains(strings.ToLower(content), "stack trace") +} + +// summarizeError creates a summary of error content, extracting key error messages. +func summarizeError(content string, maxLen int) TypeSummary { + lines := strings.Split(content, "\n") + + var errorMessages []string + var contextLines []string + + for _, line := range lines { + trimmed := strings.TrimSpace(line) + if trimmed == "" { + continue + } + lower := strings.ToLower(trimmed) + + // Capture error message lines + if strings.HasPrefix(lower, "error:") || + strings.HasPrefix(lower, "exception:") || + strings.HasPrefix(lower, "panic:") || + strings.HasPrefix(lower, "fail:") || + strings.HasPrefix(lower, "fatal:") || + strings.HasPrefix(lower, "warning:") { + errorMessages = append(errorMessages, trimmed) + } else if len(contextLines) < 3 && !strings.Contains(lower, "at ") { + // Capture first few non-stack-trace lines as context + contextLines = append(contextLines, trimmed) + } + } + + var summary string + if len(errorMessages) > 0 { + summary = "Error: " + errorMessages[0] + if len(errorMessages) > 1 { + summary += fmt.Sprintf(" (%d errors)", len(errorMessages)) + } + } else if len(contextLines) > 0 { + summary = "Error context: " + contextLines[0] + } else { + summary = "Error detected in content" + } + + // Include line count info + lineCount := len(lines) + summary = fmt.Sprintf("%s [%d lines]", summary, lineCount) + + summary = truncateSummary(summary, maxLen) + + return TypeSummary{ + ContentType: ContentTypeError, + Summary: summary, + Details: map[string]any{ + "error_count": len(errorMessages), + "line_count": lineCount, + "errors": errorMessages[:min(len(errorMessages), 5)], + "has_panic": strings.Contains(strings.ToLower(content), "panic:"), + "has_trace": strings.Contains(strings.ToLower(content), "stack trace"), + }, + } +} + +// isCSV checks if content appears to be CSV format. +// Uses multiple heuristics to reduce false positives on code with commas: +// - Header detection: first line should look like headers, not code +// - Numeric data check: subsequent lines should contain numeric values +// - Minimum 2 columns consistently across lines +func isCSV(content string) bool { + lines := strings.Split(strings.TrimSpace(content), "\n") + if len(lines) < 2 { + return false + } + + // Filter out empty lines for processing + var nonEmptyLines []string + for _, line := range lines { + if strings.TrimSpace(line) != "" { + nonEmptyLines = append(nonEmptyLines, line) + } + } + if len(nonEmptyLines) < 2 { + return false + } + + // Use at most 5 non-empty lines for analysis + checkLines := nonEmptyLines + if len(checkLines) > 5 { + checkLines = checkLines[:5] + } + + // Check for consistent comma count (minimum 2 columns = at least 1 comma) + commaCounts := make([]int, len(checkLines)) + for i, line := range checkLines { + commaCounts[i] = strings.Count(line, ",") + } + + // Require at least 2 columns (1 comma) in all lines + for _, count := range commaCounts { + if count < 1 { + return false + } + } + + // Check for consistent column count (allowing for trailing commas) + firstCount := commaCounts[0] + consistent := 0 + for _, count := range commaCounts[1:] { + if count == firstCount || count == firstCount-1 { + consistent++ + } + } + if consistent < len(commaCounts)-1 { + return false + } + + // Header detection: first line should look like headers + // Headers typically don't contain common code patterns + header := checkLines[0] + headerLower := strings.ToLower(header) + + // Reject headers that look like code statements + codePatterns := []string{ + "func ", "def ", "class ", "if ", "for ", "while ", "return", + "import ", "package ", "#include", "const ", "var ", "let ", + "public ", "private ", "protected ", "static ", "void ", + } + for _, pattern := range codePatterns { + if strings.HasPrefix(headerLower, pattern) { + return false + } + } + + // Headers typically have balanced alphanumeric content + // Count alphanumeric vs special chars in header fields + fields := strings.Split(header, ",") + if len(fields) < 2 { + return false + } + + headerLooksValid := true + for _, field := range fields { + trimmed := strings.TrimSpace(field) + if trimmed == "" { + continue + } + // Headers should contain mostly letters, not symbols + letterCount := 0 + symbolCount := 0 + for _, r := range trimmed { + if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') { + letterCount++ + } else if r == '(' || r == ')' || r == '{' || r == '}' || r == ';' || r == '=' { + symbolCount++ + } + } + // If field has more symbols than letters, likely code + if symbolCount > letterCount/2 { + headerLooksValid = false + break + } + } + if !headerLooksValid { + return false + } + + // Check data lines for CSV-like content (numeric values, quoted strings) + if len(checkLines) >= 2 { + dataLines := checkLines[1:] + numericLineCount := 0 + + for _, line := range dataLines { + dataFields := strings.Split(line, ",") + hasNumericField := false + for _, field := range dataFields { + trimmed := strings.TrimSpace(field) + if trimmed == "" { + continue + } + // Check if field looks like a number (integer, float, or quoted) + if (trimmed[0] >= '0' && trimmed[0] <= '9') || + strings.HasPrefix(trimmed, "\"") || + strings.HasPrefix(trimmed, "'") || + trimmed == "true" || trimmed == "false" || + trimmed == "null" || trimmed == "NA" { + hasNumericField = true + break + } + } + if hasNumericField { + numericLineCount++ + } + } + + // At least half of data lines should have CSV-like values + if numericLineCount < len(dataLines)/2 { + return false + } + } + + return true +} + +// summarizeCSV creates a CSV header and stats summary. +func summarizeCSV(content string, maxLen int) TypeSummary { + reader := csv.NewReader(strings.NewReader(content)) + reader.FieldsPerRecord = -1 // Allow variable + + // Read header + header, err := reader.Read() + if err != nil { + return TypeSummary{ContentType: ContentTypeCSV, Summary: truncateRunes(content, maxLen)} + } + + // Count rows + rowCount := 0 + for { + _, err := reader.Read() + if err != nil { + break + } + rowCount++ + } + + summary := fmt.Sprintf("CSV with %d columns: %s (%d data rows)", + len(header), strings.Join(header, ", "), rowCount) + + summary = truncateSummary(summary, maxLen) + + return TypeSummary{ + ContentType: ContentTypeCSV, + Summary: summary, + Details: map[string]any{"columns": header, "column_count": len(header), "row_count": rowCount}, + } +} + +// isCode uses heuristics to detect if content is source code. +func isCode(content string) bool { + codeIndicators := []string{ + "func ", "def ", "class ", "import ", "package ", + "#include", "public class", "private ", "function ", + "const ", "let ", "var ", "=> ", ":= ", + } + + lines := strings.Split(content, "\n") + indicatorCount := 0 + + for _, line := range lines { + trimmed := strings.TrimSpace(line) + for _, indicator := range codeIndicators { + if strings.HasPrefix(trimmed, indicator) { + indicatorCount++ + break + } + } + } + + // If more than 3 code-like lines, consider it code + return indicatorCount > 3 +} + +// summarizeCode creates a code structure summary (functions, classes, imports). +func summarizeCode(content string, maxLen int) TypeSummary { + lines := strings.Split(content, "\n") + + var functions, classes, imports []string + braceDepth := 0 + + for _, line := range lines { + trimmed := strings.TrimSpace(line) + + // Track brace depth for scope + for _, r := range trimmed { + switch r { + case '{', '(', '[': + braceDepth++ + case '}', ')', ']': + braceDepth-- + } + } + + // Top-level declarations + if braceDepth == 0 || braceDepth == 1 { + if strings.HasPrefix(trimmed, "func ") || strings.HasPrefix(trimmed, "def ") || + strings.HasPrefix(trimmed, "function ") { + parts := strings.Fields(trimmed) + if len(parts) >= 2 { + name := parts[1] + if idx := strings.Index(name, "("); idx > 0 { + name = name[:idx] + } + functions = append(functions, name) + } + } + if strings.HasPrefix(trimmed, "class ") || strings.HasPrefix(trimmed, "public class ") || + strings.HasPrefix(trimmed, "type ") && strings.Contains(trimmed, "struct") { + parts := strings.Fields(trimmed) + if len(parts) >= 2 { + name := parts[1] + if idx := strings.Index(name, "{"); idx > 0 { + name = name[:idx] + } + classes = append(classes, name) + } + } + if strings.HasPrefix(trimmed, "import ") || strings.HasPrefix(trimmed, "#include") || + strings.HasPrefix(trimmed, "package ") || strings.HasPrefix(trimmed, "using ") { + imports = append(imports, trimmed) + } + } + } + + summaryParts := []string{} + if len(functions) > 0 { + summaryParts = append(summaryParts, fmt.Sprintf("%d functions", len(functions))) + } + if len(classes) > 0 { + summaryParts = append(summaryParts, fmt.Sprintf("%d types/classes", len(classes))) + } + if len(imports) > 0 { + summaryParts = append(summaryParts, fmt.Sprintf("%d imports", len(imports))) + } + + summary := "Code: " + strings.Join(summaryParts, ", ") + if len(functions) > 0 { + funcList := strings.Join(functions[:min(len(functions), 5)], ", ") + if len(functions) > 5 { + funcList += "..." + } + summary += fmt.Sprintf(" [funcs: %s]", funcList) + } + + summary = truncateSummary(summary, maxLen) + + return TypeSummary{ + ContentType: ContentTypeCode, + Summary: summary, + Details: map[string]any{ + "functions": functions, + "classes": classes, + "imports": len(imports), + "function_count": len(functions), + }, + } +} + +// summarizeText creates a meaningful text excerpt (first + last parts). +func summarizeText(content string, maxLen int) TypeSummary { + lines := strings.Split(content, "\n") + + // Get first few non-empty lines + var firstLines []string + for _, line := range lines { + if strings.TrimSpace(line) != "" { + firstLines = append(firstLines, line) + if len(firstLines) >= 3 { + break + } + } + } + + // Get last few non-empty lines + var lastLines []string + for i := len(lines) - 1; i >= 0; i-- { + if strings.TrimSpace(lines[i]) != "" { + lastLines = append([]string{lines[i]}, lastLines...) + if len(lastLines) >= 2 { + break + } + } + } + + var summary string + if len(firstLines) > 0 { + summary = strings.Join(firstLines, "\n") + if len(lastLines) > 0 && len(lines) > len(firstLines)+len(lastLines) { + summary += "\n...\n" + strings.Join(lastLines, "\n") + } + } + + // Word count + wordCount := len(strings.Fields(content)) + summary = fmt.Sprintf("Text (%d words): %s", wordCount, summary) + + summary = truncateSummary(summary, maxLen) + + return TypeSummary{ + ContentType: ContentTypeText, + Summary: summary, + Details: map[string]any{"word_count": wordCount, "line_count": len(lines)}, + } +} + +func min(a, b int) int { + if a < b { + return a + } + return b +} + +// truncateSummary safely truncates a string to maxLen, adding "..." when there's room. +func truncateSummary(s string, maxLen int) string { + if len(s) <= maxLen { + return s + } + if maxLen > 3 { + return s[:maxLen-3] + "..." + } + return s[:maxLen] +} diff --git a/pkg/agent/task_completion.go b/pkg/agent/task_completion.go new file mode 100644 index 000000000..22f8d5579 --- /dev/null +++ b/pkg/agent/task_completion.go @@ -0,0 +1,70 @@ +package agent + +import ( + "context" + "time" + + "github.com/ZanzyTHEbar/dragonscale/pkg" + "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + "github.com/ZanzyTHEbar/dragonscale/pkg/logger" +) + +// TaskCompletion tracks the outcome of an agent run for RL analysis. +type TaskCompletion struct { + TaskID string + Description string + TokensUsed int + ToolCalls int + Errors int + UserCorrections int + Completed bool + SelfReports []MemoryRating // MemoryID + Score + CreatedAt time.Time +} + +// MemoryRating represents a self-reported usefulness score for a memory. +type MemoryRating struct { + MemoryID ids.UUID + Score int // 0-3 scale +} + +// TaskCompletionStore is the interface for storing task completion records. +// Implemented by the memory delegate. +type TaskCompletionStore interface { + StoreTaskCompletion(ctx context.Context, agentID string, completion TaskCompletion, conversationID, runID ids.UUID) error + GetCompletedTasks(ctx context.Context, agentID string, since time.Time) ([]TaskCompletion, error) + UpdateMemorySelfReport(ctx context.Context, memoryID ids.UUID, score int) error +} + +// endTask stores task completion data and self-reports. +// This is called at the end of an agent run to record the outcome for RL analysis. +func (al *AgentLoop) endTask(ctx context.Context, conversationID, runID ids.UUID, completion TaskCompletion) error { + if al.memDelegate == nil { + return nil + } + + // Store self-report scores if the delegate implements RLStore + if rlStore, ok := al.memDelegate.(interface { + UpdateMemorySelfReport(ctx context.Context, memoryID ids.UUID, score int) error + }); ok { + for _, rating := range completion.SelfReports { + if err := rlStore.UpdateMemorySelfReport(ctx, rating.MemoryID, rating.Score); err != nil { + // Log but don't fail - self-reports are best-effort + logger.WarnCF("agent", "Failed to store self-report", + map[string]interface{}{ + "memory_id": rating.MemoryID.String(), + "error": err.Error(), + }) + } + } + } + + // Store task completion record if the delegate implements TaskCompletionStore + if store, ok := al.memDelegate.(TaskCompletionStore); ok { + if err := store.StoreTaskCompletion(ctx, pkg.NAME, completion, conversationID, runID); err != nil { + return err + } + } + + return nil +}