diff --git a/pkg/agent/agent_run.go b/pkg/agent/agent_run.go index 5a183d53b..9f013df29 100644 --- a/pkg/agent/agent_run.go +++ b/pkg/agent/agent_run.go @@ -196,11 +196,13 @@ func (al *AgentLoop) refreshContextBlocks(ctx context.Context, opts processOptio _ = g.Wait() - al.ctxBlockCache.Store(opts.SessionKey, ctxBlockCacheEntry{ - focusBlock: focusBlock, - knowledge: kb, - cachedAt: time.Now(), - }) + if !useCached { + al.ctxBlockCache.Store(opts.SessionKey, ctxBlockCacheEntry{ + focusBlock: focusBlock, + knowledge: kb, + cachedAt: time.Now(), + }) + } // Prune stale entries from other sessions to prevent unbounded growth. al.ctxBlockCache.Range(func(key, value any) bool { @@ -616,10 +618,63 @@ func (al *AgentLoop) runStreaming(ctx context.Context, opts processOptions, ac a // auditStep logs tool calls from a Fantasy step result to the audit log. func (al *AgentLoop) auditStep(_ context.Context, step fantasy.StepResult, sessionKey string) { toolCalls := step.Content.ToolCalls() - if len(toolCalls) == 0 { + toolResults := step.Content.ToolResults() + if len(toolCalls) == 0 && len(toolResults) == 0 { return } + callByID := make(map[string]fantasy.ToolCallContent, len(toolCalls)) + for _, tc := range toolCalls { + callByID[tc.ToolCallID] = tc + } + + // Prefer result-based entries because they carry success/failure semantics. + if len(toolResults) > 0 { + for _, tr := range toolResults { + toolName := "" + toolInput := "" + if tc, ok := callByID[tr.ToolCallID]; ok { + toolName = tc.ToolName + toolInput = tc.Input + } + if toolName == "" { + toolName = "unknown_tool" + } + + action := "tool_success" + output := "" + switch out := tr.Result.(type) { + case fantasy.ToolResultOutputContentText: + output = out.Text + case fantasy.ToolResultOutputContentMedia: + output = out.Text + case fantasy.ToolResultOutputContentError: + action = "tool_error" + if out.Error != nil { + output = out.Error.Error() + } else { + output = "tool returned error output" + } + } + + entry := &memory.AuditEntry{ + ID: ids.New(), + AgentID: pkg.NAME, + SessionKey: sessionKey, + Action: action, + Target: toolName, + Input: toolInput, + Output: output, + } + if !al.enqueueAuditEntry(entry) { + logger.WarnCF("agent", "Audit channel unavailable, dropping tool result entry", + map[string]interface{}{"tool": toolName, "action": action}) + } + } + return + } + + // Legacy fallback: if no tool results were emitted, record tool calls. for _, tc := range toolCalls { entry := &memory.AuditEntry{ ID: ids.New(), @@ -629,15 +684,31 @@ func (al *AgentLoop) auditStep(_ context.Context, step fantasy.StepResult, sessi Target: tc.ToolName, Input: tc.Input, } - select { - case al.auditChan <- entry: - default: - logger.WarnCF("agent", "Audit channel full, dropping entry", + if !al.enqueueAuditEntry(entry) { + logger.WarnCF("agent", "Audit channel unavailable, dropping tool call entry", map[string]interface{}{"tool": tc.ToolName}) } } } +func (al *AgentLoop) enqueueAuditEntry(entry *memory.AuditEntry) (ok bool) { + if al.auditChan == nil || entry == nil { + return false + } + defer func() { + if recover() != nil { + ok = false + } + }() + + select { + case al.auditChan <- entry: + return true + default: + return false + } +} + // updateToolContexts updates the context for tools that need channel/chatID info. func (al *AgentLoop) updateToolContexts(channel, chatID string) { // Use ContextualTool interface instead of type assertions diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 2c565f6c7..99ef36af5 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -75,6 +75,7 @@ type AgentLoop struct { outputOverride atomic.Value // Owner: command_handler.go — CLI output redirection target for internal messages toolResultSearch fantasy.AgentTool cortex *cortex.Cortex + inflight sync.WaitGroup } type outputTarget struct { @@ -413,6 +414,11 @@ func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBu if aus, ok := al.memDelegate.(cortex.AuditAnalysisStore); ok { auditStore = aus } + // DriftStore for domain health monitoring and trend tracking + var driftStore cortex.DriftStore + if ds, ok := al.memDelegate.(cortex.DriftMemorySource); ok { + driftStore = cortex.NewMemoryDriftAdapter(ds) + } cortexTasks := []cortex.Task{ cortex.NewDecayTask(cortex.DefaultDecayConfig(), decayStore), cortex.NewBackfillTask(cortex.DefaultBackfillConfig(), backfillStore, embedFn), @@ -420,6 +426,7 @@ func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBu cortex.NewPruneTask(cortex.DefaultPruneConfig(), pruneStore), cortex.NewRLTask(rlStore, pkg.NAME), cortex.NewAuditAnalysisTask(auditStore), + cortex.NewDriftTask(cortex.DefaultDriftConfig(), driftStore), } al.cortex = cortex.New(cortexTasks, 60*time.Second) @@ -440,7 +447,11 @@ func (al *AgentLoop) Run(ctx context.Context) error { continue } - response, err := al.processMessage(ctx, msg) + al.inflight.Add(1) + response, err := func() (string, error) { + defer al.inflight.Done() + return al.processMessage(ctx, msg) + }() if err != nil { response = fmt.Sprintf("Error processing message: %v", err) } @@ -480,9 +491,15 @@ func (al *AgentLoop) Run(ctx context.Context) error { func (al *AgentLoop) Stop() { al.running.Store(false) - close(al.auditChan) - <-al.auditDone // wait for audit worker to drain - al.sessions.Close() + al.inflight.Wait() + + if al.auditChan != nil { + close(al.auditChan) + <-al.auditDone // wait for audit worker to drain + } + if al.sessions != nil { + al.sessions.Close() + } if al.identitySync != nil { al.identitySync.Close() }