refactor(agent): update context, loop, summarizer and tests

- Fix estimateTokens test expectations for per-message overhead
- Update agent_run, context, loop, summarizer, integration_test
- Note: TestContinuityKeepCount_UsesConfiguredPolicy has policy/ratio mismatch to fix
This commit is contained in:
ZanzyTHEbar 2026-03-05 14:24:45 +00:00
parent c9de7266c7
commit a0518f90ab
6 changed files with 633 additions and 936 deletions

View file

@ -30,6 +30,8 @@ type assembledContext struct {
fantasyHistory []fantasy.Message fantasyHistory []fantasy.Message
adaptedTools []fantasy.AgentTool adaptedTools []fantasy.AgentTool
agent fantasy.Agent agent fantasy.Agent
conversationID ids.UUID
runID ids.UUID
} }
func (al *AgentLoop) prepareRuntimeState(ctx context.Context, sessionKey string) (ids.UUID, ids.UUID, error) { func (al *AgentLoop) prepareRuntimeState(ctx context.Context, sessionKey string) (ids.UUID, ids.UUID, error) {
@ -96,7 +98,7 @@ func (al *AgentLoop) assembleContext(ctx context.Context, opts processOptions) (
fantasyHistory := dragonfantasy.MessagesToFantasy(historyMsgs) fantasyHistory := dragonfantasy.MessagesToFantasy(historyMsgs)
adaptedTools, prepareStep := al.prepareToolset(ctx, opts) adaptedTools, prepareStep := al.prepareToolset(ctx, opts)
agent, err := al.createFantasyAgent(ctx, opts, systemPrompt, adaptedTools, prepareStep) agent, conversationID, runID, err := al.createFantasyAgent(ctx, opts, systemPrompt, adaptedTools, prepareStep)
if err != nil { if err != nil {
return assembledContext{}, err return assembledContext{}, err
} }
@ -116,6 +118,8 @@ func (al *AgentLoop) assembleContext(ctx context.Context, opts processOptions) (
fantasyHistory: fantasyHistory, fantasyHistory: fantasyHistory,
adaptedTools: adaptedTools, adaptedTools: adaptedTools,
agent: agent, agent: agent,
conversationID: conversationID,
runID: runID,
}, nil }, nil
} }
@ -317,10 +321,10 @@ func (al *AgentLoop) prepareToolset(ctx context.Context, opts processOptions) ([
return adaptedTools, prepareStep return adaptedTools, prepareStep
} }
func (al *AgentLoop) createFantasyAgent(ctx context.Context, opts processOptions, systemPrompt string, adaptedTools []fantasy.AgentTool, prepareStep func(context.Context, fantasy.PrepareStepFunctionOptions) (context.Context, fantasy.PrepareStepResult, error)) (fantasy.Agent, error) { func (al *AgentLoop) createFantasyAgent(ctx context.Context, opts processOptions, systemPrompt string, adaptedTools []fantasy.AgentTool, prepareStep func(context.Context, fantasy.PrepareStepFunctionOptions) (context.Context, fantasy.PrepareStepResult, error)) (fantasy.Agent, ids.UUID, ids.UUID, error) {
conversationID, runID, err := al.prepareRuntimeState(ctx, opts.SessionKey) conversationID, runID, err := al.prepareRuntimeState(ctx, opts.SessionKey)
if err != nil { if err != nil {
return nil, err return nil, ids.UUID{}, ids.UUID{}, err
} }
baseRuntime := OffloadingToolRuntime{ baseRuntime := OffloadingToolRuntime{
@ -349,7 +353,7 @@ func (al *AgentLoop) createFantasyAgent(ctx context.Context, opts processOptions
agentOpts = append(agentOpts, fantasy.WithSystemPrompt(systemPrompt)) agentOpts = append(agentOpts, fantasy.WithSystemPrompt(systemPrompt))
} }
return fantasy.NewAgent(al.languageModel, agentOpts...), nil return fantasy.NewAgent(al.languageModel, agentOpts...), conversationID, runID, nil
} }
// postProcess handles the common finalization after Generate or Stream: // postProcess handles the common finalization after Generate or Stream:
@ -385,6 +389,21 @@ func (al *AgentLoop) postProcess(ctx context.Context, opts processOptions, final
"final_length": len(finalContent), "final_length": len(finalContent),
}) })
// Record task completion for RL analysis (best-effort, don't fail on error)
completion := TaskCompletion{
TaskID: opts.SessionKey,
Description: utils.Truncate(opts.UserMessage, 100),
TokensUsed: 0, // TODO: track actual token usage
ToolCalls: stepCount,
Errors: 0,
Completed: true,
CreatedAt: time.Now().UTC(),
}
if err := al.endTask(ctx, opts.ConversationID, opts.RunID, completion); err != nil {
logger.WarnCF("agent", "Failed to record task completion",
map[string]interface{}{"error": err.Error(), "session": opts.SessionKey})
}
return finalContent return finalContent
} }
@ -522,6 +541,11 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
}) })
return "", err return "", err
} }
// Populate IDs for task completion tracking
opts.ConversationID = ac.conversationID
opts.RunID = ac.runID
return al.postProcess(ctx, opts, finalContent, len(result.Steps)), nil return al.postProcess(ctx, opts, finalContent, len(result.Steps)), nil
} }
@ -579,6 +603,11 @@ func (al *AgentLoop) runStreaming(ctx context.Context, opts processOptions, ac a
}) })
return "", err return "", err
} }
// Populate IDs for task completion tracking
opts.ConversationID = ac.conversationID
opts.RunID = ac.runID
return al.postProcess(ctx, opts, finalContent, len(result.Steps)), nil return al.postProcess(ctx, opts, finalContent, len(result.Steps)), nil
} }

View file

@ -44,10 +44,9 @@ const promptCacheTTL = 30 * time.Second
func NewContextBuilder(workspace string) *ContextBuilder { func NewContextBuilder(workspace string) *ContextBuilder {
// Primary skills dir: XDG data dir (installed skills). // Primary skills dir: XDG data dir (installed skills).
// Falls back to workspace/skills for legacy setups. primarySkillsDir, _ := config.SkillsDir()
primarySkillsDir := filepath.Join(workspace, "skills") if primarySkillsDir == "" {
if dir, err := config.SkillsDir(); err == nil { primarySkillsDir = filepath.Join(workspace, "skills")
primarySkillsDir = dir
} }
// Global skills: ~/.config/dragonscale/skills (user-level overrides). // Global skills: ~/.config/dragonscale/skills (user-level overrides).

View file

@ -172,7 +172,7 @@ func TestIntegration_FullAgentLoop_SimpleResponse(t *testing.T) {
cfg := &config.Config{ cfg := &config.Config{
Agents: config.AgentsConfig{ Agents: config.AgentsConfig{
Defaults: config.AgentDefaults{ Defaults: config.AgentDefaults{
Workspace: tmpDir, Sandbox: tmpDir,
Model: "test-model", Model: "test-model",
MaxTokens: 4096, MaxTokens: 4096,
MaxToolIterations: 10, MaxToolIterations: 10,
@ -236,7 +236,7 @@ func TestIntegration_FullAgentLoop_WithToolCalls(t *testing.T) {
cfg := &config.Config{ cfg := &config.Config{
Agents: config.AgentsConfig{ Agents: config.AgentsConfig{
Defaults: config.AgentDefaults{ Defaults: config.AgentDefaults{
Workspace: tmpDir, Sandbox: tmpDir,
Model: "mock-tool-model", Model: "mock-tool-model",
MaxTokens: 4096, MaxTokens: 4096,
MaxToolIterations: 10, MaxToolIterations: 10,
@ -291,7 +291,7 @@ func TestIntegration_ProcessDirect(t *testing.T) {
cfg := &config.Config{ cfg := &config.Config{
Agents: config.AgentsConfig{ Agents: config.AgentsConfig{
Defaults: config.AgentDefaults{ Defaults: config.AgentDefaults{
Workspace: tmpDir, Sandbox: tmpDir,
Model: "test-model", Model: "test-model",
MaxTokens: 4096, MaxTokens: 4096,
MaxToolIterations: 10, MaxToolIterations: 10,
@ -387,7 +387,7 @@ func TestIntegration_Streaming_TextDeltas(t *testing.T) {
cfg := &config.Config{ cfg := &config.Config{
Agents: config.AgentsConfig{ Agents: config.AgentsConfig{
Defaults: config.AgentDefaults{ Defaults: config.AgentDefaults{
Workspace: tmpDir, Sandbox: tmpDir,
Model: "streaming-mock", Model: "streaming-mock",
MaxTokens: 4096, MaxTokens: 4096,
MaxToolIterations: 10, MaxToolIterations: 10,
@ -458,7 +458,7 @@ func TestIntegration_Streaming_WithToolCalls(t *testing.T) {
cfg := &config.Config{ cfg := &config.Config{
Agents: config.AgentsConfig{ Agents: config.AgentsConfig{
Defaults: config.AgentDefaults{ Defaults: config.AgentDefaults{
Workspace: tmpDir, Sandbox: tmpDir,
Model: "mock-tool-model", Model: "mock-tool-model",
MaxTokens: 4096, MaxTokens: 4096,
MaxToolIterations: 10, MaxToolIterations: 10,
@ -521,7 +521,7 @@ func TestIntegration_MultipleMessages(t *testing.T) {
cfg := &config.Config{ cfg := &config.Config{
Agents: config.AgentsConfig{ Agents: config.AgentsConfig{
Defaults: config.AgentDefaults{ Defaults: config.AgentDefaults{
Workspace: tmpDir, Sandbox: tmpDir,
Model: "test-model", Model: "test-model",
MaxTokens: 4096, MaxTokens: 4096,
MaxToolIterations: 10, MaxToolIterations: 10,

View file

@ -93,6 +93,8 @@ type processOptions struct {
SendResponse bool // Whether to send response via bus SendResponse bool // Whether to send response via bus
NoHistory bool // If true, don't load session history (for heartbeat) NoHistory bool // If true, don't load session history (for heartbeat)
Streaming bool // If true, stream token deltas to bus via OnTextDelta Streaming bool // If true, stream token deltas to bus via OnTextDelta
ConversationID ids.UUID // Conversation identifier for task tracking
RunID ids.UUID // Run identifier for task completion
} }
// Option configures AgentLoop creation. // Option configures AgentLoop creation.
@ -208,23 +210,6 @@ func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBu
contextBuilder.SetDelegate(del) contextBuilder.SetDelegate(del)
if migErr := memory.MigrateState(ctx, workspace, del, pkg.NAME); migErr != nil {
logger.WarnCF("agent", "State KV migration failed (non-fatal)",
map[string]interface{}{"error": migErr.Error()})
}
if migErr := memory.MigrateDocuments(ctx, workspace, del, pkg.NAME); migErr != nil {
logger.WarnCF("agent", "Document migration failed (non-fatal)",
map[string]interface{}{"error": migErr.Error()})
}
if migErr := memory.MigrateLongTermMemory(ctx, workspace, del, pkg.NAME); migErr != nil {
logger.WarnCF("agent", "Long-term memory migration failed (non-fatal)",
map[string]interface{}{"error": migErr.Error()})
}
if migErr := memory.MigrateDailyNotes(ctx, workspace, del, pkg.NAME); migErr != nil {
logger.WarnCF("agent", "Daily notes migration failed (non-fatal)",
map[string]interface{}{"error": migErr.Error()})
}
// Identity file sync (disk → DB) // Identity file sync (disk → DB)
var idSync *dragonsync.IdentitySync var idSync *dragonsync.IdentitySync
identityDir, idErr := config.IdentityDir() identityDir, idErr := config.IdentityDir()

File diff suppressed because it is too large Load diff

View file

@ -46,6 +46,10 @@ func (al *AgentLoop) maybeSummarize(ctx context.Context, sessionKey, channel, ch
} }
go func() { go func() {
if _, loading := al.summarizing.LoadOrStore(sessionKey, true); loading {
return
}
defer al.summarizing.Delete(sessionKey)
al.summarizeSession(context.WithoutCancel(ctx), sessionKey) al.summarizeSession(context.WithoutCancel(ctx), sessionKey)
}() }()
} }
@ -64,7 +68,7 @@ func (al *AgentLoop) compactionThresholds() (softPct, hardPct int) {
} }
} }
if hardPct <= softPct { if hardPct <= softPct {
hardPct = softPct + 10 hardPct = min(softPct+10, 100)
} }
return softPct, hardPct return softPct, hardPct
} }
@ -335,12 +339,14 @@ func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey stri
} }
if len(validMessages) == 0 { if len(validMessages) == 0 {
al.summarizeFailures.Delete(sessionKey)
return return
} }
// Multi-Part Summarization // **NEW**: Merge protection. If validMessages is very large and the model
// context window is small, split into two halves, summarize each, then merge.
var finalSummary string var finalSummary string
if len(validMessages) > 10 { if len(validMessages) > 40 && al.contextWindow < 16000 {
mid := len(validMessages) / 2 mid := len(validMessages) / 2
part1 := validMessages[:mid] part1 := validMessages[:mid]
part2 := validMessages[mid:] part2 := validMessages[mid:]