refactor(agent): fail-fast memory init, XDG skill paths, extract toolloop
loop.go: - NewAgentLoop now takes a context.Context and returns (*AgentLoop, error); memory init is mandatory — errors are returned instead of logged+skipped - Wire IdentitySync into the loop (identitySync field) - Use cfg.SandboxPath() / cfg.DBPath() / cfg.RestrictToSandbox() - SetContextWindow() passed to ContextBuilder from cfg.Agents.Defaults.MaxTokens context.go: - NewContextBuilder resolves skills dirs via config.SkillsDir() and config.ConfigDir() (XDG-aware) instead of hardcoded ~/.picoclaw - BuildSystemPrompt: priority-ordered sections with token-budget trimming (charsPerToken heuristic); skills section updated with tool_search guidance pkg/agent/toolloop.go (new): - Extract ToolLoopMode, DAGRunResult, DAGRunFunc, RouteFunc types from pkg/tools/toolloop.go to break the tools→dag→securebus→tools import cycle
This commit is contained in:
ZanzyTHEbar
parent
e8d6e01cea
commit
7344f90a7f
5 changed files with 459 additions and 476 deletions
|
|
@ -9,6 +9,7 @@ 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"
|
||||
|
|
@ -25,26 +26,30 @@ type ContextBuilder struct {
|
|||
observationBlock string // Pre-rendered observation block for prompt injection
|
||||
knowledgeBlock string // Pre-rendered knowledge block from Focus completions
|
||||
dagBlock string // Pre-rendered DAG compressed history
|
||||
}
|
||||
|
||||
func getGlobalConfigDir() string {
|
||||
home, err := os.UserHomeDir()
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return filepath.Join(home, ".picoclaw")
|
||||
contextWindow int // Max tokens for context window (0 = no limit)
|
||||
}
|
||||
|
||||
func NewContextBuilder(workspace string) *ContextBuilder {
|
||||
// builtin skills: skills directory in current project
|
||||
// Use the skills/ directory under the current working directory
|
||||
// Primary skills dir: XDG data dir (installed skills).
|
||||
// Falls back to workspace/skills for legacy setups.
|
||||
primarySkillsDir := filepath.Join(workspace, "skills")
|
||||
if dir, err := config.SkillsDir(); err == nil {
|
||||
primarySkillsDir = dir
|
||||
}
|
||||
|
||||
// Global skills: ~/.config/picoclaw/skills (user-level overrides).
|
||||
globalSkillsDir := ""
|
||||
if dir, err := config.ConfigDir(); err == nil {
|
||||
globalSkillsDir = filepath.Join(dir, "skills")
|
||||
}
|
||||
|
||||
// Builtin skills: skills/ directory relative to the binary's working dir.
|
||||
wd, _ := os.Getwd()
|
||||
builtinSkillsDir := filepath.Join(wd, "skills")
|
||||
globalSkillsDir := filepath.Join(getGlobalConfigDir(), "skills")
|
||||
|
||||
return &ContextBuilder{
|
||||
workspace: workspace,
|
||||
skillsLoader: skills.NewSkillsLoader(workspace, globalSkillsDir, builtinSkillsDir),
|
||||
skillsLoader: skills.NewSkillsLoader(primarySkillsDir, globalSkillsDir, builtinSkillsDir),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -81,6 +86,11 @@ func (cb *ContextBuilder) SetDAGBlock(block string) {
|
|||
cb.dagBlock = block
|
||||
}
|
||||
|
||||
// SetContextWindow configures the token budget for the system prompt.
|
||||
func (cb *ContextBuilder) SetContextWindow(tokens int) {
|
||||
cb.contextWindow = tokens
|
||||
}
|
||||
|
||||
func (cb *ContextBuilder) getIdentity() string {
|
||||
now := time.Now().Format("2006-01-02 15:04 (Monday)")
|
||||
workspacePath, _ := filepath.Abs(filepath.Join(cb.workspace))
|
||||
|
|
@ -139,86 +149,138 @@ 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 {
|
||||
parts := []string{}
|
||||
|
||||
// Core identity section
|
||||
parts = append(parts, cb.getIdentity())
|
||||
|
||||
// Bootstrap files
|
||||
bootstrapContent := cb.LoadBootstrapFiles()
|
||||
if bootstrapContent != "" {
|
||||
parts = append(parts, bootstrapContent)
|
||||
type section struct {
|
||||
name string
|
||||
content string
|
||||
priority int // lower = higher priority (kept first when trimming)
|
||||
}
|
||||
|
||||
// Skills - show summary index and inline full definitions for direct use
|
||||
skillsSummary := cb.skillsLoader.BuildSkillsSummary()
|
||||
if skillsSummary != "" {
|
||||
parts = append(parts, fmt.Sprintf(`# Skills
|
||||
// Collect sections in priority order
|
||||
sections := []section{}
|
||||
|
||||
The following skills extend your capabilities. Full definitions are included below.
|
||||
// P0: Core identity (always included)
|
||||
sections = append(sections, section{"identity", cb.getIdentity(), 0})
|
||||
|
||||
%s`, skillsSummary))
|
||||
}
|
||||
if skillsDefs := cb.loadSkills(); skillsDefs != "" {
|
||||
parts = append(parts, skillsDefs)
|
||||
// P1: Bootstrap files (user identity)
|
||||
if bc := cb.LoadBootstrapFiles(); bc != "" {
|
||||
sections = append(sections, section{"bootstrap", bc, 1})
|
||||
}
|
||||
|
||||
// Observation block (stable prefix for prompt cache alignment)
|
||||
if cb.observationBlock != "" {
|
||||
parts = append(parts, "# Observations\n\n"+cb.observationBlock)
|
||||
// P2: Skills index (lightweight Level 1 metadata)
|
||||
if summary := cb.skillsLoader.BuildSkillsSummary(); summary != "" {
|
||||
sections = append(sections, section{"skills", fmt.Sprintf(`# Skills
|
||||
|
||||
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
|
||||
|
||||
Do NOT assume skill content — always load before applying.
|
||||
|
||||
%s`, summary), 2})
|
||||
}
|
||||
|
||||
if cb.knowledgeBlock != "" {
|
||||
parts = append(parts, cb.knowledgeBlock)
|
||||
}
|
||||
|
||||
if cb.dagBlock != "" {
|
||||
parts = append(parts, "# Conversation History (Compressed)\n\n"+cb.dagBlock)
|
||||
}
|
||||
|
||||
// 3-tier MemGPT working context injection
|
||||
// P3: Working context (hot tier — highly dynamic, high value)
|
||||
if cb.memoryStore != nil {
|
||||
wcSection := cb.buildWorkingContextSection()
|
||||
if wcSection != "" {
|
||||
parts = append(parts, wcSection)
|
||||
if wc := cb.buildWorkingContextSection(); wc != "" {
|
||||
sections = append(sections, section{"working_context", wc, 3})
|
||||
}
|
||||
}
|
||||
|
||||
// Join with "---" separator
|
||||
return strings.Join(parts, "\n\n---\n\n")
|
||||
// P4: Observation block
|
||||
if cb.observationBlock != "" {
|
||||
sections = append(sections, section{"observations", "# Observations\n\n" + cb.observationBlock, 4})
|
||||
}
|
||||
|
||||
// P5: Knowledge block
|
||||
if cb.knowledgeBlock != "" {
|
||||
sections = append(sections, section{"knowledge", cb.knowledgeBlock, 5})
|
||||
}
|
||||
|
||||
// P6: DAG compressed history (lowest priority — can be reconstructed)
|
||||
if cb.dagBlock != "" {
|
||||
sections = append(sections, section{"dag", "# Conversation History (Compressed)\n\n" + cb.dagBlock, 6})
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
|
||||
if budgetChars > 0 && totalChars > budgetChars {
|
||||
logger.WarnCF("context", "System prompt exceeds token budget, trimming low-priority sections",
|
||||
map[string]interface{}{
|
||||
"total_chars": totalChars,
|
||||
"budget_chars": budgetChars,
|
||||
"sections": len(sections),
|
||||
})
|
||||
// Trim from lowest priority (highest number) first
|
||||
for i := len(sections) - 1; i >= 0 && totalChars > budgetChars; i-- {
|
||||
if sections[i].priority >= 5 { // only trim P5+ (knowledge, dag)
|
||||
totalChars -= len(sections[i].content)
|
||||
sections[i].content = ""
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
parts := make([]string, 0, len(sections))
|
||||
for _, s := range sections {
|
||||
if s.content != "" {
|
||||
parts = append(parts, s.content)
|
||||
}
|
||||
}
|
||||
|
||||
prompt := strings.Join(parts, "\n\n---\n\n")
|
||||
|
||||
// Log token estimate for observability
|
||||
tokenEst := len(prompt) / charsPerToken
|
||||
logger.DebugCF("context", "System prompt token estimate",
|
||||
map[string]interface{}{
|
||||
"chars": len(prompt),
|
||||
"tokens_est": tokenEst,
|
||||
"sections": len(parts),
|
||||
})
|
||||
|
||||
return prompt
|
||||
}
|
||||
|
||||
// tokenBudgetChars returns the maximum character count for the system prompt,
|
||||
// derived from the context window size. Returns 0 if no limit is configured.
|
||||
func (cb *ContextBuilder) tokenBudgetChars() int {
|
||||
if cb.contextWindow <= 0 {
|
||||
return 0
|
||||
}
|
||||
// Reserve ~40% of context window for system prompt
|
||||
return int(float64(cb.contextWindow) * 0.4 * charsPerToken)
|
||||
}
|
||||
|
||||
func (cb *ContextBuilder) LoadBootstrapFiles() string {
|
||||
if cb.delegate != nil {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
|
||||
docs, err := cb.delegate.ListDocumentsByCategory(ctx, "picoclaw", "bootstrap")
|
||||
if err == nil && len(docs) > 0 {
|
||||
var result string
|
||||
for _, doc := range docs {
|
||||
result += fmt.Sprintf("## %s\n\n%s\n\n", doc.Name, doc.Content)
|
||||
}
|
||||
return result
|
||||
}
|
||||
if cb.delegate == nil {
|
||||
return ""
|
||||
}
|
||||
|
||||
bootstrapFiles := []string{
|
||||
"AGENTS.md",
|
||||
"SOUL.md",
|
||||
"USER.md",
|
||||
"IDENTITY.md",
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
|
||||
docs, err := cb.delegate.ListDocumentsByCategory(ctx, "picoclaw", "bootstrap")
|
||||
if err != nil || len(docs) == 0 {
|
||||
return ""
|
||||
}
|
||||
|
||||
var result string
|
||||
for _, filename := range bootstrapFiles {
|
||||
filePath := filepath.Join(cb.workspace, filename)
|
||||
if data, err := os.ReadFile(filePath); err == nil {
|
||||
result += fmt.Sprintf("## %s\n\n%s\n\n", filename, string(data))
|
||||
}
|
||||
for _, doc := range docs {
|
||||
result += fmt.Sprintf("## %s\n\n%s\n\n", doc.Name, doc.Content)
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
|
|
@ -329,26 +391,7 @@ func (cb *ContextBuilder) AddAssistantMessage(msgs []messages.Message, content s
|
|||
return msgs
|
||||
}
|
||||
|
||||
func (cb *ContextBuilder) loadSkills() string {
|
||||
allSkills := cb.skillsLoader.ListSkills()
|
||||
if len(allSkills) == 0 {
|
||||
return ""
|
||||
}
|
||||
|
||||
var skillNames []string
|
||||
for _, s := range allSkills {
|
||||
skillNames = append(skillNames, s.Name)
|
||||
}
|
||||
|
||||
content := cb.skillsLoader.LoadSkillsForContext(skillNames)
|
||||
if content == "" {
|
||||
return ""
|
||||
}
|
||||
|
||||
return "# Skill Definitions\n\n" + content
|
||||
}
|
||||
|
||||
// GetSkillsInfo returns information about loaded skills.
|
||||
// GetSkillsInfo returns information about available skills (metadata only).
|
||||
func (cb *ContextBuilder) GetSkillsInfo() map[string]interface{} {
|
||||
allSkills := cb.skillsLoader.ListSkills()
|
||||
skillNames := make([]string, 0, len(allSkills))
|
||||
|
|
|
|||
|
|
@ -181,7 +181,7 @@ func TestIntegration_FullAgentLoop_SimpleResponse(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("Hello from Fantasy agent")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
|
@ -244,7 +244,7 @@ func TestIntegration_FullAgentLoop_WithToolCalls(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := &toolCallingModel{}
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
// Register the echo tool
|
||||
al.RegisterTool(&echoTool{})
|
||||
|
|
@ -298,7 +298,7 @@ func TestIntegration_ProcessDirect(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("Direct CLI response")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
|
@ -393,7 +393,7 @@ func TestIntegration_Streaming_TextDeltas(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newStreamingModel("Hello from streaming agent response")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
|
@ -469,7 +469,7 @@ func TestIntegration_Streaming_WithToolCalls(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := &toolCallingModel{}
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
al.RegisterTool(&echoTool{})
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
|
|
@ -531,7 +531,7 @@ func TestIntegration_MultipleMessages(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("Response")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
sessionKey := "multi-msg-session"
|
||||
ctx := context.Background()
|
||||
|
|
|
|||
|
|
@ -35,6 +35,7 @@ import (
|
|||
"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"
|
||||
)
|
||||
|
|
@ -50,11 +51,12 @@ type AgentLoop struct {
|
|||
state *state.Manager
|
||||
contextBuilder *ContextBuilder
|
||||
tools *tools.ToolRegistry
|
||||
memoryStore *memstore.MemoryStore // 3-tier MemGPT memory (nil if init failed)
|
||||
memDelegate memory.MemoryDelegate // DB delegate (nil if memory disabled)
|
||||
obsManager *observation.Manager // Observational memory (nil if memory disabled)
|
||||
secureBus *securebus.Bus // ITR SecureBus (nil = disabled, direct execution)
|
||||
activeSessionKey atomic.Value // Current session key for tool access
|
||||
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)
|
||||
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
|
||||
summarizing sync.Map // Tracks which sessions are currently being summarized
|
||||
summarizeFailures sync.Map // Tracks consecutive summarization failures per session (string -> int)
|
||||
|
|
@ -125,160 +127,151 @@ func createToolRegistry(workspace string, restrict bool, cfg *config.Config, msg
|
|||
return registry
|
||||
}
|
||||
|
||||
func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, model fantasy.LanguageModel) *AgentLoop {
|
||||
workspace := cfg.WorkspacePath()
|
||||
os.MkdirAll(workspace, 0755)
|
||||
func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBus, model fantasy.LanguageModel) (*AgentLoop, error) {
|
||||
sandbox := cfg.SandboxPath()
|
||||
os.MkdirAll(sandbox, 0755)
|
||||
|
||||
restrict := cfg.Agents.Defaults.RestrictToWorkspace
|
||||
workspace := sandbox
|
||||
restrict := cfg.RestrictToSandbox()
|
||||
|
||||
// Create tool registry for main agent
|
||||
toolsRegistry := createToolRegistry(workspace, restrict, cfg, msgBus)
|
||||
|
||||
// Create subagent manager with its own tool registry
|
||||
subagentManager := tools.NewSubagentManager(model, cfg.Agents.Defaults.Model, workspace, msgBus)
|
||||
subagentTools := createToolRegistry(workspace, restrict, cfg, msgBus)
|
||||
// Subagent doesn't need spawn/subagent tools to avoid recursion
|
||||
subagentManager.SetTools(subagentTools)
|
||||
|
||||
// Register spawn tool (for main agent)
|
||||
spawnTool := tools.NewSpawnTool(subagentManager)
|
||||
toolsRegistry.Register(spawnTool)
|
||||
|
||||
// Register subagent tool (synchronous execution)
|
||||
subagentTool := tools.NewSubagentTool(subagentManager)
|
||||
toolsRegistry.Register(subagentTool)
|
||||
|
||||
// Create context builder and set tools registry
|
||||
contextBuilder := NewContextBuilder(workspace)
|
||||
contextBuilder.SetToolsRegistry(toolsRegistry)
|
||||
contextBuilder.SetContextWindow(cfg.Agents.Defaults.MaxTokens)
|
||||
|
||||
// Progressive skill disclosure tools (skill_search → skill_read → skill_traverse)
|
||||
sl := contextBuilder.SkillsLoader()
|
||||
toolsRegistry.Register(tools.NewSkillSearchTool(sl))
|
||||
toolsRegistry.Register(tools.NewSkillReadTool(sl))
|
||||
toolsRegistry.Register(tools.NewSkillTraverseTool(sl))
|
||||
|
||||
// Initialize 3-tier MemGPT memory system
|
||||
var ms *memstore.MemoryStore
|
||||
var memDelegate memory.MemoryDelegate
|
||||
if cfg.Memory.Enabled {
|
||||
memDBPath := filepath.Join(workspace, "memory", "picoclaw.db")
|
||||
os.MkdirAll(filepath.Dir(memDBPath), 0755)
|
||||
// Initialize 3-tier MemGPT memory system (always enabled, fail-fast on error)
|
||||
memDBPath := cfg.DBPath()
|
||||
os.MkdirAll(filepath.Dir(memDBPath), 0755)
|
||||
|
||||
del, err := delegate.NewFromConfig(cfg.Memory, memDBPath)
|
||||
if err != nil {
|
||||
logger.WarnCF("agent", "Failed to create memory delegate, memory system disabled",
|
||||
map[string]interface{}{"error": err.Error()})
|
||||
} else {
|
||||
if err := del.Init(context.Background()); err != nil {
|
||||
logger.WarnCF("agent", "Failed to init memory schema, memory system disabled",
|
||||
map[string]interface{}{"error": err.Error()})
|
||||
del.Close()
|
||||
} else {
|
||||
memDelegate = del
|
||||
offloadThreshold := cfg.Memory.OffloadThresholdTokens
|
||||
if offloadThreshold <= 0 {
|
||||
offloadThreshold = 4000
|
||||
}
|
||||
chunker := memstore.NewMarkdownChunker(memstore.DefaultMarkdownChunkerConfig())
|
||||
del, err := delegate.NewFromConfig(cfg.Memory, memDBPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("memory delegate init: %w", err)
|
||||
}
|
||||
if err := del.Init(ctx); err != nil {
|
||||
del.Close()
|
||||
return nil, fmt.Errorf("memory schema init: %w", err)
|
||||
}
|
||||
memDelegate := del
|
||||
|
||||
embedder, embErr := memstore.NewEmbedderFromConfig(cfg.Memory.Embedding, cfg.Providers)
|
||||
if embErr != nil {
|
||||
logger.WarnCF("agent", "Failed to create embedding provider, archival search will use FTS5 only",
|
||||
map[string]interface{}{"error": embErr.Error()})
|
||||
}
|
||||
offloadThreshold := cfg.Memory.OffloadThresholdTokens
|
||||
if offloadThreshold <= 0 {
|
||||
offloadThreshold = 4000
|
||||
}
|
||||
chunker := memstore.NewMarkdownChunker(memstore.DefaultMarkdownChunkerConfig())
|
||||
|
||||
ms = memstore.New(del, chunker, embedder, memstore.Config{
|
||||
ContextWindowTokens: cfg.Agents.Defaults.MaxTokens,
|
||||
OffloadThresholdTokens: offloadThreshold,
|
||||
})
|
||||
contextBuilder.SetMemoryStore(ms)
|
||||
embedder, embErr := memstore.NewEmbedderFromConfig(cfg.Memory.Embedding, cfg.Providers)
|
||||
if embErr != nil {
|
||||
logger.WarnCF("agent", "Failed to create embedding provider, archival search will use FTS5 only",
|
||||
map[string]interface{}{"error": embErr.Error()})
|
||||
}
|
||||
|
||||
memTool := NewMemGPTTool(ms, "picoclaw", "default")
|
||||
toolsRegistry.Register(memTool)
|
||||
ms := memstore.New(del, chunker, embedder, memstore.Config{
|
||||
ContextWindowTokens: cfg.Agents.Defaults.MaxTokens,
|
||||
OffloadThresholdTokens: offloadThreshold,
|
||||
})
|
||||
contextBuilder.SetMemoryStore(ms)
|
||||
|
||||
// Agentic retrieval tools (keyword_search → semantic_search → chunk_read)
|
||||
toolsRegistry.Register(tools.NewKeywordSearchTool(ms, "picoclaw"))
|
||||
toolsRegistry.Register(tools.NewSemanticSearchTool(ms, "picoclaw"))
|
||||
toolsRegistry.Register(tools.NewChunkReadTool(ms, "picoclaw"))
|
||||
memTool := NewMemGPTTool(ms, "picoclaw", "default")
|
||||
toolsRegistry.Register(memTool)
|
||||
|
||||
contextBuilder.SetDelegate(del)
|
||||
toolsRegistry.Register(tools.NewKeywordSearchTool(ms, "picoclaw"))
|
||||
toolsRegistry.Register(tools.NewSemanticSearchTool(ms, "picoclaw"))
|
||||
toolsRegistry.Register(tools.NewChunkReadTool(ms, "picoclaw"))
|
||||
|
||||
// One-time migrations
|
||||
mctx := context.Background()
|
||||
if migErr := memory.MigrateState(mctx, workspace, del, "picoclaw"); migErr != nil {
|
||||
logger.WarnCF("agent", "State KV migration failed (non-fatal)",
|
||||
map[string]interface{}{"error": migErr.Error()})
|
||||
}
|
||||
if migErr := memory.MigrateDocuments(mctx, workspace, del, "picoclaw"); migErr != nil {
|
||||
logger.WarnCF("agent", "Document migration failed (non-fatal)",
|
||||
map[string]interface{}{"error": migErr.Error()})
|
||||
}
|
||||
if migErr := memory.MigrateLongTermMemory(mctx, workspace, del, "picoclaw"); migErr != nil {
|
||||
logger.WarnCF("agent", "Long-term memory migration failed (non-fatal)",
|
||||
map[string]interface{}{"error": migErr.Error()})
|
||||
}
|
||||
if migErr := memory.MigrateDailyNotes(mctx, workspace, del, "picoclaw"); migErr != nil {
|
||||
logger.WarnCF("agent", "Daily notes migration failed (non-fatal)",
|
||||
map[string]interface{}{"error": migErr.Error()})
|
||||
}
|
||||
}
|
||||
}
|
||||
contextBuilder.SetDelegate(del)
|
||||
|
||||
if migErr := memory.MigrateState(ctx, workspace, del, "picoclaw"); 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 {
|
||||
logger.WarnCF("agent", "Document migration failed (non-fatal)",
|
||||
map[string]interface{}{"error": migErr.Error()})
|
||||
}
|
||||
if migErr := memory.MigrateLongTermMemory(ctx, workspace, del, "picoclaw"); 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 {
|
||||
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()
|
||||
if idErr != nil {
|
||||
logger.WarnCF("agent", "Could not resolve identity dir, identity sync disabled",
|
||||
map[string]interface{}{"error": idErr.Error()})
|
||||
} else {
|
||||
logger.InfoCF("agent", "Memory system disabled by config", nil)
|
||||
}
|
||||
|
||||
// Create state manager -- use delegate-backed KV when memory system is active
|
||||
var stateOpts []state.Option
|
||||
if memDelegate != nil {
|
||||
stateOpts = append(stateOpts, state.WithDelegate(memDelegate))
|
||||
}
|
||||
stateManager := state.NewManager(workspace, stateOpts...)
|
||||
|
||||
// Create session manager -- use delegate for DB persistence when available
|
||||
sessionsDir := filepath.Join(workspace, "sessions")
|
||||
var sessionOpts []session.SessionOption
|
||||
if memDelegate != nil {
|
||||
sessionOpts = append(sessionOpts, session.WithSessionDelegate(memDelegate, "picoclaw"))
|
||||
}
|
||||
sessionsManager := session.NewSessionManager(sessionsDir, sessionOpts...)
|
||||
|
||||
// Register meta-tools for progressive disclosure (tool_search + tool_call)
|
||||
toolsRegistry.RegisterMetaTools()
|
||||
|
||||
// If memory tool is a gateway, mark it visible in progressive mode
|
||||
if ms != nil {
|
||||
toolsRegistry.MarkGateway("memory")
|
||||
}
|
||||
|
||||
// Apply progressive disclosure config
|
||||
if cfg.Tools.ProgressiveDisclosure {
|
||||
toolsRegistry.SetProgressiveDisclosure(true)
|
||||
logger.InfoCF("agent", "Progressive tool disclosure enabled",
|
||||
map[string]interface{}{"gateway_tools": toolsRegistry.ListVisible()})
|
||||
}
|
||||
|
||||
// Initialize observation manager if memory is enabled
|
||||
var obsManager *observation.Manager
|
||||
if memDelegate != nil {
|
||||
callModelFn := func(ctx context.Context, prompt string) (string, error) {
|
||||
temp := 0.3
|
||||
maxTokens := int64(1024)
|
||||
resp, err := model.Generate(ctx, fantasy.Call{
|
||||
Prompt: fantasy.Prompt{
|
||||
fantasy.NewUserMessage(prompt),
|
||||
},
|
||||
Temperature: &temp,
|
||||
MaxOutputTokens: &maxTokens,
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return resp.Content.Text(), nil
|
||||
idSync = picosync.New(identityDir, "picoclaw", memDelegate)
|
||||
if syncErr := idSync.SyncAll(ctx); syncErr != nil {
|
||||
logger.WarnCF("agent", "Initial identity sync failed (non-fatal)",
|
||||
map[string]interface{}{"error": syncErr.Error()})
|
||||
} else {
|
||||
logger.InfoCF("agent", "Identity files synced to DB", nil)
|
||||
}
|
||||
if watchErr := idSync.Watch(ctx); watchErr != nil {
|
||||
logger.WarnCF("agent", "Identity file watcher failed to start, using mtime fallback",
|
||||
map[string]interface{}{"error": watchErr.Error()})
|
||||
}
|
||||
obsManager = observation.NewManager(memDelegate, "picoclaw", callModelFn, observation.DefaultManagerConfig())
|
||||
}
|
||||
|
||||
// State manager (always delegate-backed)
|
||||
stateManager := state.NewManager(workspace, state.WithDelegate(memDelegate))
|
||||
|
||||
// Session manager (always delegate-backed)
|
||||
sessionsDir := filepath.Join(workspace, "sessions")
|
||||
sessionsManager := session.NewSessionManager(sessionsDir, session.WithSessionDelegate(memDelegate, "picoclaw"))
|
||||
|
||||
// Meta-tools for progressive disclosure (tool_search + tool_call)
|
||||
toolsRegistry.RegisterMetaTools()
|
||||
toolsRegistry.MarkGateway("memory")
|
||||
toolsRegistry.MarkGateway("skill_search")
|
||||
|
||||
// Wire skills loader into tool_search for unified discovery
|
||||
if ts, ok := toolsRegistry.Get("tool_search"); ok {
|
||||
if tst, ok := ts.(*tools.ToolSearchTool); ok {
|
||||
tst.SetSkillsLoader(contextBuilder.SkillsLoader())
|
||||
}
|
||||
}
|
||||
|
||||
// Observation manager
|
||||
callModelFn := func(ctx context.Context, prompt string) (string, error) {
|
||||
temp := 0.3
|
||||
maxTokens := int64(1024)
|
||||
resp, err := model.Generate(ctx, fantasy.Call{
|
||||
Prompt: fantasy.Prompt{
|
||||
fantasy.NewUserMessage(prompt),
|
||||
},
|
||||
Temperature: &temp,
|
||||
MaxOutputTokens: &maxTokens,
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return resp.Content.Text(), nil
|
||||
}
|
||||
obsManager := observation.NewManager(memDelegate, "picoclaw", callModelFn, observation.DefaultManagerConfig())
|
||||
|
||||
al := &AgentLoop{
|
||||
bus: msgBus,
|
||||
languageModel: model,
|
||||
|
|
@ -293,24 +286,22 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, model fantasy.Lang
|
|||
memoryStore: ms,
|
||||
memDelegate: memDelegate,
|
||||
obsManager: obsManager,
|
||||
identitySync: idSync,
|
||||
summarizing: sync.Map{},
|
||||
cfg: cfg,
|
||||
}
|
||||
|
||||
// Register focus tools (start_focus / complete_focus) when memory is available.
|
||||
// The sessionKeyFn closure reads the activeSessionKey set at the start of each agent turn.
|
||||
if memDelegate != nil {
|
||||
sessionKeyFn := func() string {
|
||||
if v := al.activeSessionKey.Load(); v != nil {
|
||||
return v.(string)
|
||||
}
|
||||
return ""
|
||||
// Focus tools (start_focus / complete_focus)
|
||||
sessionKeyFn := func() string {
|
||||
if v := al.activeSessionKey.Load(); v != nil {
|
||||
return v.(string)
|
||||
}
|
||||
toolsRegistry.Register(tools.NewStartFocusTool(memDelegate, sessionsManager, sessionKeyFn))
|
||||
toolsRegistry.Register(tools.NewCompleteFocusTool(memDelegate, sessionsManager, sessionKeyFn))
|
||||
return ""
|
||||
}
|
||||
toolsRegistry.Register(tools.NewStartFocusTool(memDelegate, sessionsManager, sessionKeyFn))
|
||||
toolsRegistry.Register(tools.NewCompleteFocusTool(memDelegate, sessionsManager, sessionKeyFn))
|
||||
|
||||
return al
|
||||
return al, nil
|
||||
}
|
||||
|
||||
func (al *AgentLoop) Run(ctx context.Context) error {
|
||||
|
|
@ -357,6 +348,9 @@ func (al *AgentLoop) Run(ctx context.Context) error {
|
|||
|
||||
func (al *AgentLoop) Stop() {
|
||||
al.running.Store(false)
|
||||
if al.identitySync != nil {
|
||||
al.identitySync.Close()
|
||||
}
|
||||
if al.memoryStore != nil {
|
||||
if err := al.memoryStore.Sync(); err != nil {
|
||||
logger.WarnCF("agent", "Failed to sync memory before shutdown",
|
||||
|
|
@ -405,14 +399,14 @@ func (al *AgentLoop) SetupSecureBus(ss *security.SecretStore, cfg securebus.BusC
|
|||
|
||||
// RecordLastChannel records the last active channel for this workspace.
|
||||
// This uses the atomic state save mechanism to prevent data loss on crash.
|
||||
func (al *AgentLoop) RecordLastChannel(channel string) error {
|
||||
return al.state.SetLastChannel(channel)
|
||||
func (al *AgentLoop) RecordLastChannel(ctx context.Context, channel string) error {
|
||||
return al.state.SetLastChannel(ctx, channel)
|
||||
}
|
||||
|
||||
// RecordLastChatID records the last active chat ID for this workspace.
|
||||
// This uses the atomic state save mechanism to prevent data loss on crash.
|
||||
func (al *AgentLoop) RecordLastChatID(chatID string) error {
|
||||
return al.state.SetLastChatID(chatID)
|
||||
func (al *AgentLoop) RecordLastChatID(ctx context.Context, chatID string) error {
|
||||
return al.state.SetLastChatID(ctx, chatID)
|
||||
}
|
||||
|
||||
func (al *AgentLoop) ProcessDirect(ctx context.Context, content, sessionKey string) (string, error) {
|
||||
|
|
@ -560,27 +554,30 @@ func (al *AgentLoop) processSystemMessage(_ context.Context, msg bus.InboundMess
|
|||
return "", nil
|
||||
}
|
||||
|
||||
// runAgentLoop is the core message processing logic.
|
||||
// It handles context building, Fantasy agent creation, tool execution, and response handling.
|
||||
// When opts.Streaming is true, delegates to runAgentLoopStreaming for real-time token delivery.
|
||||
func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (string, error) {
|
||||
al.activeSessionKey.Store(opts.SessionKey)
|
||||
// assembledContext holds the pre-processed context produced by assembleContext,
|
||||
// consumed by both the Generate and Stream code paths.
|
||||
type assembledContext struct {
|
||||
systemPrompt string
|
||||
userPrompt string
|
||||
fantasyHistory []fantasy.Message
|
||||
adaptedTools []fantasy.AgentTool
|
||||
agent fantasy.Agent
|
||||
}
|
||||
|
||||
if opts.Streaming {
|
||||
return al.runAgentLoopStreaming(ctx, opts)
|
||||
}
|
||||
// 0. Record last channel for heartbeat notifications (skip internal channels)
|
||||
// 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 {
|
||||
if opts.Channel != "" && opts.ChatID != "" {
|
||||
if !constants.IsInternalChannel(opts.Channel) {
|
||||
channelKey := fmt.Sprintf("%s:%s", opts.Channel, opts.ChatID)
|
||||
if err := al.RecordLastChannel(channelKey); err != nil {
|
||||
if err := al.RecordLastChannel(ctx, channelKey); err != nil {
|
||||
logger.WarnCF("agent", "Failed to record last channel: %v", map[string]interface{}{"error": err.Error()})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1. Update tool contexts
|
||||
logger.DebugCF("agent", "runAgentLoop: starting",
|
||||
logger.DebugCF("agent", "assembleContext: starting",
|
||||
map[string]interface{}{
|
||||
"session_key": opts.SessionKey,
|
||||
"channel": opts.Channel,
|
||||
|
|
@ -588,19 +585,12 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
|
|||
})
|
||||
al.updateToolContexts(opts.Channel, opts.ChatID)
|
||||
|
||||
// 2. Load observation block for system prompt injection
|
||||
if al.obsManager != nil {
|
||||
block := al.obsManager.LoadBlock(ctx, opts.SessionKey)
|
||||
al.contextBuilder.SetObservationBlock(block)
|
||||
}
|
||||
block := al.obsManager.LoadBlock(ctx, opts.SessionKey)
|
||||
al.contextBuilder.SetObservationBlock(block)
|
||||
|
||||
// 2b. Load knowledge block from completed Focus sessions
|
||||
if al.memDelegate != nil {
|
||||
kb := tools.LoadKnowledgeBlock(ctx, al.memDelegate, opts.SessionKey)
|
||||
al.contextBuilder.SetKnowledgeBlock(kb)
|
||||
}
|
||||
kb := tools.LoadKnowledgeBlock(ctx, al.memDelegate, opts.SessionKey)
|
||||
al.contextBuilder.SetKnowledgeBlock(kb)
|
||||
|
||||
// 3. Build messages with DAG compression (skip history for heartbeat)
|
||||
var history []messages.Message
|
||||
var summary string
|
||||
if !opts.NoHistory {
|
||||
|
|
@ -608,43 +598,32 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
|
|||
summary = al.sessions.GetSummary(opts.SessionKey)
|
||||
}
|
||||
|
||||
// 3a. DAG compression: compress old history, keep raw tail
|
||||
history = al.applyDAGCompression(history)
|
||||
|
||||
builtMsgs := al.contextBuilder.BuildMessages(
|
||||
history,
|
||||
summary,
|
||||
opts.UserMessage,
|
||||
nil,
|
||||
opts.Channel,
|
||||
opts.ChatID,
|
||||
)
|
||||
if al.identitySync != nil {
|
||||
_ = al.identitySync.CheckAndSync(ctx)
|
||||
}
|
||||
|
||||
// 3b. Save user message to session
|
||||
builtMsgs := al.contextBuilder.BuildMessages(history, summary, opts.UserMessage, nil, opts.Channel, opts.ChatID)
|
||||
al.sessions.AddMessage(opts.SessionKey, "user", opts.UserMessage)
|
||||
|
||||
// 4. Split built messages into system prompt, conversation history, and current user prompt.
|
||||
// BuildMessages returns: [system, ...history, user]
|
||||
systemPrompt := ""
|
||||
var historyMsgs []messages.Message
|
||||
userPrompt := opts.UserMessage
|
||||
|
||||
if len(builtMsgs) > 0 && builtMsgs[0].Role == "system" {
|
||||
systemPrompt = builtMsgs[0].Content
|
||||
// History is everything between system and last user message.
|
||||
if len(builtMsgs) > 2 {
|
||||
historyMsgs = builtMsgs[1 : len(builtMsgs)-1]
|
||||
}
|
||||
}
|
||||
|
||||
// 5. Convert history to Fantasy message format
|
||||
logger.DebugCF("agent", "runAgentLoop: history messages",
|
||||
logger.DebugCF("agent", "assembleContext: history messages",
|
||||
map[string]interface{}{
|
||||
"history": formatMessagesForLog(historyMsgs),
|
||||
})
|
||||
fantasyHistory := picofantasy.MessagesToFantasy(historyMsgs)
|
||||
|
||||
// 6. Build adapted tools from PicoClaw registry (with optional offloading)
|
||||
adaptCfg := picofantasy.AdaptedToolsConfig{
|
||||
MemStore: al.memoryStore,
|
||||
AgentID: "picoclaw",
|
||||
|
|
@ -652,7 +631,6 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
|
|||
}
|
||||
adaptedTools := picofantasy.BuildAdaptedTools(al.tools, al.bus, opts.Channel, opts.ChatID, adaptCfg)
|
||||
|
||||
// 7. Create Fantasy agent with tools and configuration
|
||||
agentOpts := []fantasy.AgentOption{
|
||||
fantasy.WithTools(adaptedTools...),
|
||||
fantasy.WithStopConditions(fantasy.StepCountIs(al.maxIterations)),
|
||||
|
|
@ -660,7 +638,6 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
|
|||
if systemPrompt != "" {
|
||||
agentOpts = append(agentOpts, fantasy.WithSystemPrompt(systemPrompt))
|
||||
}
|
||||
// Attach SecureBusToolRuntime when ITR is enabled (non-nil bus).
|
||||
if al.secureBus != nil {
|
||||
sbrt := SecureBusToolRuntime{
|
||||
Bus: al.secureBus,
|
||||
|
|
@ -676,55 +653,34 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
|
|||
"tools_count": len(adaptedTools),
|
||||
"history_count": len(historyMsgs),
|
||||
"max_iterations": al.maxIterations,
|
||||
"memory_enabled": al.memoryStore != nil,
|
||||
"memory_enabled": true,
|
||||
})
|
||||
|
||||
// 8. Call Fantasy agent.Generate()
|
||||
result, err := agent.Generate(ctx, fantasy.AgentCall{
|
||||
Prompt: userPrompt,
|
||||
Messages: fantasyHistory,
|
||||
})
|
||||
if err != nil {
|
||||
logger.ErrorCF("agent", "Fantasy Generate failed",
|
||||
map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
return "", fmt.Errorf("agent Generate failed: %w", err)
|
||||
return assembledContext{
|
||||
systemPrompt: systemPrompt,
|
||||
userPrompt: userPrompt,
|
||||
fantasyHistory: fantasyHistory,
|
||||
adaptedTools: adaptedTools,
|
||||
agent: agent,
|
||||
}
|
||||
}
|
||||
|
||||
// 9. Save all step messages to session and audit tool calls
|
||||
stepCount := len(result.Steps)
|
||||
for _, step := range result.Steps {
|
||||
stepMsgs := picofantasy.StepToMessages(step)
|
||||
for _, m := range stepMsgs {
|
||||
al.sessions.AddFullMessage(opts.SessionKey, m)
|
||||
}
|
||||
al.auditStep(ctx, step, opts.SessionKey)
|
||||
}
|
||||
|
||||
// 10. Extract final text
|
||||
finalContent := result.Response.Content.Text()
|
||||
|
||||
// 11. Handle empty response
|
||||
// postProcess handles the common finalization after Generate or Stream:
|
||||
// extract final text, save session, summarize, observe, optionally send response.
|
||||
func (al *AgentLoop) postProcess(ctx context.Context, opts processOptions, finalContent string, stepCount int) string {
|
||||
if finalContent == "" {
|
||||
finalContent = opts.DefaultResponse
|
||||
}
|
||||
|
||||
// 12. Save session
|
||||
al.sessions.Save(opts.SessionKey)
|
||||
|
||||
// 13. Optional: summarization
|
||||
if opts.EnableSummary {
|
||||
al.maybeSummarize(opts.SessionKey, opts.Channel, opts.ChatID)
|
||||
al.maybeSummarize(ctx, opts.SessionKey, opts.Channel, opts.ChatID)
|
||||
}
|
||||
|
||||
// 13b. Trigger async observation if tail exceeds token threshold
|
||||
if al.obsManager != nil {
|
||||
tail := al.sessionsToMessagePairs(opts.SessionKey)
|
||||
al.obsManager.MaybeObserveAsync(ctx, opts.SessionKey, tail)
|
||||
}
|
||||
tail := al.sessionsToMessagePairs(opts.SessionKey)
|
||||
al.obsManager.MaybeObserveAsync(ctx, opts.SessionKey, tail)
|
||||
|
||||
// 14. Optional: send response via bus
|
||||
if opts.SendResponse {
|
||||
al.bus.PublishOutbound(bus.OutboundMessage{
|
||||
Channel: opts.Channel,
|
||||
|
|
@ -733,7 +689,6 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
|
|||
})
|
||||
}
|
||||
|
||||
// 15. Log response
|
||||
responsePreview := utils.Truncate(finalContent, 120)
|
||||
logger.InfoCF("agent", fmt.Sprintf("Response: %s", responsePreview),
|
||||
map[string]interface{}{
|
||||
|
|
@ -742,110 +697,50 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (str
|
|||
"final_length": len(finalContent),
|
||||
})
|
||||
|
||||
return finalContent, nil
|
||||
return finalContent
|
||||
}
|
||||
|
||||
// runAgentLoopStreaming uses Fantasy's agent.Stream() to stream token deltas
|
||||
// to the bus in real time. Structure mirrors runAgentLoop but uses AgentStreamCall
|
||||
// with OnTextDelta, OnStepFinish, and OnToolCall callbacks.
|
||||
func (al *AgentLoop) runAgentLoopStreaming(ctx context.Context, opts processOptions) (string, error) {
|
||||
// 0. Record last channel
|
||||
if opts.Channel != "" && opts.ChatID != "" {
|
||||
if !constants.IsInternalChannel(opts.Channel) {
|
||||
channelKey := fmt.Sprintf("%s:%s", opts.Channel, opts.ChatID)
|
||||
if err := al.RecordLastChannel(channelKey); err != nil {
|
||||
logger.WarnCF("agent", "Failed to record last channel: %v", map[string]interface{}{"error": err.Error()})
|
||||
}
|
||||
// runAgentLoop is the core message processing logic.
|
||||
// It delegates to assembleContext for shared pre-processing, then branches on
|
||||
// opts.Streaming to either Generate (synchronous) or Stream (real-time deltas).
|
||||
func (al *AgentLoop) runAgentLoop(ctx context.Context, opts processOptions) (string, error) {
|
||||
al.activeSessionKey.Store(opts.SessionKey)
|
||||
|
||||
ac := al.assembleContext(ctx, opts)
|
||||
|
||||
if opts.Streaming {
|
||||
return al.runStreaming(ctx, opts, ac)
|
||||
}
|
||||
|
||||
result, err := ac.agent.Generate(ctx, fantasy.AgentCall{
|
||||
Prompt: ac.userPrompt,
|
||||
Messages: ac.fantasyHistory,
|
||||
})
|
||||
if err != nil {
|
||||
logger.ErrorCF("agent", "Fantasy Generate failed",
|
||||
map[string]interface{}{"error": err.Error()})
|
||||
return "", fmt.Errorf("agent Generate failed: %w", err)
|
||||
}
|
||||
|
||||
for _, step := range result.Steps {
|
||||
stepMsgs := picofantasy.StepToMessages(step)
|
||||
for _, m := range stepMsgs {
|
||||
al.sessions.AddFullMessage(opts.SessionKey, m)
|
||||
}
|
||||
al.auditStep(ctx, step, opts.SessionKey)
|
||||
}
|
||||
|
||||
// 1. Update tool contexts
|
||||
al.updateToolContexts(opts.Channel, opts.ChatID)
|
||||
finalContent := result.Response.Content.Text()
|
||||
return al.postProcess(ctx, opts, finalContent, len(result.Steps)), nil
|
||||
}
|
||||
|
||||
// 2. Load observation block for system prompt injection
|
||||
if al.obsManager != nil {
|
||||
block := al.obsManager.LoadBlock(ctx, opts.SessionKey)
|
||||
al.contextBuilder.SetObservationBlock(block)
|
||||
}
|
||||
|
||||
// 2b. Load knowledge block from completed Focus sessions
|
||||
if al.memDelegate != nil {
|
||||
kb := tools.LoadKnowledgeBlock(ctx, al.memDelegate, opts.SessionKey)
|
||||
al.contextBuilder.SetKnowledgeBlock(kb)
|
||||
}
|
||||
|
||||
// 3. Build messages with DAG compression
|
||||
var history []messages.Message
|
||||
var summary string
|
||||
if !opts.NoHistory {
|
||||
history = al.sessions.GetHistory(opts.SessionKey)
|
||||
summary = al.sessions.GetSummary(opts.SessionKey)
|
||||
}
|
||||
|
||||
// 3a. DAG compression
|
||||
history = al.applyDAGCompression(history)
|
||||
|
||||
builtMsgs := al.contextBuilder.BuildMessages(history, summary, opts.UserMessage, nil, opts.Channel, opts.ChatID)
|
||||
|
||||
// 4. Save user message
|
||||
al.sessions.AddMessage(opts.SessionKey, "user", opts.UserMessage)
|
||||
|
||||
// 5. Split into system/history/user
|
||||
systemPrompt := ""
|
||||
var historyMsgs []messages.Message
|
||||
userPrompt := opts.UserMessage
|
||||
|
||||
if len(builtMsgs) > 0 && builtMsgs[0].Role == "system" {
|
||||
systemPrompt = builtMsgs[0].Content
|
||||
if len(builtMsgs) > 2 {
|
||||
historyMsgs = builtMsgs[1 : len(builtMsgs)-1]
|
||||
}
|
||||
}
|
||||
|
||||
// 5. Convert history
|
||||
fantasyHistory := picofantasy.MessagesToFantasy(historyMsgs)
|
||||
|
||||
// 6. Build adapted tools (with optional offloading)
|
||||
streamAdaptCfg := picofantasy.AdaptedToolsConfig{
|
||||
MemStore: al.memoryStore,
|
||||
AgentID: "picoclaw",
|
||||
SessionKey: opts.SessionKey,
|
||||
}
|
||||
adaptedTools := picofantasy.BuildAdaptedTools(al.tools, al.bus, opts.Channel, opts.ChatID, streamAdaptCfg)
|
||||
|
||||
// 7. Create Fantasy agent
|
||||
agentOpts := []fantasy.AgentOption{
|
||||
fantasy.WithTools(adaptedTools...),
|
||||
fantasy.WithStopConditions(fantasy.StepCountIs(al.maxIterations)),
|
||||
}
|
||||
if systemPrompt != "" {
|
||||
agentOpts = append(agentOpts, fantasy.WithSystemPrompt(systemPrompt))
|
||||
}
|
||||
// Attach SecureBusToolRuntime when ITR is enabled (non-nil bus).
|
||||
if al.secureBus != nil {
|
||||
sbrt := SecureBusToolRuntime{
|
||||
Bus: al.secureBus,
|
||||
SessionKey: opts.SessionKey,
|
||||
}
|
||||
agentOpts = append(agentOpts, fantasy.WithToolRuntime(sbrt))
|
||||
}
|
||||
fantasyAgent := fantasy.NewAgent(al.languageModel, agentOpts...)
|
||||
|
||||
logger.DebugCF("agent", "Fantasy streaming agent created",
|
||||
map[string]interface{}{
|
||||
"model": al.model,
|
||||
"tools_count": len(adaptedTools),
|
||||
"history_count": len(historyMsgs),
|
||||
"max_iterations": al.maxIterations,
|
||||
"memory_enabled": al.memoryStore != nil,
|
||||
})
|
||||
|
||||
// 8. Build streaming call with callbacks
|
||||
// runStreaming uses Fantasy's agent.Stream() to stream token deltas to the bus
|
||||
// in real time, using the pre-assembled context from assembleContext.
|
||||
func (al *AgentLoop) runStreaming(ctx context.Context, opts processOptions, ac assembledContext) (string, error) {
|
||||
streamCall := fantasy.AgentStreamCall{
|
||||
Prompt: userPrompt,
|
||||
Messages: fantasyHistory,
|
||||
Prompt: ac.userPrompt,
|
||||
Messages: ac.fantasyHistory,
|
||||
|
||||
// Stream text deltas to bus in real time
|
||||
OnTextDelta: func(id, text string) error {
|
||||
if opts.Channel != "" && opts.ChatID != "" {
|
||||
al.bus.PublishOutbound(bus.OutboundMessage{
|
||||
|
|
@ -867,7 +762,6 @@ func (al *AgentLoop) runAgentLoopStreaming(ctx context.Context, opts processOpti
|
|||
return nil
|
||||
},
|
||||
|
||||
// Log tool calls as they happen
|
||||
OnToolCall: func(tc fantasy.ToolCallContent) error {
|
||||
logger.DebugCF("agent", "Streaming tool call",
|
||||
map[string]interface{}{
|
||||
|
|
@ -878,56 +772,21 @@ func (al *AgentLoop) runAgentLoopStreaming(ctx context.Context, opts processOpti
|
|||
},
|
||||
}
|
||||
|
||||
// 9. Call Fantasy agent.Stream()
|
||||
result, err := fantasyAgent.Stream(ctx, streamCall)
|
||||
result, err := ac.agent.Stream(ctx, streamCall)
|
||||
if err != nil {
|
||||
logger.ErrorCF("agent", "Fantasy Stream failed",
|
||||
map[string]interface{}{"error": err.Error()})
|
||||
return "", fmt.Errorf("agent Stream failed: %w", err)
|
||||
}
|
||||
|
||||
// 10. Extract final text
|
||||
finalContent := result.Response.Content.Text()
|
||||
if finalContent == "" {
|
||||
finalContent = opts.DefaultResponse
|
||||
}
|
||||
|
||||
// 11. Save session
|
||||
al.sessions.Save(opts.SessionKey)
|
||||
|
||||
// 12. Summarization
|
||||
if opts.EnableSummary {
|
||||
al.maybeSummarize(opts.SessionKey, opts.Channel, opts.ChatID)
|
||||
}
|
||||
|
||||
// 12b. Trigger async observation if tail exceeds token threshold
|
||||
if al.obsManager != nil {
|
||||
tail := al.sessionsToMessagePairs(opts.SessionKey)
|
||||
al.obsManager.MaybeObserveAsync(ctx, opts.SessionKey, tail)
|
||||
}
|
||||
|
||||
// 13. Log response
|
||||
stepCount := len(result.Steps)
|
||||
responsePreview := utils.Truncate(finalContent, 120)
|
||||
logger.InfoCF("agent", fmt.Sprintf("Streaming response: %s", responsePreview),
|
||||
map[string]interface{}{
|
||||
"session_key": opts.SessionKey,
|
||||
"steps": stepCount,
|
||||
"final_length": len(finalContent),
|
||||
"total_tokens": result.TotalUsage.TotalTokens,
|
||||
})
|
||||
|
||||
return finalContent, nil
|
||||
return al.postProcess(ctx, opts, finalContent, len(result.Steps)), nil
|
||||
}
|
||||
|
||||
// runLLMIteration — DELETED. Replaced by Fantasy's internal agent loop.
|
||||
|
||||
// auditStep logs tool calls from a Fantasy step result to the audit log.
|
||||
func (al *AgentLoop) auditStep(ctx context.Context, step fantasy.StepResult, sessionKey string) {
|
||||
if al.memDelegate == nil {
|
||||
return
|
||||
}
|
||||
|
||||
toolCalls := step.Content.ToolCalls()
|
||||
if len(toolCalls) == 0 {
|
||||
return
|
||||
|
|
@ -974,13 +833,12 @@ 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.
|
||||
func (al *AgentLoop) maybeSummarize(sessionKey, channel, chatID string) {
|
||||
func (al *AgentLoop) maybeSummarize(ctx context.Context, sessionKey, channel, chatID string) {
|
||||
newHistory := al.sessions.GetHistory(sessionKey)
|
||||
tokenEstimate := al.estimateTokens(newHistory)
|
||||
threshold := al.contextWindow * 75 / 100
|
||||
criticalThreshold := al.contextWindow * 95 / 100
|
||||
|
||||
// Emergency path: drop oldest messages immediately when near context limit.
|
||||
if tokenEstimate > criticalThreshold {
|
||||
al.forceCompression(sessionKey)
|
||||
return
|
||||
|
|
@ -990,7 +848,6 @@ func (al *AgentLoop) maybeSummarize(sessionKey, channel, chatID string) {
|
|||
if _, loading := al.summarizing.LoadOrStore(sessionKey, true); !loading {
|
||||
go func() {
|
||||
defer al.summarizing.Delete(sessionKey)
|
||||
// Notify user about optimization if not an internal channel
|
||||
if !constants.IsInternalChannel(channel) {
|
||||
al.bus.PublishOutbound(bus.OutboundMessage{
|
||||
Channel: channel,
|
||||
|
|
@ -998,7 +855,7 @@ func (al *AgentLoop) maybeSummarize(sessionKey, channel, chatID string) {
|
|||
Content: "⚠️ Memory threshold reached. Optimizing conversation history...",
|
||||
})
|
||||
}
|
||||
al.summarizeSession(sessionKey)
|
||||
al.summarizeSession(ctx, sessionKey)
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
|
@ -1119,8 +976,8 @@ func formatMessagesForLog(msgs []messages.Message) string {
|
|||
}
|
||||
|
||||
// summarizeSession summarizes the conversation history for a session.
|
||||
func (al *AgentLoop) summarizeSession(sessionKey string) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
|
||||
func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey string) {
|
||||
ctx, cancel := context.WithTimeout(parentCtx, 120*time.Second)
|
||||
defer cancel()
|
||||
|
||||
history := al.sessions.GetHistory(sessionKey)
|
||||
|
|
|
|||
|
|
@ -4,7 +4,6 @@ import (
|
|||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
|
|
@ -14,6 +13,16 @@ import (
|
|||
"github.com/sipeed/picoclaw/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()
|
||||
al, err := NewAgentLoop(context.Background(), cfg, msgBus, model)
|
||||
if err != nil {
|
||||
t.Fatalf("NewAgentLoop: %v", err)
|
||||
}
|
||||
return al
|
||||
}
|
||||
|
||||
// mockLanguageModel is a simple mock fantasy.LanguageModel for testing
|
||||
type mockLanguageModel struct {
|
||||
response string
|
||||
|
|
@ -76,11 +85,11 @@ func TestRecordLastChannel(t *testing.T) {
|
|||
// Create agent loop
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
// Test RecordLastChannel
|
||||
testChannel := "test-channel"
|
||||
err = al.RecordLastChannel(testChannel)
|
||||
err = al.RecordLastChannel(context.Background(), testChannel)
|
||||
if err != nil {
|
||||
t.Fatalf("RecordLastChannel failed: %v", err)
|
||||
}
|
||||
|
|
@ -92,7 +101,7 @@ func TestRecordLastChannel(t *testing.T) {
|
|||
}
|
||||
|
||||
// Verify persistence by creating a new agent loop
|
||||
al2 := NewAgentLoop(cfg, msgBus, model)
|
||||
al2 := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
if al2.state.GetLastChannel() != testChannel {
|
||||
t.Errorf("Expected persistent channel '%s', got '%s'", testChannel, al2.state.GetLastChannel())
|
||||
}
|
||||
|
|
@ -121,11 +130,11 @@ func TestRecordLastChatID(t *testing.T) {
|
|||
// Create agent loop
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
// Test RecordLastChatID
|
||||
testChatID := "test-chat-id-123"
|
||||
err = al.RecordLastChatID(testChatID)
|
||||
err = al.RecordLastChatID(context.Background(), testChatID)
|
||||
if err != nil {
|
||||
t.Fatalf("RecordLastChatID failed: %v", err)
|
||||
}
|
||||
|
|
@ -137,7 +146,7 @@ func TestRecordLastChatID(t *testing.T) {
|
|||
}
|
||||
|
||||
// Verify persistence by creating a new agent loop
|
||||
al2 := NewAgentLoop(cfg, msgBus, model)
|
||||
al2 := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
if al2.state.GetLastChatID() != testChatID {
|
||||
t.Errorf("Expected persistent chat ID '%s', got '%s'", testChatID, al2.state.GetLastChatID())
|
||||
}
|
||||
|
|
@ -166,18 +175,12 @@ func TestNewAgentLoop_StateInitialized(t *testing.T) {
|
|||
// Create agent loop
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
// Verify state manager is initialized
|
||||
// Verify state manager is initialized (delegate-backed via always-on memory)
|
||||
if al.state == nil {
|
||||
t.Error("Expected state manager to be initialized")
|
||||
}
|
||||
|
||||
// Verify state directory was created
|
||||
stateDir := filepath.Join(tmpDir, "state")
|
||||
if _, err := os.Stat(stateDir); os.IsNotExist(err) {
|
||||
t.Error("Expected state directory to exist")
|
||||
}
|
||||
}
|
||||
|
||||
// TestToolRegistry_ToolRegistration verifies tools can be registered and retrieved
|
||||
|
|
@ -201,7 +204,7 @@ func TestToolRegistry_ToolRegistration(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
// Register a custom tool
|
||||
customTool := &mockCustomTool{}
|
||||
|
|
@ -247,7 +250,7 @@ func TestToolContext_Updates(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("OK")
|
||||
_ = NewAgentLoop(cfg, msgBus, model)
|
||||
_ = mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
// Verify that ContextualTool interface is defined and can be implemented
|
||||
// This test validates the interface contract exists
|
||||
|
|
@ -278,7 +281,7 @@ func TestToolRegistry_GetDefinitions(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
// Register a test tool and verify it shows up in startup info
|
||||
testTool := &mockCustomTool{}
|
||||
|
|
@ -322,7 +325,7 @@ func TestAgentLoop_GetStartupInfo(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
info := al.GetStartupInfo()
|
||||
|
||||
|
|
@ -369,7 +372,7 @@ func TestAgentLoop_Stop(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
|
||||
// Note: running is only set to true when Run() is called
|
||||
// We can't test that without starting the event loop
|
||||
|
|
@ -476,7 +479,7 @@ func TestToolResult_SilentToolDoesNotSendUserMessage(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("File operation complete")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
helper := testHelper{al: al}
|
||||
|
||||
// ReadFileTool returns SilentResult, which should not send user message
|
||||
|
|
@ -518,7 +521,7 @@ func TestToolResult_UserFacingToolDoesSendMessage(t *testing.T) {
|
|||
|
||||
msgBus := bus.NewMessageBus()
|
||||
model := newMockLanguageModel("Command output: hello world")
|
||||
al := NewAgentLoop(cfg, msgBus, model)
|
||||
al := mustNewAgentLoop(t, cfg, msgBus, model)
|
||||
helper := testHelper{al: al}
|
||||
|
||||
// ExecTool returns UserResult, which should send user message
|
||||
|
|
|
|||
80
pkg/agent/toolloop.go
Normal file
80
pkg/agent/toolloop.go
Normal file
|
|
@ -0,0 +1,80 @@
|
|||
// PicoClaw - Ultra-lightweight personal AI agent
|
||||
// License: MIT
|
||||
//
|
||||
// Copyright (c) 2026 PicoClaw contributors
|
||||
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
fantasy "charm.land/fantasy"
|
||||
picofantasy "github.com/sipeed/picoclaw/pkg/fantasy"
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
memstore "github.com/sipeed/picoclaw/pkg/memory/store"
|
||||
"github.com/sipeed/picoclaw/pkg/tools"
|
||||
)
|
||||
|
||||
// RunToolLoop executes an agent tool loop using Fantasy with the canonical
|
||||
// PicoToolAdapter (schema unwrapping + offloading). This is the single
|
||||
// implementation used by both main agent and subagents.
|
||||
func RunToolLoop(ctx context.Context, config tools.ToolLoopConfig, systemPrompt, userPrompt, channel, chatID string) (*tools.ToolLoopResult, error) {
|
||||
return runToolLoopWithMem(ctx, config, systemPrompt, userPrompt, channel, chatID, nil)
|
||||
}
|
||||
|
||||
// MakeRunLoopFunc returns a RunLoopFunc that uses the given MemoryStore for
|
||||
// tool result offloading. This is wired into SubagentManager so subagent tool
|
||||
// results get offloaded to archival memory.
|
||||
func MakeRunLoopFunc(ms *memstore.MemoryStore) tools.RunLoopFunc {
|
||||
return func(ctx context.Context, config tools.ToolLoopConfig, systemPrompt, userPrompt, channel, chatID string) (*tools.ToolLoopResult, error) {
|
||||
return runToolLoopWithMem(ctx, config, systemPrompt, userPrompt, channel, chatID, ms)
|
||||
}
|
||||
}
|
||||
|
||||
func runToolLoopWithMem(ctx context.Context, config tools.ToolLoopConfig, systemPrompt, userPrompt, channel, chatID string, ms *memstore.MemoryStore) (*tools.ToolLoopResult, error) {
|
||||
adaptCfg := picofantasy.AdaptedToolsConfig{
|
||||
MemStore: ms,
|
||||
AgentID: "picoclaw",
|
||||
SessionKey: "",
|
||||
}
|
||||
adaptedTools := picofantasy.BuildAdaptedTools(config.Tools, config.Bus, channel, chatID, adaptCfg)
|
||||
|
||||
agentOpts := []fantasy.AgentOption{
|
||||
fantasy.WithTools(adaptedTools...),
|
||||
fantasy.WithStopConditions(fantasy.StepCountIs(config.MaxIterations)),
|
||||
}
|
||||
if systemPrompt != "" {
|
||||
agentOpts = append(agentOpts, fantasy.WithSystemPrompt(systemPrompt))
|
||||
}
|
||||
agent := fantasy.NewAgent(config.Model, agentOpts...)
|
||||
|
||||
logger.DebugCF("toolloop", "Agent created",
|
||||
map[string]any{
|
||||
"tools_count": len(adaptedTools),
|
||||
"max_iterations": config.MaxIterations,
|
||||
})
|
||||
|
||||
result, err := agent.Generate(ctx, fantasy.AgentCall{
|
||||
Prompt: userPrompt,
|
||||
})
|
||||
if err != nil {
|
||||
logger.ErrorCF("toolloop", "Fantasy agent.Generate failed",
|
||||
map[string]any{"error": err.Error()})
|
||||
return nil, fmt.Errorf("agent Generate failed: %w", err)
|
||||
}
|
||||
|
||||
finalContent := result.Response.Content.Text()
|
||||
stepCount := len(result.Steps)
|
||||
|
||||
logger.InfoCF("toolloop", "Tool loop completed",
|
||||
map[string]any{
|
||||
"steps": stepCount,
|
||||
"content_chars": len(finalContent),
|
||||
})
|
||||
|
||||
return &tools.ToolLoopResult{
|
||||
Content: finalContent,
|
||||
Iterations: stepCount,
|
||||
}, nil
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue