From 5ca4a8c6dab87a47b7c5490051bf14cca84512f0 Mon Sep 17 00:00:00 2001 From: ZanzyTHEbar Date: Sat, 21 Feb 2026 19:01:17 +0000 Subject: [PATCH] feat(agent): integrate DAG tools, map operators, obligations, and continuity config pkg/agent/loop.go - Register dag_expand, dag_describe, dag_grep, agentic_map, llm_map, map_run_status, map_run_cancel, and obligation_* tools in the agent loop - Wire DAGStore and MapRuntime into the tool registry at startup - Add projection pointer tracking: save pointer after each successful run, detect continuity breaks on session restore and log warnings - Emergency compression path: when context exceeds token budget, trigger DAG compression and save a snapshot before retrying pkg/agent/context.go - ContextBuilder now uses ContinuityRetentionConfig to bound the number of recent messages retained unsummarized (min/max/target ratio/failure fallback) - BuildSystemPrompt: add section for obligation awareness when due obligations exist in the current session pkg/config/config.go + config_test.go - Add ContinuityRetentionConfig struct with MinMessages, MaxMessages, TargetContextRatio, FailureKeepMessages fields - AgentDefaults.ContinuityRetention field wires the new config into the agent context builder - All env var names updated to DRAGONSCALE_* prefix pkg/tools/subagent.go - SubagentTask: add ParentTaskID, Depth, DelegatedScope, KeptWork fields for hierarchical delegation tracking - delegationCtxKey context values propagate task ID and depth through the call chain; prevents runaway recursion via max-depth guard - SubagentManager: add Cancel(), ListActive(), and GetTask() methods pkg/tools/subagent_tool_test.go + spawn_test.go + subagent_manager_test.go - Tests for delegation depth limiting, parent task ID propagation, cancel/list/get operations, and spawn tool integration pkg/tools/toolloop_test.go - Tests for ToolLoopConfig validation and result aggregation --- pkg/agent/context.go | 63 ++- pkg/agent/integration_test.go | 6 +- pkg/agent/loop.go | 743 +++++++++++++++++++++-------- pkg/agent/loop_test.go | 322 ++++++++++++- pkg/config/config.go | 278 +++++++---- pkg/config/config_test.go | 37 ++ pkg/tools/spawn_test.go | 44 ++ pkg/tools/subagent.go | 370 +++++++++++--- pkg/tools/subagent_manager_test.go | 227 +++++++++ pkg/tools/subagent_tool_test.go | 66 ++- pkg/tools/toolloop_test.go | 20 + 11 files changed, 1777 insertions(+), 399 deletions(-) create mode 100644 pkg/tools/spawn_test.go create mode 100644 pkg/tools/subagent_manager_test.go create mode 100644 pkg/tools/toolloop_test.go diff --git a/pkg/agent/context.go b/pkg/agent/context.go index dd6b38e74..8e8537db7 100644 --- a/pkg/agent/context.go +++ b/pkg/agent/context.go @@ -9,12 +9,13 @@ import ( "strings" "time" - "github.com/sipeed/picoclaw/pkg/config" - "github.com/sipeed/picoclaw/pkg/logger" - "github.com/sipeed/picoclaw/pkg/memory" - "github.com/sipeed/picoclaw/pkg/messages" - "github.com/sipeed/picoclaw/pkg/skills" - "github.com/sipeed/picoclaw/pkg/tools" + "github.com/ZanzyTHEbar/dragonscale/pkg/config" + "github.com/ZanzyTHEbar/dragonscale/pkg/logger" + "github.com/ZanzyTHEbar/dragonscale/pkg/memory" + "github.com/ZanzyTHEbar/dragonscale/pkg/memory/observation" + "github.com/ZanzyTHEbar/dragonscale/pkg/messages" + "github.com/ZanzyTHEbar/dragonscale/pkg/skills" + "github.com/ZanzyTHEbar/dragonscale/pkg/tools" ) type ContextBuilder struct { @@ -37,7 +38,7 @@ func NewContextBuilder(workspace string) *ContextBuilder { primarySkillsDir = dir } - // Global skills: ~/.config/picoclaw/skills (user-level overrides). + // Global skills: ~/.config/dragonscale/skills (user-level overrides). globalSkillsDir := "" if dir, err := config.ConfigDir(); err == nil { globalSkillsDir = filepath.Join(dir, "skills") @@ -99,9 +100,9 @@ func (cb *ContextBuilder) getIdentity() string { // Build tools section dynamically toolsSection := cb.buildToolsSection() - return fmt.Sprintf(`# picoclaw 🦞 + return fmt.Sprintf(`# dragonscale 🦞 -You are picoclaw, a helpful AI assistant. +You are dragonscale, a helpful AI assistant. ## Current Time %s @@ -153,12 +154,6 @@ func (cb *ContextBuilder) buildToolsSection() string { return sb.String() } -// roughTokenEstimate gives a conservative char-to-token ratio for budget checks. -// ~4 chars per token for English text is a standard heuristic. -// FIXME: This is a rough estimate and may not be accurate for all languages. -// FIXME: Implement a proper token estimator. -const charsPerToken = 4 - func (cb *ContextBuilder) BuildSystemPrompt() string { type section struct { name string @@ -183,8 +178,8 @@ func (cb *ContextBuilder) BuildSystemPrompt() string { The following skills extend your capabilities. To use a skill: 1. Use **skill_search** to find relevant skills by keyword -2. Use **skill_read** via tool_call to load the full skill content -3. Use **skill_traverse** via tool_call to explore related skills +2. Call **skill_read** directly to load the full skill content +3. Call **skill_traverse** directly to explore related skills Do NOT assume skill content β€” always load before applying. @@ -215,23 +210,25 @@ Do NOT assume skill content β€” always load before applying. // Token budget enforcement: if we exceed ~40% of context window for the // system prompt, trim lowest-priority sections first. - budgetChars := cb.tokenBudgetChars() - totalChars := 0 - for _, s := range sections { - totalChars += len(s.content) + budgetTokens := cb.tokenBudgetTokens() + totalTokens := 0 + sectionTokens := make([]int, len(sections)) + for i, s := range sections { + sectionTokens[i] = observation.EstimateTokens(s.content) + totalTokens += sectionTokens[i] } - if budgetChars > 0 && totalChars > budgetChars { + if budgetTokens > 0 && totalTokens > budgetTokens { logger.WarnCF("context", "System prompt exceeds token budget, trimming low-priority sections", map[string]interface{}{ - "total_chars": totalChars, - "budget_chars": budgetChars, - "sections": len(sections), + "total_tokens": totalTokens, + "budget_tokens": budgetTokens, + "sections": len(sections), }) // Trim from lowest priority (highest number) first - for i := len(sections) - 1; i >= 0 && totalChars > budgetChars; i-- { + for i := len(sections) - 1; i >= 0 && totalTokens > budgetTokens; i-- { if sections[i].priority >= 5 { // only trim P5+ (knowledge, dag) - totalChars -= len(sections[i].content) + totalTokens -= sectionTokens[i] sections[i].content = "" } } @@ -247,7 +244,7 @@ Do NOT assume skill content β€” always load before applying. prompt := strings.Join(parts, "\n\n---\n\n") // Log token estimate for observability - tokenEst := len(prompt) / charsPerToken + tokenEst := observation.EstimateTokens(prompt) logger.DebugCF("context", "System prompt token estimate", map[string]interface{}{ "chars": len(prompt), @@ -258,14 +255,14 @@ Do NOT assume skill content β€” always load before applying. return prompt } -// tokenBudgetChars returns the maximum character count for the system prompt, +// tokenBudgetTokens returns the maximum token count for the system prompt, // derived from the context window size. Returns 0 if no limit is configured. -func (cb *ContextBuilder) tokenBudgetChars() int { +func (cb *ContextBuilder) tokenBudgetTokens() int { if cb.contextWindow <= 0 { return 0 } // Reserve ~40% of context window for system prompt - return int(float64(cb.contextWindow) * 0.4 * charsPerToken) + return int(float64(cb.contextWindow) * 0.4) } func (cb *ContextBuilder) LoadBootstrapFiles() string { @@ -276,7 +273,7 @@ func (cb *ContextBuilder) LoadBootstrapFiles() string { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() - docs, err := cb.delegate.ListDocumentsByCategory(ctx, "picoclaw", "bootstrap") + docs, err := cb.delegate.ListDocumentsByCategory(ctx, "dragonscale", "bootstrap") if err != nil || len(docs) == 0 { return "" } @@ -302,7 +299,7 @@ func (cb *ContextBuilder) buildWorkingContextSection() string { var parts []string // Inject working context (hot tier) - wc, err := cb.memoryStore.GetWorkingContext(ctx, "picoclaw", "default") + wc, err := cb.memoryStore.GetWorkingContext(ctx, "dragonscale", "default") if err == nil && wc != "" { parts = append(parts, "## Working Context\n\n"+wc) } diff --git a/pkg/agent/integration_test.go b/pkg/agent/integration_test.go index 37ffe2ec8..86a03b19a 100644 --- a/pkg/agent/integration_test.go +++ b/pkg/agent/integration_test.go @@ -9,9 +9,9 @@ import ( "time" fantasy "charm.land/fantasy" - "github.com/sipeed/picoclaw/pkg/bus" - "github.com/sipeed/picoclaw/pkg/config" - "github.com/sipeed/picoclaw/pkg/tools" + "github.com/ZanzyTHEbar/dragonscale/pkg/bus" + "github.com/ZanzyTHEbar/dragonscale/pkg/config" + "github.com/ZanzyTHEbar/dragonscale/pkg/tools" ) // --- Mock language model that simulates tool calls --- diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 762856c0f..1f7271cba 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -1,13 +1,27 @@ -// PicoClaw - Ultra-lightweight personal AI agent +// DragonScale - Ultra-lightweight personal AI agent // Inspired by and based on nanobot: https://github.com/HKUDS/nanobot // License: MIT // -// Copyright (c) 2026 PicoClaw contributors +// Copyright (c) 2026 DragonScale contributors + +// FIXME: This file is a mess, we need to clean it up and make it more readable and maintainable +// Break this into modules with single responsibility and composability in mind +// Leverage Ports and Adapters pattern to achieve this (where boundaries are defined sensibly) +// - AgentLoop: Main agent loop and orchestrator +// - ContextBuilder: Builds the context for the agent +// - Tools: Tool registry and management +// - Memory: Memory store and management +// - State: State management +// - Session: Session management +// - Identity: Identity management +// - SecureBus: SecureBus management package agent import ( "context" + "encoding/json" + "errors" "fmt" "os" "path/filepath" @@ -15,29 +29,29 @@ import ( "sync" "sync/atomic" "time" - "unicode/utf8" fantasy "charm.land/fantasy" - "github.com/sipeed/picoclaw/pkg/bus" - "github.com/sipeed/picoclaw/pkg/channels" - "github.com/sipeed/picoclaw/pkg/config" - "github.com/sipeed/picoclaw/pkg/constants" - picofantasy "github.com/sipeed/picoclaw/pkg/fantasy" - "github.com/sipeed/picoclaw/pkg/ids" - "github.com/sipeed/picoclaw/pkg/logger" - "github.com/sipeed/picoclaw/pkg/memory" - "github.com/sipeed/picoclaw/pkg/memory/dag" - "github.com/sipeed/picoclaw/pkg/memory/delegate" - "github.com/sipeed/picoclaw/pkg/memory/observation" - memstore "github.com/sipeed/picoclaw/pkg/memory/store" - "github.com/sipeed/picoclaw/pkg/messages" - "github.com/sipeed/picoclaw/pkg/security" - "github.com/sipeed/picoclaw/pkg/security/securebus" - "github.com/sipeed/picoclaw/pkg/session" - "github.com/sipeed/picoclaw/pkg/state" - picosync "github.com/sipeed/picoclaw/pkg/sync" - "github.com/sipeed/picoclaw/pkg/tools" - "github.com/sipeed/picoclaw/pkg/utils" + "github.com/ZanzyTHEbar/dragonscale/pkg/bus" + "github.com/ZanzyTHEbar/dragonscale/pkg/channels" + "github.com/ZanzyTHEbar/dragonscale/pkg/config" + "github.com/ZanzyTHEbar/dragonscale/pkg/constants" + picofantasy "github.com/ZanzyTHEbar/dragonscale/pkg/fantasy" + "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + "github.com/ZanzyTHEbar/dragonscale/pkg/logger" + "github.com/ZanzyTHEbar/dragonscale/pkg/memory" + "github.com/ZanzyTHEbar/dragonscale/pkg/memory/dag" + "github.com/ZanzyTHEbar/dragonscale/pkg/memory/delegate" + "github.com/ZanzyTHEbar/dragonscale/pkg/memory/observation" + memsqlc "github.com/ZanzyTHEbar/dragonscale/pkg/memory/sqlc" + memstore "github.com/ZanzyTHEbar/dragonscale/pkg/memory/store" + "github.com/ZanzyTHEbar/dragonscale/pkg/messages" + "github.com/ZanzyTHEbar/dragonscale/pkg/security" + "github.com/ZanzyTHEbar/dragonscale/pkg/security/securebus" + "github.com/ZanzyTHEbar/dragonscale/pkg/session" + "github.com/ZanzyTHEbar/dragonscale/pkg/state" + picosync "github.com/ZanzyTHEbar/dragonscale/pkg/sync" + "github.com/ZanzyTHEbar/dragonscale/pkg/tools" + "github.com/ZanzyTHEbar/dragonscale/pkg/utils" ) type AgentLoop struct { @@ -54,7 +68,12 @@ type AgentLoop struct { memoryStore *memstore.MemoryStore // 3-tier MemGPT memory (always initialized) memDelegate memory.MemoryDelegate // DB delegate (always initialized) obsManager *observation.Manager // Observational memory (always initialized) - secureBus *securebus.Bus // ITR SecureBus (nil = disabled, direct execution) + secureBus *securebus.Bus // ITR SecureBus (always initialized) + queries *memsqlc.Queries // SQL query surface for runtime persistence + kvDelegate KVDelegate // KV adapter for offloaded tool results + stateStore *StateStore // Agent run state persistence + conversationIDs sync.Map // map[sessionKey]ids.UUID + conversationMu sync.Mutex // serializes conversation creation path identitySync *picosync.IdentitySync // Fileβ†’DB sync for identity docs (nil if memory disabled) activeSessionKey atomic.Value // Current session key for tool access running atomic.Bool @@ -62,6 +81,7 @@ type AgentLoop struct { summarizeFailures sync.Map // Tracks consecutive summarization failures per session (string -> int) cfg *config.Config // Stored for subagent factory access channelManager *channels.Manager + toolResultSearch fantasy.AgentTool } // processOptions configures how a message is processed @@ -77,8 +97,30 @@ type processOptions struct { Streaming bool // If true, stream token deltas to bus via OnTextDelta } +func initSecretStore() (*security.SecretStore, error) { + cfgDir, err := config.ConfigDir() + if err != nil { + return nil, fmt.Errorf("resolve config dir: %w", err) + } + + secretsPath := filepath.Join(cfgDir, "secrets.json") + keyring := security.NewEnvKeyring(security.MasterKeyEnvVar) + ss, err := security.NewSecretStore(secretsPath, keyring) + if err != nil { + return nil, fmt.Errorf("initialize secret store: %w", err) + } + + if os.Getenv(security.MasterKeyEnvVar) == "" { + logger.WarnCF("security", "master key env var is not set; secret injection requiring stored secrets will fail", + map[string]interface{}{"env_var": security.MasterKeyEnvVar}) + } + return ss, nil +} + // createToolRegistry creates a tool registry with common tools. -// This is shared between main agent and subagents. +// createToolRegistry builds the base tool set (filesystem, shell, web, etc.). +// Parent and subagent registries start from the same base; memory/search/skill +// tools are registered separately on each so they have isolated discovery state. func createToolRegistry(workspace string, restrict bool, cfg *config.Config, msgBus *bus.MessageBus) *tools.ToolRegistry { registry := tools.NewToolRegistry() @@ -144,6 +186,7 @@ func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBu subagentTool := tools.NewSubagentTool(subagentManager) toolsRegistry.Register(subagentTool) + agenticMapTool := tools.NewAgenticMapTool(subagentManager) contextBuilder := NewContextBuilder(workspace) contextBuilder.SetToolsRegistry(toolsRegistry) @@ -167,10 +210,20 @@ func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBu return nil, fmt.Errorf("memory schema init: %w", err) } memDelegate := del + queries := memDelegate.Queries() + if queries == nil { + del.Close() + return nil, fmt.Errorf("memory delegate queries are not initialized") + } + kv := NewDelegateKV(memDelegate, "dragonscale") + stateStore := NewStateStore(queries) offloadThreshold := cfg.Memory.OffloadThresholdTokens if offloadThreshold <= 0 { - offloadThreshold = 4000 + // Default to ~3% of the context window, floored at 2048 tokens. + // Small windows (4K-8K) get proportionally tighter thresholds; + // large windows (128K+) avoid wasteful offloading of short results. + offloadThreshold = max(cfg.Agents.Defaults.MaxTokens*3/100, 2048) } chunker := memstore.NewMarkdownChunker(memstore.DefaultMarkdownChunkerConfig()) @@ -186,34 +239,55 @@ func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBu }) contextBuilder.SetMemoryStore(ms) - memTool := NewMemGPTTool(ms, "picoclaw", "default") + memTool := NewMemGPTTool(ms, "dragonscale", "default") toolsRegistry.Register(memTool) + toolsRegistry.Register(tools.NewObligationTool(memDelegate, "dragonscale")) - toolsRegistry.Register(tools.NewKeywordSearchTool(ms, "picoclaw")) - toolsRegistry.Register(tools.NewSemanticSearchTool(ms, "picoclaw")) - toolsRegistry.Register(tools.NewChunkReadTool(ms, "picoclaw")) + toolsRegistry.Register(tools.NewKeywordSearchTool(ms, "dragonscale")) + toolsRegistry.Register(tools.NewSemanticSearchTool(ms, "dragonscale")) + toolsRegistry.Register(tools.NewChunkReadTool(ms, "dragonscale")) + mapRuntime := tools.NewMapRuntime(queries, "dragonscale", model, cfg.Agents.Defaults.Model, subagentManager) + agenticMapTool.SetRuntime(mapRuntime) + toolsRegistry.Register(agenticMapTool) + llmMapTool := tools.NewLLMMapTool(model, cfg.Agents.Defaults.Model) + llmMapTool.SetRuntime(mapRuntime) + toolsRegistry.Register(llmMapTool) + toolsRegistry.Register(tools.NewMapRunStatusTool(mapRuntime)) + toolsRegistry.Register(tools.NewMapRunReadTool(mapRuntime)) + + // Register memory/search/skill tools on subagent registry so spawned + // agents can search knowledge, offload results, and use skills. + subagentTools.Register(NewMemGPTTool(ms, "dragonscale", "default")) + subagentTools.Register(tools.NewObligationTool(memDelegate, "dragonscale")) + subagentTools.Register(tools.NewKeywordSearchTool(ms, "dragonscale")) + subagentTools.Register(tools.NewSemanticSearchTool(ms, "dragonscale")) + subagentTools.Register(tools.NewChunkReadTool(ms, "dragonscale")) + subagentTools.Register(tools.NewSkillSearchTool(sl)) + subagentTools.Register(tools.NewSkillReadTool(sl)) + subagentTools.RegisterMetaTools() + for _, name := range []string{"memory", "read_file", "write_file", "list_dir", "exec"} { + subagentTools.MarkGateway(name) + } contextBuilder.SetDelegate(del) - if migErr := memory.MigrateState(ctx, workspace, del, "picoclaw"); migErr != nil { + if migErr := memory.MigrateState(ctx, workspace, del, "dragonscale"); migErr != nil { logger.WarnCF("agent", "State KV migration failed (non-fatal)", map[string]interface{}{"error": migErr.Error()}) } - if migErr := memory.MigrateDocuments(ctx, workspace, del, "picoclaw"); migErr != nil { + if migErr := memory.MigrateDocuments(ctx, workspace, del, "dragonscale"); migErr != nil { logger.WarnCF("agent", "Document migration failed (non-fatal)", map[string]interface{}{"error": migErr.Error()}) } - if migErr := memory.MigrateLongTermMemory(ctx, workspace, del, "picoclaw"); migErr != nil { + if migErr := memory.MigrateLongTermMemory(ctx, workspace, del, "dragonscale"); 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, "picoclaw"); migErr != nil { + if migErr := memory.MigrateDailyNotes(ctx, workspace, del, "dragonscale"); migErr != nil { logger.WarnCF("agent", "Daily notes migration failed (non-fatal)", map[string]interface{}{"error": migErr.Error()}) } - subagentManager.SetRunLoop(MakeRunLoopFunc(ms)) - // Identity file sync (disk β†’ DB) var idSync *picosync.IdentitySync identityDir, idErr := config.IdentityDir() @@ -221,7 +295,7 @@ func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBu logger.WarnCF("agent", "Could not resolve identity dir, identity sync disabled", map[string]interface{}{"error": idErr.Error()}) } else { - idSync = picosync.New(identityDir, "picoclaw", memDelegate) + idSync = picosync.New(identityDir, "dragonscale", memDelegate) if syncErr := idSync.SyncAll(ctx); syncErr != nil { logger.WarnCF("agent", "Initial identity sync failed (non-fatal)", map[string]interface{}{"error": syncErr.Error()}) @@ -239,7 +313,24 @@ func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBu // Session manager (always delegate-backed) sessionsDir := filepath.Join(workspace, "sessions") - sessionsManager := session.NewSessionManager(sessionsDir, session.WithSessionDelegate(memDelegate, "picoclaw")) + sessionsManager := session.NewSessionManager(sessionsDir, session.WithSessionDelegate(memDelegate, "dragonscale")) + + // One-shot DAG backfill for pre-existing session histories. + backfillCtx, cancelBackfill := context.WithTimeout(ctx, 10*time.Second) + defer cancelBackfill() + if status, err := dag.BackfillMissingSessionDAGs(backfillCtx, memDelegate, queries, "dragonscale", dag.DefaultBackfillOptions()); err != nil { + logger.WarnCF("agent", "DAG backfill failed (non-fatal)", + map[string]interface{}{"error": err.Error()}) + } else if status != nil { + logger.InfoCF("agent", "DAG backfill pass completed", + map[string]interface{}{ + "sessions_scanned": status.SessionsScanned, + "snapshots_created": status.SnapshotsCreated, + "skipped_existing": status.SkippedExisting, + "failures": status.Failures, + "skipped": status.Skipped, + }) + } // Meta-tools for progressive disclosure (tool_search + tool_call). // tool_search returns full parameter schemas; discovered tools are @@ -281,25 +372,29 @@ func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBu } return resp.Content.Text(), nil } - obsManager := observation.NewManager(memDelegate, "picoclaw", callModelFn, observation.DefaultManagerConfig()) + obsManager := observation.NewManager(memDelegate, "dragonscale", callModelFn, observation.DefaultManagerConfig()) al := &AgentLoop{ - bus: msgBus, - languageModel: model, - workspace: workspace, - model: cfg.Agents.Defaults.Model, - contextWindow: cfg.Agents.Defaults.MaxTokens, - maxIterations: cfg.Agents.Defaults.MaxToolIterations, - sessions: sessionsManager, - state: stateManager, - contextBuilder: contextBuilder, - tools: toolsRegistry, - memoryStore: ms, - memDelegate: memDelegate, - obsManager: obsManager, - identitySync: idSync, - summarizing: sync.Map{}, - cfg: cfg, + bus: msgBus, + languageModel: model, + workspace: workspace, + model: cfg.Agents.Defaults.Model, + contextWindow: cfg.Agents.Defaults.MaxTokens, + maxIterations: cfg.Agents.Defaults.MaxToolIterations, + sessions: sessionsManager, + state: stateManager, + contextBuilder: contextBuilder, + tools: toolsRegistry, + memoryStore: ms, + memDelegate: memDelegate, + obsManager: obsManager, + queries: queries, + kvDelegate: kv, + stateStore: stateStore, + toolResultSearch: NewToolResultSearchTool(queries, kv), + identitySync: idSync, + summarizing: sync.Map{}, + cfg: cfg, } // Focus tools (start_focus / complete_focus) @@ -312,6 +407,26 @@ func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBu toolsRegistry.Register(tools.NewStartFocusTool(memDelegate, sessionsManager, sessionKeyFn)) toolsRegistry.Register(tools.NewCompleteFocusTool(memDelegate, sessionsManager, sessionKeyFn)) + // DAG tools: dag_expand, dag_describe, dag_grep (require delegate-backed session) + dagDeps := tools.DAGToolDeps{ + Queries: queries, + Lister: del, + Delegate: del, + AgentID: "dragonscale", + SessionFn: sessionKeyFn, + } + toolsRegistry.Register(tools.NewDagExpandTool(dagDeps)) + toolsRegistry.Register(tools.NewDagDescribeTool(dagDeps)) + toolsRegistry.Register(tools.NewDagGrepTool(dagDeps)) + + // Unified runtime invariant: SecureBus is always enabled for tool execution. + secretStore, err := initSecretStore() + if err != nil { + return nil, fmt.Errorf("failed to create secret store: %w", err) + } + al.SetupSecureBus(secretStore, securebus.DefaultBusConfig()) + subagentManager.SetRunLoop(MakeUnifiedRunLoopFunc(al)) + return al, nil } @@ -379,10 +494,9 @@ func (al *AgentLoop) SetChannelManager(cm *channels.Manager) { al.channelManager = cm } -// SetSecureBus attaches a SecureBus to the agent loop. When set, all tool +// SetSecureBus attaches a SecureBus to the agent loop. All tool // calls are routed through the bus for capability enforcement, secret injection, // leak scanning, and audit logging. Call before the first message is processed. -// Pass nil to disable SecureBus enforcement (direct execution, default). func (al *AgentLoop) SetSecureBus(b *securebus.Bus) { al.secureBus = b } @@ -461,7 +575,16 @@ func (al *AgentLoop) ProcessDirectStreaming(ctx context.Context, content, sessio // ProcessHeartbeat processes a heartbeat request without session history. // Each heartbeat is independent and doesn't accumulate context. +// It injects the active session's summary so the agent has awareness of +// recent user conversation context. func (al *AgentLoop) ProcessHeartbeat(ctx context.Context, content, channel, chatID string) (string, error) { + if v := al.activeSessionKey.Load(); v != nil { + if key, ok := v.(string); ok && key != "" { + if summary := al.sessions.GetSummary(key); summary != "" { + content = content + "\n\n## Recent User Context\n" + summary + } + } + } return al.runAgentLoop(ctx, processOptions{ SessionKey: "heartbeat", Channel: channel, @@ -469,7 +592,7 @@ func (al *AgentLoop) ProcessHeartbeat(ctx context.Context, content, channel, cha UserMessage: content, EnableSummary: false, SendResponse: false, - NoHistory: true, // Don't load session history for heartbeat + NoHistory: true, }) } @@ -572,10 +695,47 @@ type assembledContext struct { agent fantasy.Agent } +func (al *AgentLoop) prepareRuntimeState(ctx context.Context, sessionKey string) (ids.UUID, ids.UUID, error) { + if al.queries == nil || al.stateStore == nil || al.kvDelegate == nil { + return ids.UUID{}, ids.UUID{}, errors.New("runtime persistence dependencies are not initialized") + } + if strings.TrimSpace(sessionKey) == "" { + return ids.UUID{}, ids.UUID{}, errors.New("session key is required") + } + + var conversationID ids.UUID + if cached, ok := al.conversationIDs.Load(sessionKey); ok { + conversationID = cached.(ids.UUID) + } else { + al.conversationMu.Lock() + defer al.conversationMu.Unlock() + if cached, ok := al.conversationIDs.Load(sessionKey); ok { + conversationID = cached.(ids.UUID) + } else { + conversationID = ids.New() + title := sessionKey + if _, err := al.queries.CreateAgentConversation(ctx, memsqlc.CreateAgentConversationParams{ + ID: conversationID, + Title: &title, + }); err != nil { + return ids.UUID{}, ids.UUID{}, fmt.Errorf("create agent conversation: %w", err) + } + al.conversationIDs.Store(sessionKey, conversationID) + } + } + + run, err := al.stateStore.CreateRun(ctx, conversationID) + if err != nil { + return ids.UUID{}, ids.UUID{}, fmt.Errorf("create agent run: %w", err) + } + + return conversationID, run.ID, nil +} + // assembleContext performs the shared pre-processing for every agent turn: // record channel, update tool contexts, load memory blocks, build messages, // DAG-compress history, split into system/history/user, adapt tools, create Fantasy agent. -func (al *AgentLoop) assembleContext(ctx context.Context, opts processOptions) assembledContext { +func (al *AgentLoop) assembleContext(ctx context.Context, opts processOptions) (assembledContext, error) { if opts.Channel != "" && opts.ChatID != "" { if !constants.IsInternalChannel(opts.Channel) { channelKey := fmt.Sprintf("%s:%s", opts.Channel, opts.ChatID) @@ -606,7 +766,7 @@ func (al *AgentLoop) assembleContext(ctx context.Context, opts processOptions) a summary = al.sessions.GetSummary(opts.SessionKey) } - history = al.applyDAGCompression(history) + history = al.applyDAGCompression(ctx, opts.SessionKey, history) if al.identitySync != nil { _ = al.identitySync.CheckAndSync(ctx) @@ -634,10 +794,13 @@ func (al *AgentLoop) assembleContext(ctx context.Context, opts processOptions) a adaptCfg := picofantasy.AdaptedToolsConfig{ MemStore: al.memoryStore, - AgentID: "picoclaw", + AgentID: "dragonscale", SessionKey: opts.SessionKey, } adaptedTools := picofantasy.BuildAdaptedTools(al.tools, al.bus, opts.Channel, opts.ChatID, adaptCfg) + if al.toolResultSearch != nil { + adaptedTools = append(adaptedTools, al.toolResultSearch) + } // Dynamic tool promotion via PrepareStep: after tool_search discovers tools, // they become native callables in the next inference step β€” no tool_call needed. @@ -651,6 +814,7 @@ func (al *AgentLoop) assembleContext(ctx context.Context, opts processOptions) a chatID := opts.ChatID prepareStep := func(ctx context.Context, psOpts fantasy.PrepareStepFunctionOptions) (context.Context, fantasy.PrepareStepResult, error) { + _ = psOpts discovered := registry.DrainDiscovered() if len(discovered) == 0 { return ctx, fantasy.PrepareStepResult{}, nil @@ -685,21 +849,35 @@ func (al *AgentLoop) assembleContext(ctx context.Context, opts processOptions) a }, nil } + conversationID, runID, err := al.prepareRuntimeState(ctx, opts.SessionKey) + if err != nil { + return assembledContext{}, err + } + + baseRuntime := OffloadingToolRuntime{ + Base: fantasy.DAGToolRuntime{MaxConcurrency: defaultToolMaxConcurrency}, + KV: al.kvDelegate, + Queries: al.queries, + ConversationID: conversationID, + RunID: runID, + } + toolRuntime := SecureBusToolRuntime{ + Base: baseRuntime, + Bus: al.secureBus, + SessionKey: opts.SessionKey, + StateStore: al.stateStore, + RunID: runID, + } + agentOpts := []fantasy.AgentOption{ fantasy.WithTools(adaptedTools...), fantasy.WithStopConditions(fantasy.StepCountIs(al.maxIterations)), fantasy.WithPrepareStep(prepareStep), + fantasy.WithToolRuntime(toolRuntime), } if systemPrompt != "" { agentOpts = append(agentOpts, fantasy.WithSystemPrompt(systemPrompt)) } - if al.secureBus != nil { - sbrt := SecureBusToolRuntime{ - Bus: al.secureBus, - SessionKey: opts.SessionKey, - } - agentOpts = append(agentOpts, fantasy.WithToolRuntime(sbrt)) - } agent := fantasy.NewAgent(al.languageModel, agentOpts...) logger.DebugCF("agent", "Fantasy agent created", @@ -717,7 +895,7 @@ func (al *AgentLoop) assembleContext(ctx context.Context, opts processOptions) a fantasyHistory: fantasyHistory, adaptedTools: adaptedTools, agent: agent, - } + }, nil } // postProcess handles the common finalization after Generate or Stream: @@ -849,7 +1027,10 @@ func (al *AgentLoop) resolveFinalContent(finalContent string, steps []fantasy.St func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (string, error) { al.activeSessionKey.Store(opts.SessionKey) - ac := al.assembleContext(ctx, opts) + ac, err := al.assembleContext(ctx, opts) + if err != nil { + return "", err + } if opts.Streaming { return al.runStreaming(ctx, opts, ac) @@ -954,7 +1135,7 @@ func (al *AgentLoop) auditStep(ctx context.Context, step fantasy.StepResult, ses for _, tc := range toolCalls { entry := &memory.AuditEntry{ ID: ids.New(), - AgentID: "picoclaw", + AgentID: "dragonscale", SessionKey: sessionKey, Action: "tool_call", Target: tc.ToolName, @@ -989,95 +1170,107 @@ func (al *AgentLoop) updateToolContexts(channel, chatID string) { } } -// maybeSummarize triggers summarization if the session history exceeds thresholds. -// At the critical threshold (β‰₯95% of context window) it synchronously force-compresses -// the history before the normal async summarization path runs. +// maybeSummarize only triggers emergency compression when hard limits are exceeded. +// Normal background compaction is intentionally disabled for the unified kernel. func (al *AgentLoop) maybeSummarize(ctx context.Context, sessionKey, channel, chatID string) { + _ = channel + _ = chatID newHistory := al.sessions.GetHistory(sessionKey) tokenEstimate := al.estimateTokens(newHistory) - threshold := al.contextWindow * 75 / 100 criticalThreshold := al.contextWindow * 95 / 100 if tokenEstimate > criticalThreshold { - al.forceCompression(sessionKey) - return + al.forceCompression(ctx, sessionKey) } - if len(newHistory) > 20 || tokenEstimate > threshold { - if _, loading := al.summarizing.LoadOrStore(sessionKey, true); !loading { - go func() { - defer al.summarizing.Delete(sessionKey) - if !constants.IsInternalChannel(channel) { - al.bus.PublishOutbound(bus.OutboundMessage{ - Channel: channel, - ChatID: chatID, - Content: "⚠️ Memory threshold reached. Optimizing conversation history...", - }) - } - al.summarizeSession(ctx, sessionKey) - }() - } + // TODO: actually use the channel and chatID to push data to the bus + //if _, loading := al.summarizing.LoadOrStore(sessionKey, true); !loading { + // go func() { + // defer al.summarizing.Delete(sessionKey) + // if !constants.IsInternalChannel(channel) { + // al.bus.PublishOutbound(bus.OutboundMessage{ + // Channel: channel, + // ChatID: chatID, + // Content: "⚠️ Memory threshold reached. Optimizing conversation history...", + // }) + // } + // al.summarizeSession(ctx, sessionKey) + // }() + // } +} + +// EmergencyProvenance captures provenance metadata for postmortem when +// emergency compression cycles run. Persisted via the audit pipeline. +type EmergencyProvenance struct { + SessionKey string `json:"session_key"` + Cycle int `json:"cycle"` + TokenEstimate int `json:"token_estimate"` + CriticalBudget int `json:"critical_budget"` + HistoryMsgCount int `json:"history_msg_count"` +} + +// persistEmergencyProvenance writes provenance metadata to the audit log. +// Best-effort: logs warning on failure, never fails the compression path. +func (al *AgentLoop) persistEmergencyProvenance(ctx context.Context, prov EmergencyProvenance) { + if al.memDelegate == nil { + return + } + input, err := json.Marshal(prov) + if err != nil { + logger.WarnCF("agent", "Failed to marshal emergency provenance", + map[string]interface{}{"error": err.Error()}) + return + } + entry := &memory.AuditEntry{ + ID: ids.New(), + AgentID: "dragonscale", + SessionKey: prov.SessionKey, + Action: "emergency_compression", + Target: fmt.Sprintf("cycle_%d", prov.Cycle), + Input: string(input), + } + aCtx, cancel := context.WithTimeout(ctx, time.Second) + defer cancel() + if err := al.memDelegate.InsertAuditEntry(aCtx, entry); err != nil { + logger.WarnCF("agent", "Failed to persist emergency provenance", + map[string]interface{}{"error": err.Error(), "session_key": prov.SessionKey}) } } -// forceCompression aggressively reduces context when the limit is hit. -// It drops the oldest 50% of messages (keeping system prompt and last user message). -func (al *AgentLoop) forceCompression(sessionKey string) { - history := al.sessions.GetHistory(sessionKey) - if len(history) <= 4 { - return +// forceCompression performs emergency recursive compression by repeatedly +// summarizing older history until under hard budget, without deleting immutable +// persisted session records. +func (al *AgentLoop) forceCompression(ctx context.Context, sessionKey string) { + const maxCycles = 3 + for cycle := 1; cycle <= maxCycles; cycle++ { + history := al.sessions.GetHistory(sessionKey) + if len(history) <= al.continuityKeepCount(history) { + return + } + tokenEstimate := al.estimateTokens(history) + criticalThreshold := al.contextWindow * 95 / 100 + if tokenEstimate <= criticalThreshold { + return + } + + logger.WarnCF("agent", "Emergency compression cycle triggered", + map[string]interface{}{ + "session_key": sessionKey, + "cycle": cycle, + "token_estimate": tokenEstimate, + "critical_budget": criticalThreshold, + }) + + al.persistEmergencyProvenance(ctx, EmergencyProvenance{ + SessionKey: sessionKey, + Cycle: cycle, + TokenEstimate: tokenEstimate, + CriticalBudget: criticalThreshold, + HistoryMsgCount: len(history), + }) + + al.summarizeSession(ctx, sessionKey) } - - // Keep system prompt (usually [0]) and the very last message (user's trigger) - // We want to drop the oldest half of the *conversation* - // Assuming [0] is system, [1:] is conversation - conversation := history[1 : len(history)-1] - if len(conversation) == 0 { - return - } - - // Helper to find the mid-point of the conversation - mid := len(conversation) / 2 - - // New history structure: - // 1. System Prompt - // 2. [Summary of dropped part] - synthesized - // 3. Second half of conversation - // 4. Last message - - // Simplified approach for emergency: Drop first half of conversation - // and rely on existing summary if present, or create a placeholder. - - droppedCount := mid - keptConversation := conversation[mid:] - - newHistory := make([]messages.Message, 0) - newHistory = append(newHistory, history[0]) // System prompt - - // Add a note about compression - compressionNote := fmt.Sprintf("[System: Emergency compression dropped %d oldest messages due to context limit]", droppedCount) - // If there was an existing summary, we might lose it if it was in the dropped part (which is just messages). - // The summary is stored separately in session.Summary, so it persists! - // We just need to ensure the user knows there's a gap. - - // We only modify the messages list here - newHistory = append(newHistory, messages.Message{ - Role: "system", - Content: compressionNote, - }) - - newHistory = append(newHistory, keptConversation...) - newHistory = append(newHistory, history[len(history)-1]) // Last message - - // Update session - al.sessions.SetHistory(sessionKey, newHistory) - al.sessions.Save(sessionKey) - - logger.WarnCF("agent", "Forced compression executed", map[string]interface{}{ - "session_key": sessionKey, - "dropped_msgs": droppedCount, - "new_count": len(newHistory), - }) } // MemoryDelegate returns the active memory delegate (nil if memory system is disabled). @@ -1085,6 +1278,17 @@ func (al *AgentLoop) MemoryDelegate() memory.MemoryDelegate { return al.memDelegate } +// HasSecureBus reports whether SecureBus enforcement is active. +func (al *AgentLoop) HasSecureBus() bool { + return al.secureBus != nil +} + +// HasUnifiedRuntimeDeps reports whether unified runtime persistence dependencies +// are available. +func (al *AgentLoop) HasUnifiedRuntimeDeps() bool { + return al.queries != nil && al.kvDelegate != nil && al.stateStore != nil +} + // GetStartupInfo returns information about loaded tools and skills for logging. func (al *AgentLoop) GetStartupInfo() map[string]interface{} { info := make(map[string]interface{}) @@ -1134,6 +1338,133 @@ func formatMessagesForLog(msgs []messages.Message) string { return result } +func (al *AgentLoop) continuityRetentionPolicy() config.ContinuityRetentionConfig { + policy := config.ContinuityRetentionConfig{ + MinMessages: 4, + MaxMessages: 24, + TargetContextRatio: 0.10, + FailureKeepMessages: 10, + } + + if al.cfg != nil { + cfgPolicy := al.cfg.Agents.Defaults.ContinuityRetention + if cfgPolicy.MinMessages > 0 { + policy.MinMessages = cfgPolicy.MinMessages + } + if cfgPolicy.MaxMessages > 0 { + policy.MaxMessages = cfgPolicy.MaxMessages + } + if cfgPolicy.TargetContextRatio > 0 && cfgPolicy.TargetContextRatio <= 0.5 { + policy.TargetContextRatio = cfgPolicy.TargetContextRatio + } + if cfgPolicy.FailureKeepMessages > 0 { + policy.FailureKeepMessages = cfgPolicy.FailureKeepMessages + } + } + + if policy.MaxMessages < policy.MinMessages { + policy.MaxMessages = policy.MinMessages + } + if policy.FailureKeepMessages < policy.MinMessages { + policy.FailureKeepMessages = policy.MinMessages + } + + return policy +} + +func (al *AgentLoop) continuityKeepCount(history []messages.Message) int { + if len(history) == 0 { + return 0 + } + + policy := al.continuityRetentionPolicy() + minKeep := policy.MinMessages + if minKeep > len(history) { + minKeep = len(history) + } + maxKeep := policy.MaxMessages + if maxKeep > len(history) { + maxKeep = len(history) + } + if maxKeep < minKeep { + maxKeep = minKeep + } + + contextWindow := al.contextWindow + if contextWindow <= 0 && al.cfg != nil { + contextWindow = al.cfg.Agents.Defaults.MaxTokens + } + if contextWindow <= 0 { + return minKeep + } + + targetTokens := int(float64(contextWindow) * policy.TargetContextRatio) + if targetTokens <= 0 { + return minKeep + } + + keep := 0 + keptTokens := 0 + for i := len(history) - 1; i >= 0 && keep < maxKeep; i-- { + msgTokens := observation.EstimateTokens(history[i].Content) + 4 + if keep >= minKeep && keptTokens+msgTokens > targetTokens { + break + } + keptTokens += msgTokens + keep++ + } + if keep < minKeep { + keep = minKeep + } + return keep +} + +type oversizedRecoveryCandidate struct { + Message messages.Message + OriginalIndex int + TokenEstimate int +} + +func (al *AgentLoop) persistOversizedRecoveryRefs(ctx context.Context, sessionKey string, omitted []oversizedRecoveryCandidate) ([]string, error) { + if len(omitted) == 0 { + return nil, nil + } + if al.memDelegate == nil { + return nil, fmt.Errorf("memory delegate is not configured") + } + + const maxPersistedRefs = 8 + refs := make([]string, 0, len(omitted)) + now := time.Now().UTC() + + for i, candidate := range omitted { + if i >= maxPersistedRefs { + break + } + nodeID := tools.DAGRecoveryNodePrefix + ids.New().String() + record := tools.DAGRecoveryRecord{ + NodeID: nodeID, + SessionKey: sessionKey, + OriginalIndex: candidate.OriginalIndex, + Role: candidate.Message.Role, + Content: candidate.Message.Content, + TokenEstimate: candidate.TokenEstimate, + Reason: "oversized_message_omitted_from_summary", + CreatedAt: now, + } + + data, err := json.Marshal(record) + if err != nil { + return refs, fmt.Errorf("marshal DAG recovery record: %w", err) + } + if err := al.memDelegate.UpsertKV(ctx, "dragonscale", tools.DAGRecoveryKVKey(sessionKey, nodeID), string(data)); err != nil { + return refs, fmt.Errorf("persist DAG recovery record: %w", err) + } + refs = append(refs, nodeID) + } + return refs, nil +} + // summarizeSession summarizes the conversation history for a session. func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey string) { ctx, cancel := context.WithTimeout(parentCtx, 120*time.Second) @@ -1142,26 +1473,37 @@ func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey stri history := al.sessions.GetHistory(sessionKey) summary := al.sessions.GetSummary(sessionKey) - // Keep last 4 messages for continuity - if len(history) <= 4 { + keepLast := al.continuityKeepCount(history) + if len(history) <= keepLast { return } - toSummarize := history[:len(history)-4] + toSummarize := history[:len(history)-keepLast] - // Oversized Message Guard - // Skip messages larger than 50% of context window to prevent summarizer overflow - maxMessageTokens := al.contextWindow / 2 + // Oversized Message Guard: skip individual messages that would consume too + // much of the summarizer's context. Use 40% of the window for the summarizer + // input budget, reserving the rest for system prompt + summary output. + // Oversized omissions are persisted as DAG recovery references. + maxMessageTokens := al.contextWindow * 40 / 100 + if maxMessageTokens < 2048 { + maxMessageTokens = 2048 + } validMessages := make([]messages.Message, 0) omitted := false + omittedMessages := make([]oversizedRecoveryCandidate, 0) - for _, m := range toSummarize { + for idx, m := range toSummarize { if m.Role != "user" && m.Role != "assistant" { continue } - msgTokens := len(m.Content) / 2 + msgTokens := observation.EstimateTokens(m.Content) if msgTokens > maxMessageTokens { omitted = true + omittedMessages = append(omittedMessages, oversizedRecoveryCandidate{ + Message: m, + OriginalIndex: idx, + TokenEstimate: msgTokens, + }) continue } validMessages = append(validMessages, m) @@ -1194,12 +1536,26 @@ func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey stri } if omitted && finalSummary != "" { - finalSummary += "\n[Note: Some oversized messages were omitted from this summary for efficiency.]" + recoveryRefs, err := al.persistOversizedRecoveryRefs(ctx, sessionKey, omittedMessages) + if err != nil { + logger.WarnCF("agent", "Failed to persist DAG recovery references for oversized messages", + map[string]interface{}{ + "session_key": sessionKey, + "error": err.Error(), + "omitted": len(omittedMessages), + }) + finalSummary += "\n[Note: Some oversized messages were omitted from this summary for efficiency.]" + } else if len(recoveryRefs) > 0 { + finalSummary += fmt.Sprintf("\n[Note: %d oversized message(s) were omitted from this summary. Recovery refs: %s. Use dag_expand with node_id= to recover full content.]", + len(omittedMessages), strings.Join(recoveryRefs, ", ")) + } else { + finalSummary += "\n[Note: Some oversized messages were omitted from this summary for efficiency.]" + } } if finalSummary != "" { al.sessions.SetSummary(sessionKey, finalSummary) - al.sessions.TruncateHistory(sessionKey, 4) + al.sessions.TruncateHistory(sessionKey, keepLast) al.sessions.Save(sessionKey) al.summarizeFailures.Delete(sessionKey) } else { @@ -1211,7 +1567,7 @@ func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey stri al.summarizeFailures.Store(sessionKey, count) const maxSummarizeFailures = 3 - const emergencyKeep = 10 + emergencyKeep := al.continuityRetentionPolicy().FailureKeepMessages if count >= maxSummarizeFailures { logger.ErrorCF("agent", "Summarization failed repeatedly, force-truncating session", map[string]interface{}{ @@ -1228,16 +1584,17 @@ func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey stri // summarizeBatch summarizes a batch of messages using the Fantasy LanguageModel directly. func (al *AgentLoop) summarizeBatch(ctx context.Context, batch []messages.Message, existingSummary string) (string, error) { - prompt := "Provide a concise summary of this conversation segment, preserving core context and key points.\n" + var prompt strings.Builder + prompt.WriteString("Provide a concise summary of this conversation segment, preserving core context and key points.\n") if existingSummary != "" { - prompt += "Existing context: " + existingSummary + "\n" + fmt.Fprintf(&prompt, "Existing context: %s\n", existingSummary) } - prompt += "\nCONVERSATION:\n" + prompt.WriteString("\nCONVERSATION:\n") for _, m := range batch { - prompt += fmt.Sprintf("%s: %s\n", m.Role, m.Content) + fmt.Fprintf(&prompt, "%s: %s\n", m.Role, m.Content) } - return al.callModel(ctx, prompt) + return al.callModel(ctx, prompt.String()) } // callModel makes a direct call to the Fantasy LanguageModel (no tools, no agent loop). @@ -1273,7 +1630,8 @@ func (al *AgentLoop) sessionsToMessagePairs(sessionKey string) []observation.Mes // 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. -func (al *AgentLoop) applyDAGCompression(history []messages.Message) []messages.Message { +// When memDelegate implements dag.DAGPersister, the DAG is persisted for dag_expand/describe/grep. +func (al *AgentLoop) applyDAGCompression(ctx context.Context, sessionKey string, history []messages.Message) []messages.Message { const minHistoryForDAG = 16 if len(history) < minHistoryForDAG { @@ -1314,6 +1672,19 @@ func (al *AgentLoop) applyDAGCompression(history []messages.Message) []messages. rendered := dag.RenderDAGForBudget(d, budget.DAGSummaries) al.contextBuilder.SetDAGBlock(rendered) + // Persist DAG for dag_expand, dag_describe, dag_grep (additive; in-memory behavior unchanged) + if dp, ok := al.memDelegate.(dag.DAGPersister); ok { + if err := dp.PersistDAG(ctx, "dragonscale", sessionKey, &dag.PersistSnapshot{ + FromMsgIdx: 0, + ToMsgIdx: len(compressible), + MsgCount: len(compressible), + DAG: d, + }); err != nil { + logger.WarnCF("agent", "DAG persist failed (non-fatal)", + map[string]interface{}{"error": err.Error(), "session_key": sessionKey}) + } + } + logger.DebugCF("agent", "DAG compression applied", map[string]interface{}{ "total_msgs": len(history), @@ -1333,38 +1704,12 @@ func listConfiguredModels(cfg *config.Config) string { return "No configuration available." } - type entry struct { - name string - key string - } - - // Ordered list of well-known providers. - candidates := []entry{ - {"anthropic", cfg.Providers.Anthropic.APIKey}, - {"openai", cfg.Providers.OpenAI.APIKey}, - {"openrouter", cfg.Providers.OpenRouter.APIKey}, - {"gemini", cfg.Providers.Gemini.APIKey}, - {"groq", cfg.Providers.Groq.APIKey}, - {"zhipu", cfg.Providers.Zhipu.APIKey}, - {"deepseek", cfg.Providers.DeepSeek.APIKey}, - {"moonshot", cfg.Providers.Moonshot.APIKey}, - {"nvidia", cfg.Providers.Nvidia.APIKey}, - {"shengsuanyun", cfg.Providers.ShengSuanYun.APIKey}, - {"vllm", cfg.Providers.VLLM.APIBase}, // vllm uses base URL, not API key - } - - var configured []string - for _, c := range candidates { - if c.key != "" { - configured = append(configured, c.name) - } - } - current := fmt.Sprintf("Current model: %s", cfg.Agents.Defaults.Model) if cfg.Agents.Defaults.Provider != "" { current += fmt.Sprintf(" (provider: %s)", cfg.Agents.Defaults.Provider) } + configured := cfg.Providers.ConfiguredNames() if len(configured) == 0 { return current + "\nNo providers configured β€” set API keys in config.json or environment variables." } @@ -1373,13 +1718,17 @@ func listConfiguredModels(cfg *config.Config) string { } func (al *AgentLoop) estimateTokens(msgs []messages.Message) int { - totalChars := 0 + pairs := make([]observation.MessagePair, 0, len(msgs)) for _, m := range msgs { - totalChars += utf8.RuneCountInString(m.Content) + pairs = append(pairs, observation.MessagePair{ + Role: m.Role, + Content: m.Content, + }) } - return totalChars * 2 / 5 + return observation.EstimateMessagesTokens(pairs) } +// FIXME: Leverage Cobra with subcommand command palette pattern for commands func (al *AgentLoop) handleCommand(_ context.Context, msg bus.InboundMessage) (string, bool) { content := strings.TrimSpace(msg.Content) if !strings.HasPrefix(content, "/") { @@ -1441,7 +1790,7 @@ func (al *AgentLoop) handleCommand(_ context.Context, msg bus.InboundMessage) (s al.model = value return fmt.Sprintf("Switched model from %s to %s", oldModel, value), true case "channel": - // This changes the 'default' channel for some operations, or effectively redirects output? + // FIXME: This changes the 'default' channel for some operations, or effectively redirects output? // For now, let's just validate if the channel exists if al.channelManager == nil { return "Channel manager not initialized", true @@ -1450,7 +1799,7 @@ func (al *AgentLoop) handleCommand(_ context.Context, msg bus.InboundMessage) (s return fmt.Sprintf("Channel '%s' not found or not enabled", value), true } - // If message came from CLI, maybe we want to redirect CLI output to this channel? + // FIXME: If message came from CLI, maybe we want to redirect CLI output to this channel? // That would require state persistence about "redirected channel" // For now, just acknowledged. return fmt.Sprintf("Switched target channel to %s (Note: this currently only validates existence)", value), true diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 93a2b4074..9fbc4884b 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -2,20 +2,29 @@ package agent import ( "context" + "encoding/json" "fmt" "os" + "path/filepath" + "strings" + "sync" "testing" "time" fantasy "charm.land/fantasy" - "github.com/sipeed/picoclaw/pkg/bus" - "github.com/sipeed/picoclaw/pkg/config" - "github.com/sipeed/picoclaw/pkg/tools" + "github.com/ZanzyTHEbar/dragonscale/pkg/bus" + "github.com/ZanzyTHEbar/dragonscale/pkg/config" + memsqlc "github.com/ZanzyTHEbar/dragonscale/pkg/memory/sqlc" + "github.com/ZanzyTHEbar/dragonscale/pkg/messages" + "github.com/ZanzyTHEbar/dragonscale/pkg/tools" ) // mustNewAgentLoop wraps NewAgentLoop and fails the test on error. func mustNewAgentLoop(t *testing.T, cfg *config.Config, msgBus *bus.MessageBus, model fantasy.LanguageModel) *AgentLoop { t.Helper() + if cfg != nil && cfg.Memory.DBPath == "" && strings.TrimSpace(cfg.Agents.Defaults.Workspace) != "" { + cfg.Memory.DBPath = filepath.Join(cfg.Agents.Defaults.Workspace, "agent-loop-test.db") + } al, err := NewAgentLoop(context.Background(), cfg, msgBus, model) if err != nil { t.Fatalf("NewAgentLoop: %v", err) @@ -62,6 +71,125 @@ func (m *mockLanguageModel) StreamObject(_ context.Context, _ fantasy.ObjectCall func (m *mockLanguageModel) Provider() string { return "mock" } func (m *mockLanguageModel) Model() string { return "mock-model" } +func TestContinuityKeepCount_UsesConfiguredPolicy(t *testing.T) { + buildHistory := func(n int, content string) []messages.Message { + history := make([]messages.Message, 0, n) + for i := 0; i < n; i++ { + role := "user" + if i%2 == 1 { + role = "assistant" + } + history = append(history, messages.Message{ + Role: role, + Content: content, + }) + } + return history + } + + cfg := config.DefaultConfig() + cfg.Agents.Defaults.ContinuityRetention.MinMessages = 3 + cfg.Agents.Defaults.ContinuityRetention.MaxMessages = 7 + cfg.Agents.Defaults.ContinuityRetention.TargetContextRatio = 0.01 + + al := &AgentLoop{ + cfg: cfg, + contextWindow: 256, + } + + history := buildHistory(24, strings.Repeat("long message token payload ", 12)) + keepSmallBudget := al.continuityKeepCount(history) + if keepSmallBudget != 3 { + t.Fatalf("expected keep count to respect min_messages=3 under tight budget, got %d", keepSmallBudget) + } + + cfg.Agents.Defaults.ContinuityRetention.TargetContextRatio = 0.40 + al.contextWindow = 8192 + keepLargeBudget := al.continuityKeepCount(history) + if keepLargeBudget != 7 { + t.Fatalf("expected keep count to cap at max_messages=7 under large budget, got %d", keepLargeBudget) + } +} + +func TestPrepareRuntimeState_ConcurrentSameSessionUsesSingleConversation(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + model := newMockLanguageModel("") + al := mustNewAgentLoop(t, cfg, msgBus, model) + beforeConversations, err := al.queries.ListAgentConversations(context.Background(), memsqlc.ListAgentConversationsParams{ + Limit: 10000, + }) + if err != nil { + t.Fatalf("ListAgentConversations (before) failed: %v", err) + } + beforeCount := len(beforeConversations) + + const workers = 12 + start := make(chan struct{}) + var wg sync.WaitGroup + conversationIDs := make(chan string, workers) + errorsCh := make(chan error, workers) + + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + conversationID, _, prepareErr := al.prepareRuntimeState(context.Background(), "race-session") + if prepareErr != nil { + errorsCh <- prepareErr + return + } + conversationIDs <- conversationID.String() + }() + } + + close(start) + wg.Wait() + close(errorsCh) + close(conversationIDs) + + for prepareErr := range errorsCh { + if prepareErr != nil { + t.Fatalf("unexpected prepareRuntimeState error: %v", prepareErr) + } + } + + uniqueConversationIDs := make(map[string]struct{}) + for id := range conversationIDs { + uniqueConversationIDs[id] = struct{}{} + } + if len(uniqueConversationIDs) != 1 { + t.Fatalf("expected one conversation id, got %d (%v)", len(uniqueConversationIDs), uniqueConversationIDs) + } + + conversations, err := al.queries.ListAgentConversations(context.Background(), memsqlc.ListAgentConversationsParams{ + Limit: 10000, + }) + if err != nil { + t.Fatalf("ListAgentConversations failed: %v", err) + } + if len(conversations) != beforeCount+1 { + t.Fatalf("expected conversation count delta +1, got before=%d after=%d", beforeCount, len(conversations)) + } +} + func TestRecordLastChannel(t *testing.T) { // Create temp workspace tmpDir, err := os.MkdirTemp("", "agent-test-*") @@ -183,6 +311,36 @@ func TestNewAgentLoop_StateInitialized(t *testing.T) { } } +func TestNewAgentLoop_UnifiedKernelDependenciesInitialized(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + model := newMockLanguageModel("") + al := mustNewAgentLoop(t, cfg, msgBus, model) + + if !al.HasSecureBus() { + t.Fatal("Expected secure bus to be configured") + } + if !al.HasUnifiedRuntimeDeps() { + t.Fatal("Expected unified runtime dependencies to be configured") + } +} + // TestToolRegistry_ToolRegistration verifies tools can be registered and retrieved func TestToolRegistry_ToolRegistration(t *testing.T) { tmpDir, err := os.MkdirTemp("", "agent-test-*") @@ -613,3 +771,161 @@ func TestResolveFinalContent_RecoversFromToolResultText(t *testing.T) { t.Fatalf("expected tool result text, got %q", got) } } + +// TestForceCompression_PersistsProvenance verifies that emergency compression +// cycles persist provenance metadata to the audit log for postmortem. +func TestForceCompression_PersistsProvenance(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-provenance-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + // Small context window so we can exceed 95% threshold with modest history + // 1000 * 0.95 = 950 tokens; estimateTokens = chars*2/5, so need chars > 2375 + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 1000, + MaxToolIterations: 10, + ContinuityRetention: config.ContinuityRetentionConfig{ + MinMessages: 3, + MaxMessages: 8, + TargetContextRatio: 0.05, + FailureKeepMessages: 8, + }, + }, + }, + } + + msgBus := bus.NewMessageBus() + model := newMockLanguageModel("Summary of conversation.") + al := mustNewAgentLoop(t, cfg, msgBus, model) + al.contextWindow = 1000 + sessionKey := "provenance-test-session" + + // Seed history to exceed critical threshold by a wide margin so this test + // stays deterministic across tokenizer/estimator behavior changes. + const charsPerMsg = 900 + for i := 0; i < 16; i++ { + content := fmt.Sprintf("user message %d: %s", i, strings.Repeat("x", charsPerMsg-20)) + al.sessions.AddMessage(sessionKey, "user", content) + al.sessions.AddMessage(sessionKey, "assistant", "short reply") + } + al.sessions.Save(sessionKey) + history := al.sessions.GetHistory(sessionKey) + keep := al.continuityKeepCount(history) + if len(history) <= keep { + t.Fatalf("test precondition failed: history=%d keep=%d", len(history), keep) + } + tokenEstimate := al.estimateTokens(history) + criticalThreshold := al.contextWindow * 95 / 100 + if tokenEstimate <= criticalThreshold { + t.Fatalf("test precondition failed: token_estimate=%d threshold=%d", tokenEstimate, criticalThreshold) + } + + ctx := context.Background() + al.forceCompression(ctx, sessionKey) + + del := al.MemoryDelegate() + if del == nil { + t.Fatal("MemoryDelegate is nil") + } + entries, err := del.ListAuditEntriesByAction(ctx, "dragonscale", "emergency_compression", 50) + if err != nil { + t.Fatalf("ListAuditEntriesByAction: %v", err) + } + matching := make([]EmergencyProvenance, 0, len(entries)) + for _, entry := range entries { + var prov EmergencyProvenance + if err := json.Unmarshal([]byte(entry.Input), &prov); err != nil { + continue + } + if prov.SessionKey == sessionKey { + matching = append(matching, prov) + } + } + if len(matching) == 0 { + t.Fatal("Expected at least one emergency_compression audit entry") + } + + // Verify metadata shape: session_key, cycle, token_estimate, critical_budget + prov := matching[0] + if prov.SessionKey != sessionKey { + t.Errorf("session_key: want %q, got %q", sessionKey, prov.SessionKey) + } + if prov.Cycle < 1 || prov.Cycle > 3 { + t.Errorf("cycle: want 1..3, got %d", prov.Cycle) + } + if prov.TokenEstimate <= 0 { + t.Errorf("token_estimate: want > 0, got %d", prov.TokenEstimate) + } + if prov.CriticalBudget != 950 { + t.Errorf("critical_budget: want 950, got %d", prov.CriticalBudget) + } + if prov.HistoryMsgCount < 8 { + t.Errorf("history_msg_count: want >= 8, got %d", prov.HistoryMsgCount) + } +} + +func TestPersistOversizedRecoveryRefs_CreatesRecoverableReferences(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "agent-recovery-ref-test-*") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + Model: "test-model", + MaxTokens: 2048, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + model := newMockLanguageModel("ok") + al := mustNewAgentLoop(t, cfg, msgBus, model) + + omitted := []oversizedRecoveryCandidate{ + { + Message: messages.Message{ + Role: "user", + Content: "omitted oversized content for recovery", + }, + OriginalIndex: 2, + TokenEstimate: 9999, + }, + } + + refs, err := al.persistOversizedRecoveryRefs(context.Background(), "recovery-session", omitted) + if err != nil { + t.Fatalf("persistOversizedRecoveryRefs failed: %v", err) + } + if len(refs) != 1 { + t.Fatalf("expected one recovery ref, got %d", len(refs)) + } + + dagTool := tools.NewDagExpandTool(tools.DAGToolDeps{ + Delegate: al.MemoryDelegate(), + AgentID: "dragonscale", + SessionFn: func() string { + return "recovery-session" + }, + }) + res := dagTool.Execute(context.Background(), map[string]interface{}{ + "node_id": refs[0], + "session_key": "recovery-session", + }) + if res.IsError { + t.Fatalf("expected recovery ref expansion to succeed, got: %s", res.ForLLM) + } + if !strings.Contains(res.ForLLM, "omitted oversized content for recovery") { + t.Fatalf("expected recovered content in output, got: %s", res.ForLLM) + } +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 8d77ef2dc..5c257c800 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -61,17 +61,17 @@ type Config struct { // Memory is always enabled; there is no opt-out. Configuration controls // the database path, embedding dimensions, offloading threshold, and sync. type MemoryConfig struct { - // DBPath overrides the default database path (workspace/memory/picoclaw.db). + // DBPath overrides the default database path (workspace/memory/dragonscale.db). // Empty string uses the default. - DBPath string `json:"db_path" env:"PICOCLAW_MEMORY_DB_PATH"` + DBPath string `json:"db_path" env:"DRAGONSCALE_MEMORY_DB_PATH"` // EmbeddingDims is the vector dimensionality for archival embeddings. // Default: 768 (sentence-transformers). Use 1536 for OpenAI ada-002, 384 for MiniLM. - EmbeddingDims int `json:"embedding_dims" env:"PICOCLAW_MEMORY_EMBEDDING_DIMS"` + EmbeddingDims int `json:"embedding_dims" env:"DRAGONSCALE_MEMORY_EMBEDDING_DIMS"` // OffloadThresholdTokens is the token count above which tool results // are automatically offloaded to archival memory. Default: 4000. - OffloadThresholdTokens int `json:"offload_threshold_tokens" env:"PICOCLAW_MEMORY_OFFLOAD_THRESHOLD_TOKENS"` + OffloadThresholdTokens int `json:"offload_threshold_tokens" env:"DRAGONSCALE_MEMORY_OFFLOAD_THRESHOLD_TOKENS"` // Embedding configures the embedding provider for archival vector search. Embedding EmbeddingConfig `json:"embedding"` @@ -85,21 +85,21 @@ type MemoryConfig struct { type EmbeddingConfig struct { // Provider selects the embedding backend: "ollama", "openai", or "". // Empty string disables embeddings (FTS5-only search). - Provider string `json:"provider" env:"PICOCLAW_MEMORY_EMBEDDING_PROVIDER"` + Provider string `json:"provider" env:"DRAGONSCALE_MEMORY_EMBEDDING_PROVIDER"` // Model is the embedding model name (e.g., "nomic-embed-text", "text-embedding-3-small"). // Defaults depend on provider: "nomic-embed-text" for Ollama, "text-embedding-3-small" for OpenAI. - Model string `json:"model" env:"PICOCLAW_MEMORY_EMBEDDING_MODEL"` + Model string `json:"model" env:"DRAGONSCALE_MEMORY_EMBEDDING_MODEL"` // APIBase overrides the provider's API base URL. // For Ollama defaults to "http://localhost:11434". // For OpenAI defaults to "https://api.openai.com/v1". // Empty string uses the default for the selected provider. - APIBase string `json:"api_base" env:"PICOCLAW_MEMORY_EMBEDDING_API_BASE"` + APIBase string `json:"api_base" env:"DRAGONSCALE_MEMORY_EMBEDDING_API_BASE"` // APIKey for the embedding provider. Required for OpenAI, optional for Ollama. // If empty, falls back to the matching provider's key from providers config. - APIKey string `json:"api_key" env:"PICOCLAW_MEMORY_EMBEDDING_API_KEY"` + APIKey string `json:"api_key" env:"DRAGONSCALE_MEMORY_EMBEDDING_API_KEY"` } // MemorySyncConfig configures Turso embedded replica synchronization. @@ -107,38 +107,57 @@ type EmbeddingConfig struct { type MemorySyncConfig struct { // SyncURL is the Turso primary database URL (e.g., "libsql://mydb.turso.io"). // Empty string disables replication (local-only mode). - SyncURL string `json:"sync_url" env:"PICOCLAW_MEMORY_SYNC_URL"` + SyncURL string `json:"sync_url" env:"DRAGONSCALE_MEMORY_SYNC_URL"` // AuthToken is the Turso authentication token for the remote database. - AuthToken string `json:"auth_token" env:"PICOCLAW_MEMORY_SYNC_AUTH_TOKEN"` + AuthToken string `json:"auth_token" env:"DRAGONSCALE_MEMORY_SYNC_AUTH_TOKEN"` // SyncIntervalSeconds is how often to sync with the remote primary (in seconds). // Zero means manual sync only. Default: 60. - SyncIntervalSeconds int `json:"sync_interval_seconds" env:"PICOCLAW_MEMORY_SYNC_INTERVAL_SECONDS"` + SyncIntervalSeconds int `json:"sync_interval_seconds" env:"DRAGONSCALE_MEMORY_SYNC_INTERVAL_SECONDS"` // EncryptionKey enables encryption-at-rest on the local database file. // Empty string means no encryption. - EncryptionKey string `json:"encryption_key" env:"PICOCLAW_MEMORY_SYNC_ENCRYPTION_KEY"` + EncryptionKey string `json:"encryption_key" env:"DRAGONSCALE_MEMORY_SYNC_ENCRYPTION_KEY"` } type AgentsConfig struct { Defaults AgentDefaults `json:"defaults"` } +type ContinuityRetentionConfig struct { + // MinMessages is the minimum number of recent messages always retained + // unsummarized for conversational continuity. + MinMessages int `json:"min_messages" env:"DRAGONSCALE_AGENTS_DEFAULTS_CONTINUITY_RETENTION_MIN_MESSAGES"` + + // MaxMessages is the upper bound on retained recent messages, even when + // the token budget would allow more. + MaxMessages int `json:"max_messages" env:"DRAGONSCALE_AGENTS_DEFAULTS_CONTINUITY_RETENTION_MAX_MESSAGES"` + + // TargetContextRatio is the target fraction of model context window reserved + // for retained recent messages. + TargetContextRatio float64 `json:"target_context_ratio" env:"DRAGONSCALE_AGENTS_DEFAULTS_CONTINUITY_RETENTION_TARGET_CONTEXT_RATIO"` + + // FailureKeepMessages is the fallback retained-message count used when + // summarization repeatedly fails. + FailureKeepMessages int `json:"failure_keep_messages" env:"DRAGONSCALE_AGENTS_DEFAULTS_CONTINUITY_RETENTION_FAILURE_KEEP_MESSAGES"` +} + type AgentDefaults struct { // Sandbox is the directory for agent file operations (tools sandbox). - // Defaults to $XDG_DATA_HOME/picoclaw/sandbox when empty. - Sandbox string `json:"sandbox" env:"PICOCLAW_AGENTS_DEFAULTS_SANDBOX"` - RestrictToSandbox bool `json:"restrict_to_sandbox" env:"PICOCLAW_AGENTS_DEFAULTS_RESTRICT_TO_SANDBOX"` - Provider string `json:"provider" env:"PICOCLAW_AGENTS_DEFAULTS_PROVIDER"` - Model string `json:"model" env:"PICOCLAW_AGENTS_DEFAULTS_MODEL"` - MaxTokens int `json:"max_tokens" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_TOKENS"` - Temperature float64 `json:"temperature" env:"PICOCLAW_AGENTS_DEFAULTS_TEMPERATURE"` - MaxToolIterations int `json:"max_tool_iterations" env:"PICOCLAW_AGENTS_DEFAULTS_MAX_TOOL_ITERATIONS"` + // Defaults to $XDG_DATA_HOME/dragonscale/sandbox when empty. + Sandbox string `json:"sandbox" env:"DRAGONSCALE_AGENTS_DEFAULTS_SANDBOX"` + RestrictToSandbox bool `json:"restrict_to_sandbox" env:"DRAGONSCALE_AGENTS_DEFAULTS_RESTRICT_TO_SANDBOX"` + Provider string `json:"provider" env:"DRAGONSCALE_AGENTS_DEFAULTS_PROVIDER"` + Model string `json:"model" env:"DRAGONSCALE_AGENTS_DEFAULTS_MODEL"` + MaxTokens int `json:"max_tokens" env:"DRAGONSCALE_AGENTS_DEFAULTS_MAX_TOKENS"` + Temperature float64 `json:"temperature" env:"DRAGONSCALE_AGENTS_DEFAULTS_TEMPERATURE"` + MaxToolIterations int `json:"max_tool_iterations" env:"DRAGONSCALE_AGENTS_DEFAULTS_MAX_TOOL_ITERATIONS"` + ContinuityRetention ContinuityRetentionConfig `json:"continuity_retention"` // Deprecated: Use Sandbox instead. Kept for backward compatibility during migration. - Workspace string `json:"workspace,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_WORKSPACE"` - RestrictToWorkspace bool `json:"restrict_to_workspace,omitempty" env:"PICOCLAW_AGENTS_DEFAULTS_RESTRICT_TO_WORKSPACE"` + Workspace string `json:"workspace,omitempty" env:"DRAGONSCALE_AGENTS_DEFAULTS_WORKSPACE"` + RestrictToWorkspace bool `json:"restrict_to_workspace,omitempty" env:"DRAGONSCALE_AGENTS_DEFAULTS_RESTRICT_TO_WORKSPACE"` } type ChannelsConfig struct { @@ -155,88 +174,88 @@ type ChannelsConfig struct { } type WhatsAppConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_WHATSAPP_ENABLED"` - BridgeURL string `json:"bridge_url" env:"PICOCLAW_CHANNELS_WHATSAPP_BRIDGE_URL"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_WHATSAPP_ALLOW_FROM"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_CHANNELS_WHATSAPP_ENABLED"` + BridgeURL string `json:"bridge_url" env:"DRAGONSCALE_CHANNELS_WHATSAPP_BRIDGE_URL"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"DRAGONSCALE_CHANNELS_WHATSAPP_ALLOW_FROM"` } type TelegramConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_TELEGRAM_ENABLED"` - Token string `json:"token" env:"PICOCLAW_CHANNELS_TELEGRAM_TOKEN"` - Proxy string `json:"proxy" env:"PICOCLAW_CHANNELS_TELEGRAM_PROXY"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_TELEGRAM_ALLOW_FROM"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_CHANNELS_TELEGRAM_ENABLED"` + Token string `json:"token" env:"DRAGONSCALE_CHANNELS_TELEGRAM_TOKEN"` + Proxy string `json:"proxy" env:"DRAGONSCALE_CHANNELS_TELEGRAM_PROXY"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"DRAGONSCALE_CHANNELS_TELEGRAM_ALLOW_FROM"` } type FeishuConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_FEISHU_ENABLED"` - AppID string `json:"app_id" env:"PICOCLAW_CHANNELS_FEISHU_APP_ID"` - AppSecret string `json:"app_secret" env:"PICOCLAW_CHANNELS_FEISHU_APP_SECRET"` - EncryptKey string `json:"encrypt_key" env:"PICOCLAW_CHANNELS_FEISHU_ENCRYPT_KEY"` - VerificationToken string `json:"verification_token" env:"PICOCLAW_CHANNELS_FEISHU_VERIFICATION_TOKEN"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_FEISHU_ALLOW_FROM"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_CHANNELS_FEISHU_ENABLED"` + AppID string `json:"app_id" env:"DRAGONSCALE_CHANNELS_FEISHU_APP_ID"` + AppSecret string `json:"app_secret" env:"DRAGONSCALE_CHANNELS_FEISHU_APP_SECRET"` + EncryptKey string `json:"encrypt_key" env:"DRAGONSCALE_CHANNELS_FEISHU_ENCRYPT_KEY"` + VerificationToken string `json:"verification_token" env:"DRAGONSCALE_CHANNELS_FEISHU_VERIFICATION_TOKEN"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"DRAGONSCALE_CHANNELS_FEISHU_ALLOW_FROM"` } type DiscordConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_DISCORD_ENABLED"` - Token string `json:"token" env:"PICOCLAW_CHANNELS_DISCORD_TOKEN"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_DISCORD_ALLOW_FROM"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_CHANNELS_DISCORD_ENABLED"` + Token string `json:"token" env:"DRAGONSCALE_CHANNELS_DISCORD_TOKEN"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"DRAGONSCALE_CHANNELS_DISCORD_ALLOW_FROM"` } type MaixCamConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_MAIXCAM_ENABLED"` - Host string `json:"host" env:"PICOCLAW_CHANNELS_MAIXCAM_HOST"` - Port int `json:"port" env:"PICOCLAW_CHANNELS_MAIXCAM_PORT"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_MAIXCAM_ALLOW_FROM"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_CHANNELS_MAIXCAM_ENABLED"` + Host string `json:"host" env:"DRAGONSCALE_CHANNELS_MAIXCAM_HOST"` + Port int `json:"port" env:"DRAGONSCALE_CHANNELS_MAIXCAM_PORT"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"DRAGONSCALE_CHANNELS_MAIXCAM_ALLOW_FROM"` } type QQConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_QQ_ENABLED"` - AppID string `json:"app_id" env:"PICOCLAW_CHANNELS_QQ_APP_ID"` - AppSecret string `json:"app_secret" env:"PICOCLAW_CHANNELS_QQ_APP_SECRET"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_QQ_ALLOW_FROM"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_CHANNELS_QQ_ENABLED"` + AppID string `json:"app_id" env:"DRAGONSCALE_CHANNELS_QQ_APP_ID"` + AppSecret string `json:"app_secret" env:"DRAGONSCALE_CHANNELS_QQ_APP_SECRET"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"DRAGONSCALE_CHANNELS_QQ_ALLOW_FROM"` } type DingTalkConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_DINGTALK_ENABLED"` - ClientID string `json:"client_id" env:"PICOCLAW_CHANNELS_DINGTALK_CLIENT_ID"` - ClientSecret string `json:"client_secret" env:"PICOCLAW_CHANNELS_DINGTALK_CLIENT_SECRET"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_DINGTALK_ALLOW_FROM"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_CHANNELS_DINGTALK_ENABLED"` + ClientID string `json:"client_id" env:"DRAGONSCALE_CHANNELS_DINGTALK_CLIENT_ID"` + ClientSecret string `json:"client_secret" env:"DRAGONSCALE_CHANNELS_DINGTALK_CLIENT_SECRET"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"DRAGONSCALE_CHANNELS_DINGTALK_ALLOW_FROM"` } type SlackConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_SLACK_ENABLED"` - BotToken string `json:"bot_token" env:"PICOCLAW_CHANNELS_SLACK_BOT_TOKEN"` - AppToken string `json:"app_token" env:"PICOCLAW_CHANNELS_SLACK_APP_TOKEN"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_SLACK_ALLOW_FROM"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_CHANNELS_SLACK_ENABLED"` + BotToken string `json:"bot_token" env:"DRAGONSCALE_CHANNELS_SLACK_BOT_TOKEN"` + AppToken string `json:"app_token" env:"DRAGONSCALE_CHANNELS_SLACK_APP_TOKEN"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"DRAGONSCALE_CHANNELS_SLACK_ALLOW_FROM"` } type LINEConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_LINE_ENABLED"` - ChannelSecret string `json:"channel_secret" env:"PICOCLAW_CHANNELS_LINE_CHANNEL_SECRET"` - ChannelAccessToken string `json:"channel_access_token" env:"PICOCLAW_CHANNELS_LINE_CHANNEL_ACCESS_TOKEN"` - WebhookHost string `json:"webhook_host" env:"PICOCLAW_CHANNELS_LINE_WEBHOOK_HOST"` - WebhookPort int `json:"webhook_port" env:"PICOCLAW_CHANNELS_LINE_WEBHOOK_PORT"` - WebhookPath string `json:"webhook_path" env:"PICOCLAW_CHANNELS_LINE_WEBHOOK_PATH"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_LINE_ALLOW_FROM"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_CHANNELS_LINE_ENABLED"` + ChannelSecret string `json:"channel_secret" env:"DRAGONSCALE_CHANNELS_LINE_CHANNEL_SECRET"` + ChannelAccessToken string `json:"channel_access_token" env:"DRAGONSCALE_CHANNELS_LINE_CHANNEL_ACCESS_TOKEN"` + WebhookHost string `json:"webhook_host" env:"DRAGONSCALE_CHANNELS_LINE_WEBHOOK_HOST"` + WebhookPort int `json:"webhook_port" env:"DRAGONSCALE_CHANNELS_LINE_WEBHOOK_PORT"` + WebhookPath string `json:"webhook_path" env:"DRAGONSCALE_CHANNELS_LINE_WEBHOOK_PATH"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"DRAGONSCALE_CHANNELS_LINE_ALLOW_FROM"` } type OneBotConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_ONEBOT_ENABLED"` - WSUrl string `json:"ws_url" env:"PICOCLAW_CHANNELS_ONEBOT_WS_URL"` - AccessToken string `json:"access_token" env:"PICOCLAW_CHANNELS_ONEBOT_ACCESS_TOKEN"` - ReconnectInterval int `json:"reconnect_interval" env:"PICOCLAW_CHANNELS_ONEBOT_RECONNECT_INTERVAL"` - GroupTriggerPrefix []string `json:"group_trigger_prefix" env:"PICOCLAW_CHANNELS_ONEBOT_GROUP_TRIGGER_PREFIX"` - AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_ONEBOT_ALLOW_FROM"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_CHANNELS_ONEBOT_ENABLED"` + WSUrl string `json:"ws_url" env:"DRAGONSCALE_CHANNELS_ONEBOT_WS_URL"` + AccessToken string `json:"access_token" env:"DRAGONSCALE_CHANNELS_ONEBOT_ACCESS_TOKEN"` + ReconnectInterval int `json:"reconnect_interval" env:"DRAGONSCALE_CHANNELS_ONEBOT_RECONNECT_INTERVAL"` + GroupTriggerPrefix []string `json:"group_trigger_prefix" env:"DRAGONSCALE_CHANNELS_ONEBOT_GROUP_TRIGGER_PREFIX"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"DRAGONSCALE_CHANNELS_ONEBOT_ALLOW_FROM"` } type HeartbeatConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_HEARTBEAT_ENABLED"` - Interval int `json:"interval" env:"PICOCLAW_HEARTBEAT_INTERVAL"` // minutes, min 5 + Enabled bool `json:"enabled" env:"DRAGONSCALE_HEARTBEAT_ENABLED"` + Interval int `json:"interval" env:"DRAGONSCALE_HEARTBEAT_INTERVAL"` // minutes, min 5 } type DevicesConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_DEVICES_ENABLED"` - MonitorUSB bool `json:"monitor_usb" env:"PICOCLAW_DEVICES_MONITOR_USB"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_DEVICES_ENABLED"` + MonitorUSB bool `json:"monitor_usb" env:"DRAGONSCALE_DEVICES_MONITOR_USB"` } type ProvidersConfig struct { @@ -255,40 +274,70 @@ type ProvidersConfig struct { GitHubCopilot ProviderConfig `json:"github_copilot"` } +// ConfiguredNames returns the names of providers that have credentials set +// (either an API key or an API base URL for local inference servers). +func (p ProvidersConfig) ConfiguredNames() []string { + entries := []struct { + name string + key string + }{ + {"anthropic", p.Anthropic.APIKey}, + {"openai", p.OpenAI.APIKey}, + {"openrouter", p.OpenRouter.APIKey}, + {"gemini", p.Gemini.APIKey}, + {"groq", p.Groq.APIKey}, + {"zhipu", p.Zhipu.APIKey}, + {"deepseek", p.DeepSeek.APIKey}, + {"moonshot", p.Moonshot.APIKey}, + {"nvidia", p.Nvidia.APIKey}, + {"shengsuanyun", p.ShengSuanYun.APIKey}, + {"ollama", p.Ollama.APIBase}, + {"vllm", p.VLLM.APIBase}, + {"github_copilot", p.GitHubCopilot.APIKey}, + } + var names []string + for _, e := range entries { + if e.key != "" { + names = append(names, e.name) + } + } + return names +} + type ProviderConfig struct { - APIKey string `json:"api_key" env:"PICOCLAW_PROVIDERS_{{.Name}}_API_KEY"` - APIBase string `json:"api_base" env:"PICOCLAW_PROVIDERS_{{.Name}}_API_BASE"` - Proxy string `json:"proxy,omitzero" env:"PICOCLAW_PROVIDERS_{{.Name}}_PROXY"` - AuthMethod string `json:"auth_method,omitzero" env:"PICOCLAW_PROVIDERS_{{.Name}}_AUTH_METHOD"` - Timeout int `json:"timeout,omitzero" env:"PICOCLAW_PROVIDERS_{{.Name}}_TIMEOUT"` // seconds, 0 = default (120s) - ConnectMode string `json:"connect_mode,omitzero" env:"PICOCLAW_PROVIDERS_{{.Name}}_CONNECT_MODE"` // only for Github Copilot, `stdio` or `grpc` + APIKey string `json:"api_key" env:"DRAGONSCALE_PROVIDERS_{{.Name}}_API_KEY"` + APIBase string `json:"api_base" env:"DRAGONSCALE_PROVIDERS_{{.Name}}_API_BASE"` + Proxy string `json:"proxy,omitzero" env:"DRAGONSCALE_PROVIDERS_{{.Name}}_PROXY"` + AuthMethod string `json:"auth_method,omitzero" env:"DRAGONSCALE_PROVIDERS_{{.Name}}_AUTH_METHOD"` + Timeout int `json:"timeout,omitzero" env:"DRAGONSCALE_PROVIDERS_{{.Name}}_TIMEOUT"` // seconds, 0 = default (120s) + ConnectMode string `json:"connect_mode,omitzero" env:"DRAGONSCALE_PROVIDERS_{{.Name}}_CONNECT_MODE"` // only for Github Copilot, `stdio` or `grpc` } type OpenAIProviderConfig struct { ProviderConfig - WebSearch bool `json:"web_search" env:"PICOCLAW_PROVIDERS_OPENAI_WEB_SEARCH"` + WebSearch bool `json:"web_search" env:"DRAGONSCALE_PROVIDERS_OPENAI_WEB_SEARCH"` } type GatewayConfig struct { - Host string `json:"host" env:"PICOCLAW_GATEWAY_HOST"` - Port int `json:"port" env:"PICOCLAW_GATEWAY_PORT"` + Host string `json:"host" env:"DRAGONSCALE_GATEWAY_HOST"` + Port int `json:"port" env:"DRAGONSCALE_GATEWAY_PORT"` } type BraveConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_TOOLS_WEB_BRAVE_ENABLED"` - APIKey string `json:"api_key" env:"PICOCLAW_TOOLS_WEB_BRAVE_API_KEY"` - MaxResults int `json:"max_results" env:"PICOCLAW_TOOLS_WEB_BRAVE_MAX_RESULTS"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_TOOLS_WEB_BRAVE_ENABLED"` + APIKey string `json:"api_key" env:"DRAGONSCALE_TOOLS_WEB_BRAVE_API_KEY"` + MaxResults int `json:"max_results" env:"DRAGONSCALE_TOOLS_WEB_BRAVE_MAX_RESULTS"` } type DuckDuckGoConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_TOOLS_WEB_DUCKDUCKGO_ENABLED"` - MaxResults int `json:"max_results" env:"PICOCLAW_TOOLS_WEB_DUCKDUCKGO_MAX_RESULTS"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_TOOLS_WEB_DUCKDUCKGO_ENABLED"` + MaxResults int `json:"max_results" env:"DRAGONSCALE_TOOLS_WEB_DUCKDUCKGO_MAX_RESULTS"` } type PerplexityConfig struct { - Enabled bool `json:"enabled" env:"PICOCLAW_TOOLS_WEB_PERPLEXITY_ENABLED"` - APIKey string `json:"api_key" env:"PICOCLAW_TOOLS_WEB_PERPLEXITY_API_KEY"` - MaxResults int `json:"max_results" env:"PICOCLAW_TOOLS_WEB_PERPLEXITY_MAX_RESULTS"` + Enabled bool `json:"enabled" env:"DRAGONSCALE_TOOLS_WEB_PERPLEXITY_ENABLED"` + APIKey string `json:"api_key" env:"DRAGONSCALE_TOOLS_WEB_PERPLEXITY_API_KEY"` + MaxResults int `json:"max_results" env:"DRAGONSCALE_TOOLS_WEB_PERPLEXITY_MAX_RESULTS"` } type WebToolsConfig struct { @@ -298,7 +347,7 @@ type WebToolsConfig struct { } type CronToolsConfig struct { - ExecTimeoutMinutes int `json:"exec_timeout_minutes" env:"PICOCLAW_TOOLS_CRON_EXEC_TIMEOUT_MINUTES"` // 0 means no timeout + ExecTimeoutMinutes int `json:"exec_timeout_minutes" env:"DRAGONSCALE_TOOLS_CRON_EXEC_TIMEOUT_MINUTES"` // 0 means no timeout } type ToolsConfig struct { @@ -317,6 +366,12 @@ func DefaultConfig() *Config { MaxTokens: 8192, Temperature: 0.7, MaxToolIterations: 20, + ContinuityRetention: ContinuityRetentionConfig{ + MinMessages: 4, + MaxMessages: 24, + TargetContextRatio: 0.10, + FailureKeepMessages: 10, + }, }, }, Channels: ChannelsConfig{ @@ -493,6 +548,23 @@ func (c *Config) Validate() []string { warnings = append(warnings, fmt.Sprintf("agents.defaults.max_tool_iterations=%d: should be > 0", c.Agents.Defaults.MaxToolIterations)) } + continuity := c.Agents.Defaults.ContinuityRetention + if continuity.MinMessages <= 0 { + warnings = append(warnings, fmt.Sprintf("agents.defaults.continuity_retention.min_messages=%d: should be > 0", continuity.MinMessages)) + } + if continuity.MaxMessages <= 0 { + warnings = append(warnings, fmt.Sprintf("agents.defaults.continuity_retention.max_messages=%d: should be > 0", continuity.MaxMessages)) + } + if continuity.MaxMessages > 0 && continuity.MinMessages > continuity.MaxMessages { + warnings = append(warnings, fmt.Sprintf("agents.defaults.continuity_retention.min_messages=%d exceeds max_messages=%d", continuity.MinMessages, continuity.MaxMessages)) + } + if continuity.TargetContextRatio <= 0 || continuity.TargetContextRatio > 0.5 { + warnings = append(warnings, fmt.Sprintf("agents.defaults.continuity_retention.target_context_ratio=%.4f: expected (0, 0.5]", continuity.TargetContextRatio)) + } + if continuity.FailureKeepMessages <= 0 { + warnings = append(warnings, fmt.Sprintf("agents.defaults.continuity_retention.failure_keep_messages=%d: should be > 0", continuity.FailureKeepMessages)) + } + if c.Gateway.Port < 0 || c.Gateway.Port > 65535 { warnings = append(warnings, fmt.Sprintf("gateway.port=%d: must be in range 1-65535", c.Gateway.Port)) } @@ -569,7 +641,7 @@ func (c *Config) SandboxPath() string { if dir, err := SandboxDir(); err == nil { return dir } - return expandHome("~/.local/share/picoclaw/sandbox") + return expandHome("~/.local/share/dragonscale/sandbox") } // RestrictToSandbox returns whether tool file operations should be restricted @@ -602,7 +674,7 @@ func (c *Config) DBPath() string { if p, err := DefaultDBPath(); err == nil { return p } - return expandHome("~/.local/share/picoclaw/picoclaw.db") + return expandHome("~/.local/share/dragonscale/dragonscale.db") } func (c *Config) GetAPIKey() string { @@ -669,12 +741,12 @@ func expandHome(path string) string { // ─── XDG / platform path helpers ───────────────────────────────────────────── -const appName = "picoclaw" +const appName = "dragonscale" // ConfigDir returns the platform-appropriate user configuration directory for -// picoclaw, following XDG Base Directory spec on Linux -// (~/.config/picoclaw), Library/Application Support on macOS, and -// %AppData%\picoclaw on Windows. The directory is created if it does not exist. +// dragonscale, following XDG Base Directory spec on Linux +// (~/.config/dragonscale), Library/Application Support on macOS, and +// %AppData%\dragonscale on Windows. The directory is created if it does not exist. func ConfigDir() (string, error) { base, err := os.UserConfigDir() if err != nil { @@ -687,10 +759,10 @@ func ConfigDir() (string, error) { return dir, nil } -// DataDir returns the platform-appropriate user data directory for picoclaw. -// On Linux this respects XDG_DATA_HOME (default ~/.local/share/picoclaw). -// On macOS it uses ~/Library/Application Support/picoclaw; on Windows -// %LOCALAPPDATA%\picoclaw. The directory is created if it does not exist. +// DataDir returns the platform-appropriate user data directory for dragonscale. +// On Linux this respects XDG_DATA_HOME (default ~/.local/share/dragonscale). +// On macOS it uses ~/Library/Application Support/dragonscale; on Windows +// %LOCALAPPDATA%\dragonscale. The directory is created if it does not exist. func DataDir() (string, error) { var base string switch runtime.GOOS { @@ -765,8 +837,8 @@ func SandboxDir() (string, error) { return dir, nil } -// CacheDir returns the platform-appropriate user cache directory for picoclaw -// (XDG_CACHE_HOME on Linux β†’ ~/.cache/picoclaw). The directory is created if +// CacheDir returns the platform-appropriate user cache directory for dragonscale +// (XDG_CACHE_HOME on Linux β†’ ~/.cache/dragonscale). The directory is created if // it does not exist. func CacheDir() (string, error) { base, err := os.UserCacheDir() @@ -782,7 +854,7 @@ func CacheDir() (string, error) { // DefaultDBPath returns the canonical SQLite database path inside DataDir. // Callers that want to override this should check for a CLI flag or the -// PICOCLAW_DB_PATH environment variable before falling back to this value. +// DRAGONSCALE_DB_PATH environment variable before falling back to this value. func DefaultDBPath() (string, error) { dataDir, err := DataDir() if err != nil { @@ -792,7 +864,7 @@ func DefaultDBPath() (string, error) { } // DefaultConfigPath returns the path to the primary JSON config file inside -// ConfigDir (picoclaw/config.json). +// ConfigDir (dragonscale/config.json). func DefaultConfigPath() (string, error) { cfgDir, err := ConfigDir() if err != nil { diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 27bf27d2d..3a29efd79 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -53,6 +53,23 @@ func TestDefaultConfig_MaxToolIterations(t *testing.T) { } } +func TestDefaultConfig_ContinuityRetention(t *testing.T) { + cfg := DefaultConfig() + + if cfg.Agents.Defaults.ContinuityRetention.MinMessages <= 0 { + t.Error("ContinuityRetention.MinMessages should be > 0") + } + if cfg.Agents.Defaults.ContinuityRetention.MaxMessages < cfg.Agents.Defaults.ContinuityRetention.MinMessages { + t.Error("ContinuityRetention.MaxMessages should be >= MinMessages") + } + if cfg.Agents.Defaults.ContinuityRetention.TargetContextRatio <= 0 { + t.Error("ContinuityRetention.TargetContextRatio should be > 0") + } + if cfg.Agents.Defaults.ContinuityRetention.FailureKeepMessages < cfg.Agents.Defaults.ContinuityRetention.MinMessages { + t.Error("ContinuityRetention.FailureKeepMessages should be >= MinMessages") + } +} + // TestDefaultConfig_Temperature verifies temperature has default value func TestDefaultConfig_Temperature(t *testing.T) { cfg := DefaultConfig() @@ -300,6 +317,26 @@ func TestValidate_MemoryConfig(t *testing.T) { } } +func TestValidate_ContinuityRetentionConfig(t *testing.T) { + cfg := DefaultConfig() + cfg.Agents.Defaults.ContinuityRetention.MinMessages = 8 + cfg.Agents.Defaults.ContinuityRetention.MaxMessages = 4 + cfg.Agents.Defaults.ContinuityRetention.TargetContextRatio = 0 + cfg.Agents.Defaults.ContinuityRetention.FailureKeepMessages = 0 + + warnings := cfg.Validate() + joined := strings.Join(warnings, "\n") + if !strings.Contains(joined, "continuity_retention.min_messages") { + t.Fatalf("expected continuity retention min/max warning, got: %v", warnings) + } + if !strings.Contains(joined, "continuity_retention.target_context_ratio") { + t.Fatalf("expected continuity retention ratio warning, got: %v", warnings) + } + if !strings.Contains(joined, "continuity_retention.failure_keep_messages") { + t.Fatalf("expected continuity retention failure keep warning, got: %v", warnings) + } +} + func containsMemoryWarning(s string) bool { return strings.Contains(s, "memory.") } diff --git a/pkg/tools/spawn_test.go b/pkg/tools/spawn_test.go new file mode 100644 index 000000000..db114e67f --- /dev/null +++ b/pkg/tools/spawn_test.go @@ -0,0 +1,44 @@ +package tools + +import ( + "context" + "strings" + "testing" + + "github.com/ZanzyTHEbar/dragonscale/pkg/bus" +) + +func TestSpawnTool_Execute_NestedDelegationGuardrails(t *testing.T) { + provider := &MockLanguageModel{} + manager := NewSubagentManager(provider, "test-model", "/tmp/test", bus.NewMessageBus()) + manager.SetRunLoop(func(_ context.Context, _ ToolLoopConfig, _, _, _, _ string) (*ToolLoopResult, error) { + return &ToolLoopResult{Content: "ok", Iterations: 1}, nil + }) + + tool := NewSpawnTool(manager) + ctx := withDelegationContext(context.Background(), "parent", 1) + + missingMetadata := tool.Execute(ctx, map[string]interface{}{ + "task": "nested task", + "label": "n1", + }) + if !missingMetadata.IsError { + t.Fatal("expected missing delegated metadata to fail") + } + if !strings.Contains(missingMetadata.ForLLM, "nested delegation requires delegated_scope and kept_work") { + t.Fatalf("unexpected error: %s", missingMetadata.ForLLM) + } + + withMetadata := tool.Execute(ctx, map[string]interface{}{ + "task": "nested task", + "label": "n2", + "delegated_scope": "collect upstream context", + "kept_work": "final answer synthesis", + }) + if withMetadata.IsError { + t.Fatalf("expected nested delegation with metadata to succeed: %s", withMetadata.ForLLM) + } + if !withMetadata.Async { + t.Fatal("expected spawn tool to return async result") + } +} diff --git a/pkg/tools/subagent.go b/pkg/tools/subagent.go index ddedd5f47..3f73da87c 100644 --- a/pkg/tools/subagent.go +++ b/pkg/tools/subagent.go @@ -3,69 +3,150 @@ package tools import ( "context" "fmt" + "strings" "sync" "time" fantasy "charm.land/fantasy" - "github.com/sipeed/picoclaw/pkg/bus" + "github.com/ZanzyTHEbar/dragonscale/pkg/bus" ) type SubagentTask struct { - ID string - Task string - Label string - OriginChannel string - OriginChatID string - Status string - Result string - Created int64 + ID string + ParentTaskID string + Depth int + Task string + Label string + DelegatedScope string + KeptWork string + OriginChannel string + OriginChatID string + Status string + Result string + Created int64 } // RunLoopFunc executes an agent tool loop. Injected from pkg/agent to break // the import cycle between pkg/tools and pkg/fantasy. type RunLoopFunc func(ctx context.Context, config ToolLoopConfig, systemPrompt, userPrompt, channel, chatID string) (*ToolLoopResult, error) +type delegationCtxKey string + +const ( + delegationTaskIDKey delegationCtxKey = "delegation_task_id" + delegationDepthKey delegationCtxKey = "delegation_depth" +) + +func delegationTaskIDFromContext(ctx context.Context) string { + if v, ok := ctx.Value(delegationTaskIDKey).(string); ok { + return v + } + return "" +} + +func delegationDepthFromContext(ctx context.Context) int { + if v, ok := ctx.Value(delegationDepthKey).(int); ok { + return v + } + return 0 +} + +func withDelegationContext(ctx context.Context, taskID string, depth int) context.Context { + ctx = context.WithValue(ctx, delegationTaskIDKey, taskID) + ctx = context.WithValue(ctx, delegationDepthKey, depth) + return ctx +} + +// DelegationAuditEvent captures lineage and outcomes for delegated work. +type DelegationAuditEvent struct { + TaskID string + ParentTaskID string + Mode string + Depth int + Status string + Label string + DelegatedScope string + KeptWork string + Iterations int + ResultChars int + Error string + OriginChannel string + OriginChatID string +} + type SubagentManager struct { - tasks map[string]*SubagentTask - mu sync.RWMutex - model fantasy.LanguageModel - defaultModel string - bus *bus.MessageBus - workspace string - tools *ToolRegistry - maxIterations int - nextID int - runLoop RunLoopFunc + tasks map[string]*SubagentTask + mu sync.RWMutex + model fantasy.LanguageModel + defaultModel string + bus *bus.MessageBus + workspace string + tools *ToolRegistry + maxIterations int + maxDepth int + maxFanout int + activeChildren map[string]int + nextID int + runLoop RunLoopFunc + auditHook func(context.Context, DelegationAuditEvent) } func NewSubagentManager(model fantasy.LanguageModel, defaultModel, workspace string, bus *bus.MessageBus) *SubagentManager { return &SubagentManager{ - tasks: make(map[string]*SubagentTask), - model: model, - defaultModel: defaultModel, - bus: bus, - workspace: workspace, - tools: NewToolRegistry(), - maxIterations: 10, - nextID: 1, + tasks: make(map[string]*SubagentTask), + model: model, + defaultModel: defaultModel, + bus: bus, + workspace: workspace, + tools: NewToolRegistry(), + maxIterations: 10, + maxDepth: 3, + maxFanout: 4, + activeChildren: make(map[string]int), + nextID: 1, } } // SetRunLoop injects the loop runner function. Must be called before any -// subagent execution. When nil, falls back to the local RunToolLoop. +// subagent execution. func (sm *SubagentManager) SetRunLoop(fn RunLoopFunc) { sm.mu.Lock() defer sm.mu.Unlock() sm.runLoop = fn } +// SetDelegationLimits configures nested delegation guardrails. +func (sm *SubagentManager) SetDelegationLimits(maxDepth, maxFanout int) { + sm.mu.Lock() + defer sm.mu.Unlock() + if maxDepth > 0 { + sm.maxDepth = maxDepth + } + if maxFanout > 0 { + sm.maxFanout = maxFanout + } +} + +// SetAuditHook registers a callback for delegation lineage/events. +func (sm *SubagentManager) SetAuditHook(hook func(context.Context, DelegationAuditEvent)) { + sm.mu.Lock() + defer sm.mu.Unlock() + sm.auditHook = hook +} + +func (sm *SubagentManager) emitAudit(ctx context.Context, evt DelegationAuditEvent) { + sm.mu.RLock() + hook := sm.auditHook + sm.mu.RUnlock() + if hook != nil { + hook(ctx, evt) + } +} + func (sm *SubagentManager) getRunLoop() RunLoopFunc { sm.mu.RLock() defer sm.mu.RUnlock() - if sm.runLoop != nil { - return sm.runLoop - } - return RunToolLoop + return sm.runLoop } // SetTools sets the tool registry for subagent execution. @@ -82,23 +163,66 @@ func (sm *SubagentManager) RegisterTool(tool Tool) { sm.tools.Register(tool) } -func (sm *SubagentManager) Spawn(ctx context.Context, task, label, originChannel, originChatID string, callback AsyncCallback) (string, error) { +func (sm *SubagentManager) Spawn(ctx context.Context, task, label, delegatedScope, keptWork, originChannel, originChatID string, callback AsyncCallback) (string, error) { sm.mu.Lock() - defer sm.mu.Unlock() + parentTaskID := delegationTaskIDFromContext(ctx) + if parentTaskID == "" { + parentTaskID = "root" + } + parentDepth := delegationDepthFromContext(ctx) + childDepth := parentDepth + 1 + + if childDepth > sm.maxDepth { + sm.mu.Unlock() + return "", fmt.Errorf("delegation depth exceeded: %d > %d", childDepth, sm.maxDepth) + } + if sm.activeChildren[parentTaskID] >= sm.maxFanout { + sm.mu.Unlock() + return "", fmt.Errorf("delegation fanout exceeded for %s: %d >= %d", parentTaskID, sm.activeChildren[parentTaskID], sm.maxFanout) + } + if parentDepth > 0 { + if strings.TrimSpace(delegatedScope) == "" || strings.TrimSpace(keptWork) == "" { + sm.mu.Unlock() + return "", fmt.Errorf("nested delegation requires delegated_scope and kept_work") + } + } + if sm.runLoop == nil { + sm.mu.Unlock() + return "", ErrRunLoopNotConfigured + } taskID := fmt.Sprintf("subagent-%d", sm.nextID) sm.nextID++ + sm.activeChildren[parentTaskID]++ subagentTask := &SubagentTask{ - ID: taskID, - Task: task, - Label: label, - OriginChannel: originChannel, - OriginChatID: originChatID, - Status: "running", - Created: time.Now().UnixMilli(), + ID: taskID, + ParentTaskID: parentTaskID, + Depth: childDepth, + Task: task, + Label: label, + DelegatedScope: delegatedScope, + KeptWork: keptWork, + OriginChannel: originChannel, + OriginChatID: originChatID, + Status: "running", + Created: time.Now().UnixMilli(), } sm.tasks[taskID] = subagentTask + sm.mu.Unlock() + + sm.emitAudit(ctx, DelegationAuditEvent{ + TaskID: taskID, + ParentTaskID: parentTaskID, + Mode: "spawn", + Depth: childDepth, + Status: "created", + Label: label, + DelegatedScope: delegatedScope, + KeptWork: keptWork, + OriginChannel: originChannel, + OriginChatID: originChatID, + }) // Start task in background with context cancellation support go sm.runTask(ctx, subagentTask, callback) @@ -110,12 +234,10 @@ func (sm *SubagentManager) Spawn(ctx context.Context, task, label, originChannel } func (sm *SubagentManager) runTask(ctx context.Context, task *SubagentTask, callback AsyncCallback) { - task.Status = "running" - task.Created = time.Now().UnixMilli() - - systemPrompt := `You are a subagent. Complete the given task independently and report the result. -You have access to tools - use them as needed to complete your task. -After completing the task, provide a clear summary of what was done.` + systemPrompt := `You are a subagent operating under the same runtime discipline as the main agent. +Use tools for actions. Do not claim actions without tool execution. +When discovering tools, call discovered tools directly; use tool_call only as fallback. +Complete the task independently and provide a clear summary of what was done.` // Check if context is already cancelled before starting select { @@ -135,18 +257,51 @@ After completing the task, provide a clear summary of what was done.` sm.mu.RUnlock() runLoop := sm.getRunLoop() - loopResult, err := runLoop(ctx, ToolLoopConfig{ - Model: sm.model, - ModelID: sm.defaultModel, - Tools: tools, - Bus: sm.bus, - MaxIterations: maxIter, - }, systemPrompt, task.Task, task.OriginChannel, task.OriginChatID) + var loopResult *ToolLoopResult + var err error + if runLoop == nil { + err = ErrRunLoopNotConfigured + } else { + taskCtx := withDelegationContext(ctx, task.ID, task.Depth) + loopResult, err = runLoop(taskCtx, ToolLoopConfig{ + Model: sm.model, + ModelID: sm.defaultModel, + Tools: tools, + Bus: sm.bus, + MaxIterations: maxIter, + }, systemPrompt, task.Task, task.OriginChannel, task.OriginChatID) + } sm.mu.Lock() var result *ToolResult + iterations := 0 + resultChars := 0 + errText := "" + finalStatus := task.Status defer func() { + if n := sm.activeChildren[task.ParentTaskID]; n <= 1 { + delete(sm.activeChildren, task.ParentTaskID) + } else { + sm.activeChildren[task.ParentTaskID] = n - 1 + } + finalStatus = task.Status + resultChars = len(task.Result) sm.mu.Unlock() + sm.emitAudit(ctx, DelegationAuditEvent{ + TaskID: task.ID, + ParentTaskID: task.ParentTaskID, + Mode: "spawn", + Depth: task.Depth, + Status: finalStatus, + Label: task.Label, + DelegatedScope: task.DelegatedScope, + KeptWork: task.KeptWork, + Iterations: iterations, + ResultChars: resultChars, + Error: errText, + OriginChannel: task.OriginChannel, + OriginChatID: task.OriginChatID, + }) if callback != nil && result != nil { callback(ctx, result) } @@ -155,6 +310,7 @@ After completing the task, provide a clear summary of what was done.` if err != nil { task.Status = "failed" task.Result = fmt.Sprintf("Error: %v", err) + errText = err.Error() if ctx.Err() != nil { task.Status = "cancelled" task.Result = "Task cancelled during execution" @@ -170,6 +326,7 @@ After completing the task, provide a clear summary of what was done.` } else { task.Status = "completed" task.Result = loopResult.Content + iterations = loopResult.Iterations result = &ToolResult{ ForLLM: fmt.Sprintf("Subagent '%s' completed (iterations: %d): %s", task.Label, loopResult.Iterations, loopResult.Content), ForUser: loopResult.Content, @@ -195,7 +352,11 @@ func (sm *SubagentManager) GetTask(taskID string) (*SubagentTask, bool) { sm.mu.RLock() defer sm.mu.RUnlock() task, ok := sm.tasks[taskID] - return task, ok + if !ok || task == nil { + return nil, false + } + copied := *task + return &copied, true } func (sm *SubagentManager) ListTasks() []*SubagentTask { @@ -204,7 +365,11 @@ func (sm *SubagentManager) ListTasks() []*SubagentTask { tasks := make([]*SubagentTask, 0, len(sm.tasks)) for _, task := range sm.tasks { - tasks = append(tasks, task) + if task == nil { + continue + } + copied := *task + tasks = append(tasks, &copied) } return tasks } @@ -244,6 +409,14 @@ func (t *SubagentTool) Parameters() map[string]interface{} { "type": "string", "description": "Optional short label for the task (for display)", }, + "delegated_scope": map[string]interface{}{ + "type": "string", + "description": "What part of the parent task is being delegated. Required for nested delegation.", + }, + "kept_work": map[string]interface{}{ + "type": "string", + "description": "What work remains with the delegator. Required for nested delegation.", + }, }, "required": []string{"task"}, } @@ -261,21 +434,73 @@ func (t *SubagentTool) Execute(ctx context.Context, args map[string]interface{}) } label, _ := args["label"].(string) + delegatedScope, _ := args["delegated_scope"].(string) + keptWork, _ := args["kept_work"].(string) if t.manager == nil { return ErrorResult("Subagent manager not configured").WithError(fmt.Errorf("manager is nil")) } - systemPrompt := "You are a subagent. Complete the given task independently and provide a clear, concise result." - sm := t.manager + parentTaskID := delegationTaskIDFromContext(ctx) + if parentTaskID == "" { + parentTaskID = "root" + } + parentDepth := delegationDepthFromContext(ctx) + childDepth := parentDepth + 1 sm.mu.RLock() tools := sm.tools maxIter := sm.maxIterations + maxDepth := sm.maxDepth sm.mu.RUnlock() + if childDepth > maxDepth { + return ErrorResult(fmt.Sprintf("delegation depth exceeded: %d > %d", childDepth, maxDepth)) + } + if parentDepth > 0 { + if strings.TrimSpace(delegatedScope) == "" || strings.TrimSpace(keptWork) == "" { + return ErrorResult("nested delegation requires delegated_scope and kept_work") + } + } + + systemPrompt := "You are a subagent operating with main-loop control flow. Execute actions via tools, call discovered tools directly, and provide a clear concise result." + + sm.mu.Lock() + if sm.activeChildren[parentTaskID] >= sm.maxFanout { + sm.mu.Unlock() + return ErrorResult(fmt.Sprintf("delegation fanout exceeded for %s: %d >= %d", parentTaskID, sm.activeChildren[parentTaskID], sm.maxFanout)) + } + sm.activeChildren[parentTaskID]++ + sm.mu.Unlock() + defer func() { + sm.mu.Lock() + if n := sm.activeChildren[parentTaskID]; n <= 1 { + delete(sm.activeChildren, parentTaskID) + } else { + sm.activeChildren[parentTaskID] = n - 1 + } + sm.mu.Unlock() + }() + + taskID := fmt.Sprintf("subagent-sync-%d", time.Now().UnixNano()) + taskCtx := withDelegationContext(ctx, taskID, childDepth) + sm.emitAudit(ctx, DelegationAuditEvent{ + TaskID: taskID, + ParentTaskID: parentTaskID, + Mode: "sync", + Depth: childDepth, + Status: "created", + Label: label, + DelegatedScope: delegatedScope, + KeptWork: keptWork, + OriginChannel: t.originChannel, + OriginChatID: t.originChatID, + }) runLoop := sm.getRunLoop() - loopResult, err := runLoop(ctx, ToolLoopConfig{ + if runLoop == nil { + return ErrorResult("Subagent runtime is not configured").WithError(ErrRunLoopNotConfigured) + } + loopResult, err := runLoop(taskCtx, ToolLoopConfig{ Model: sm.model, ModelID: sm.defaultModel, Tools: tools, @@ -284,6 +509,19 @@ func (t *SubagentTool) Execute(ctx context.Context, args map[string]interface{}) }, systemPrompt, task, t.originChannel, t.originChatID) if err != nil { + sm.emitAudit(ctx, DelegationAuditEvent{ + TaskID: taskID, + ParentTaskID: parentTaskID, + Mode: "sync", + Depth: childDepth, + Status: "failed", + Label: label, + DelegatedScope: delegatedScope, + KeptWork: keptWork, + Error: err.Error(), + OriginChannel: t.originChannel, + OriginChatID: t.originChatID, + }) return ErrorResult(fmt.Sprintf("Subagent execution failed: %v", err)).WithError(err) } @@ -301,6 +539,20 @@ func (t *SubagentTool) Execute(ctx context.Context, args map[string]interface{}) } llmContent := fmt.Sprintf("Subagent task completed:\nLabel: %s\nIterations: %d\nResult: %s", labelStr, loopResult.Iterations, loopResult.Content) + sm.emitAudit(ctx, DelegationAuditEvent{ + TaskID: taskID, + ParentTaskID: parentTaskID, + Mode: "sync", + Depth: childDepth, + Status: "completed", + Label: label, + DelegatedScope: delegatedScope, + KeptWork: keptWork, + Iterations: loopResult.Iterations, + ResultChars: len(loopResult.Content), + OriginChannel: t.originChannel, + OriginChatID: t.originChatID, + }) return &ToolResult{ ForLLM: llmContent, diff --git a/pkg/tools/subagent_manager_test.go b/pkg/tools/subagent_manager_test.go new file mode 100644 index 000000000..65a794c61 --- /dev/null +++ b/pkg/tools/subagent_manager_test.go @@ -0,0 +1,227 @@ +package tools + +import ( + "context" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/ZanzyTHEbar/dragonscale/pkg/bus" +) + +func waitForCondition(t *testing.T, timeout time.Duration, cond func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("condition not met within %s", timeout) +} + +func TestSubagentManager_SpawnRequiresRunLoop(t *testing.T) { + provider := &MockLanguageModel{} + manager := NewSubagentManager(provider, "test-model", "/tmp/test", bus.NewMessageBus()) + + _, err := manager.Spawn(context.Background(), "task-without-loop", "label", "", "", "cli", "chat", nil) + if err == nil { + t.Fatal("expected spawn to fail when run loop is not configured") + } + if !strings.Contains(err.Error(), ErrRunLoopNotConfigured.Error()) { + t.Fatalf("expected run loop contract error, got: %v", err) + } +} + +func TestSubagentManager_SpawnDelegationGuardrails(t *testing.T) { + provider := &MockLanguageModel{} + manager := NewSubagentManager(provider, "test-model", "/tmp/test", bus.NewMessageBus()) + manager.SetDelegationLimits(2, 1) + + block := make(chan struct{}) + manager.SetRunLoop(func(_ context.Context, _ ToolLoopConfig, _, _, _, _ string) (*ToolLoopResult, error) { + <-block + return &ToolLoopResult{Content: "done", Iterations: 1}, nil + }) + + _, err := manager.Spawn(context.Background(), "task-1", "one", "", "", "cli", "chat", nil) + if err != nil { + t.Fatalf("first spawn should succeed: %v", err) + } + + _, err = manager.Spawn(context.Background(), "task-2", "two", "", "", "cli", "chat", nil) + if err == nil || !strings.Contains(err.Error(), "delegation fanout exceeded") { + t.Fatalf("expected fanout error, got: %v", err) + } + + nestedCtx := withDelegationContext(context.Background(), "parent", 1) + _, err = manager.Spawn(nestedCtx, "task-3", "three", "", "", "cli", "chat", nil) + if err == nil || !strings.Contains(err.Error(), "nested delegation requires delegated_scope and kept_work") { + t.Fatalf("expected nested delegation metadata error, got: %v", err) + } + + deepCtx := withDelegationContext(context.Background(), "parent", 2) + _, err = manager.Spawn(deepCtx, "task-4", "four", "lookup", "synthesize", "cli", "chat", nil) + if err == nil || !strings.Contains(err.Error(), "delegation depth exceeded") { + t.Fatalf("expected depth error, got: %v", err) + } + + close(block) + waitForCondition(t, 2*time.Second, func() bool { + for _, task := range manager.ListTasks() { + if task.Status == "running" { + return false + } + } + return true + }) +} + +func TestSubagentManager_ConcurrentSpawnRespectsFanout(t *testing.T) { + provider := &MockLanguageModel{} + manager := NewSubagentManager(provider, "test-model", "/tmp/test", bus.NewMessageBus()) + manager.SetDelegationLimits(3, 2) + + block := make(chan struct{}) + manager.SetRunLoop(func(_ context.Context, _ ToolLoopConfig, _, _, _, _ string) (*ToolLoopResult, error) { + <-block + return &ToolLoopResult{Content: "done", Iterations: 1}, nil + }) + + const attempts = 10 + start := make(chan struct{}) + var wg sync.WaitGroup + + var mu sync.Mutex + successes := 0 + fanoutErrors := 0 + otherErrors := 0 + + for i := 0; i < attempts; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + _, err := manager.Spawn(context.Background(), fmt.Sprintf("task-%d", i), fmt.Sprintf("label-%d", i), "", "", "cli", "chat", nil) + mu.Lock() + defer mu.Unlock() + if err == nil { + successes++ + return + } + if strings.Contains(err.Error(), "delegation fanout exceeded") { + fanoutErrors++ + return + } + otherErrors++ + }(i) + } + + close(start) + wg.Wait() + + mu.Lock() + assertSuccesses := successes + assertFanoutErrors := fanoutErrors + assertOtherErrors := otherErrors + mu.Unlock() + + if assertSuccesses != 2 { + t.Fatalf("expected exactly 2 successful spawns, got %d", assertSuccesses) + } + if assertFanoutErrors != attempts-2 { + t.Fatalf("expected %d fanout errors, got %d", attempts-2, assertFanoutErrors) + } + if assertOtherErrors != 0 { + t.Fatalf("expected 0 non-fanout errors, got %d", assertOtherErrors) + } + + close(block) + waitForCondition(t, 2*time.Second, func() bool { + for _, task := range manager.ListTasks() { + if task.Status == "running" { + return false + } + } + return true + }) +} + +func TestSubagentManager_SpawnAuditLineageAndRuntimeContext(t *testing.T) { + provider := &MockLanguageModel{} + manager := NewSubagentManager(provider, "test-model", "/tmp/test", bus.NewMessageBus()) + + var gotTaskID string + var gotDepth int + var gotChannel string + var gotChatID string + manager.SetRunLoop(func(ctx context.Context, _ ToolLoopConfig, _, _, channel, chatID string) (*ToolLoopResult, error) { + gotTaskID = delegationTaskIDFromContext(ctx) + gotDepth = delegationDepthFromContext(ctx) + gotChannel = channel + gotChatID = chatID + return &ToolLoopResult{Content: "delegated work complete", Iterations: 3}, nil + }) + + eventsCh := make(chan DelegationAuditEvent, 4) + manager.SetAuditHook(func(_ context.Context, evt DelegationAuditEvent) { + eventsCh <- evt + }) + + parentCtx := withDelegationContext(context.Background(), "parent-9", 1) + _, err := manager.Spawn(parentCtx, "task-a", "label-a", "collect facts", "final synthesis", "telegram", "chat-7", nil) + if err != nil { + t.Fatalf("spawn failed: %v", err) + } + + var created *DelegationAuditEvent + var completed *DelegationAuditEvent + timeout := time.After(2 * time.Second) + for created == nil || completed == nil { + select { + case evt := <-eventsCh: + e := evt + switch evt.Status { + case "created": + created = &e + case "completed": + completed = &e + } + case <-timeout: + t.Fatal("timed out waiting for delegation audit events") + } + } + + if created.ParentTaskID != "parent-9" { + t.Fatalf("expected parent task parent-9, got %s", created.ParentTaskID) + } + if created.Depth != 2 { + t.Fatalf("expected child depth 2, got %d", created.Depth) + } + if created.DelegatedScope != "collect facts" || created.KeptWork != "final synthesis" { + t.Fatalf("unexpected delegation metadata: %+v", *created) + } + + if completed.TaskID != created.TaskID { + t.Fatalf("expected completion for created task %s, got %s", created.TaskID, completed.TaskID) + } + if completed.Iterations != 3 { + t.Fatalf("expected completion iterations=3, got %d", completed.Iterations) + } + if completed.ResultChars == 0 { + t.Fatal("expected completion to include non-zero result chars") + } + + if gotTaskID != created.TaskID { + t.Fatalf("run loop context task id mismatch: got %s want %s", gotTaskID, created.TaskID) + } + if gotDepth != 2 { + t.Fatalf("run loop context depth mismatch: got %d want 2", gotDepth) + } + if gotChannel != "telegram" || gotChatID != "chat-7" { + t.Fatalf("run loop origin context mismatch: channel=%s chat=%s", gotChannel, gotChatID) + } +} diff --git a/pkg/tools/subagent_tool_test.go b/pkg/tools/subagent_tool_test.go index fc3b8396d..e5f8205b5 100644 --- a/pkg/tools/subagent_tool_test.go +++ b/pkg/tools/subagent_tool_test.go @@ -7,7 +7,7 @@ import ( "testing" fantasy "charm.land/fantasy" - "github.com/sipeed/picoclaw/pkg/bus" + "github.com/ZanzyTHEbar/dragonscale/pkg/bus" ) // MockLanguageModel is a test implementation of fantasy.LanguageModel @@ -123,6 +123,24 @@ func TestSubagentTool_Parameters(t *testing.T) { t.Errorf("Label type should be 'string', got: %v", label["type"]) } + // Verify delegated_scope parameter + delegatedScope, ok := props["delegated_scope"].(map[string]interface{}) + if !ok { + t.Fatal("delegated_scope parameter should exist") + } + if delegatedScope["type"] != "string" { + t.Errorf("delegated_scope type should be 'string', got: %v", delegatedScope["type"]) + } + + // Verify kept_work parameter + keptWork, ok := props["kept_work"].(map[string]interface{}) + if !ok { + t.Fatal("kept_work parameter should exist") + } + if keptWork["type"] != "string" { + t.Errorf("kept_work type should be 'string', got: %v", keptWork["type"]) + } + // Check required fields required, ok := params["required"].([]string) if !ok { @@ -316,6 +334,52 @@ func TestSubagentTool_Execute_ContextPassing(t *testing.T) { // but execution success indicates context was handled properly } +func TestSubagentTool_Execute_NestedDelegationRequiresScopeAndKeptWork(t *testing.T) { + provider := &MockLanguageModel{} + msgBus := bus.NewMessageBus() + manager := NewSubagentManager(provider, "test-model", "/tmp/test", msgBus) + manager.SetRunLoop(func(_ context.Context, _ ToolLoopConfig, _, userPrompt, _, _ string) (*ToolLoopResult, error) { + return &ToolLoopResult{Content: "Task completed: " + userPrompt, Iterations: 1}, nil + }) + tool := NewSubagentTool(manager) + + // Simulate nested delegation (depth > 0) without delegated scope metadata. + ctx := withDelegationContext(context.Background(), "parent-task", 1) + result := tool.Execute(ctx, map[string]interface{}{ + "task": "nested task", + "label": "nested", + }) + + if !result.IsError { + t.Fatal("Expected nested delegation without delegated_scope/kept_work to fail") + } + if !strings.Contains(result.ForLLM, "nested delegation requires delegated_scope and kept_work") { + t.Fatalf("unexpected error: %s", result.ForLLM) + } +} + +func TestSubagentTool_Execute_NestedDelegationWithMetadataSucceeds(t *testing.T) { + provider := &MockLanguageModel{} + msgBus := bus.NewMessageBus() + manager := NewSubagentManager(provider, "test-model", "/tmp/test", msgBus) + manager.SetRunLoop(func(_ context.Context, _ ToolLoopConfig, _, userPrompt, _, _ string) (*ToolLoopResult, error) { + return &ToolLoopResult{Content: "Task completed: " + userPrompt, Iterations: 2}, nil + }) + tool := NewSubagentTool(manager) + + ctx := withDelegationContext(context.Background(), "parent-task", 1) + result := tool.Execute(ctx, map[string]interface{}{ + "task": "nested task", + "label": "nested", + "delegated_scope": "collect additional facts", + "kept_work": "final synthesis", + }) + + if result.IsError { + t.Fatalf("Expected nested delegation with metadata to succeed, got: %s", result.ForLLM) + } +} + // TestSubagentTool_ForUserTruncation verifies long content is truncated for user func TestSubagentTool_ForUserTruncation(t *testing.T) { provider := &MockLanguageModel{} diff --git a/pkg/tools/toolloop_test.go b/pkg/tools/toolloop_test.go new file mode 100644 index 000000000..dc21efef4 --- /dev/null +++ b/pkg/tools/toolloop_test.go @@ -0,0 +1,20 @@ +package tools + +import ( + "context" + "errors" + "testing" +) + +func TestRunToolLoop_ReturnsContractError(t *testing.T) { + result, err := RunToolLoop(context.Background(), ToolLoopConfig{}, "", "", "", "") + if result != nil { + t.Fatalf("expected nil result when run loop is not configured, got %#v", result) + } + if err == nil { + t.Fatal("expected contract error from RunToolLoop fallback") + } + if !errors.Is(err, ErrRunLoopNotConfigured) { + t.Fatalf("expected ErrRunLoopNotConfigured, got: %v", err) + } +}