diff --git a/.gitignore b/.gitignore index 72f3b1761..9fe252222 100644 --- a/.gitignore +++ b/.gitignore @@ -30,6 +30,9 @@ config/config.json coverage.txt coverage.html +# Dependencies +node_modules/ + # OS .DS_Store diff --git a/cmd/picoclaw/internal/gateway/command.go b/cmd/picoclaw/internal/gateway/command.go index 06e170957..14080c583 100644 --- a/cmd/picoclaw/internal/gateway/command.go +++ b/cmd/picoclaw/internal/gateway/command.go @@ -36,7 +36,10 @@ func NewGatewayCommand() *cobra.Command { return nil }, RunE: func(_ *cobra.Command, _ []string) error { - return gateway.Run(debug, internal.GetPicoclawHome(), internal.GetConfigPath(), orchestration, enableStats, allowEmpty) + return gateway.Run( + debug, internal.GetPicoclawHome(), internal.GetConfigPath(), + orchestration, enableStats, allowEmpty, + ) }, } diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index ca238a875..78dbeb26d 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -57,9 +57,9 @@ type AgentLoop struct { cmdRegistry *commands.Registry mcp mcpRuntime hookRuntime hookRuntime - steering *steeringQueue - pendingSkills sync.Map - mu sync.RWMutex + steering *steeringQueue + pendingSkills sync.Map // sessionKey → skillName (armed by /use ) + mu sync.RWMutex providerCache map[string]providers.LLMProvider @@ -103,12 +103,6 @@ type processOptions struct { SystemMessage bool // If true, this is a system message (subagent result) — skip placeholder and plan nudge } -type continuationTarget struct { - SessionKey string - Channel string - ChatID string -} - const ( defaultResponse = "The model returned an empty response. This may indicate a provider error or token limit." toolLimitResponse = "I've reached `max_tool_iterations` without a final response. Increase `max_tool_iterations` in config.json if this task needs more tool steps." @@ -609,60 +603,6 @@ func (al *AgentLoop) Stop() { al.running.Store(false) } -func (al *AgentLoop) publishResponseIfNeeded(ctx context.Context, channel, chatID, response string) { - if response == "" { - return - } - - alreadySent := false - defaultAgent := al.GetRegistry().GetDefaultAgent() - if defaultAgent != nil { - if tool, ok := defaultAgent.Tools.Get("message"); ok { - if mt, ok := tool.(*tools.MessageTool); ok { - alreadySent = mt.HasSentInRound() - } - } - } - - if alreadySent { - logger.DebugCF( - "agent", - "Skipped outbound (message tool already sent)", - map[string]any{"channel": channel}, - ) - return - } - - al.bus.PublishOutbound(ctx, bus.OutboundMessage{ - Channel: channel, - ChatID: chatID, - Content: response, - }) - logger.InfoCF("agent", "Published outbound response", - map[string]any{ - "channel": channel, - "chat_id": chatID, - "content_len": len(response), - }) -} - -func (al *AgentLoop) buildContinuationTarget(msg bus.InboundMessage) (*continuationTarget, error) { - if msg.Channel == "system" { - return nil, nil - } - - route, _, err := al.resolveMessageRoute(msg) - if err != nil { - return nil, err - } - - return &continuationTarget{ - SessionKey: resolveScopeKey(route, msg.SessionKey), - Channel: msg.Channel, - ChatID: msg.ChatID, - }, nil -} - // Close releases resources held by the loop (e.g. flushes write-behind stats // and dirty session data). Should be called during graceful shutdown. func (al *AgentLoop) Close() { @@ -747,17 +687,6 @@ func (al *AgentLoop) newTurnEventScope(agentID, sessionKey string) turnEventScop } } -func (ts turnEventScope) meta(iteration int, source, tracePath string) EventMeta { - return EventMeta{ - AgentID: ts.agentID, - TurnID: ts.turnID, - SessionKey: ts.sessionKey, - Iteration: iteration, - Source: source, - TracePath: tracePath, - } -} - func (al *AgentLoop) emitEvent(kind EventKind, meta EventMeta, payload any) { evt := Event{ Kind: kind, @@ -774,43 +703,6 @@ func (al *AgentLoop) emitEvent(kind EventKind, meta EventMeta, payload any) { al.eventBus.Emit(evt) } -func cloneEventArguments(args map[string]any) map[string]any { - if len(args) == 0 { - return nil - } - - cloned := make(map[string]any, len(args)) - for k, v := range args { - cloned[k] = v - } - return cloned -} - -func (al *AgentLoop) hookAbortError(ts *turnState, stage string, decision HookDecision) error { - reason := decision.Reason - if reason == "" { - reason = "hook requested turn abort" - } - - err := fmt.Errorf("hook aborted turn during %s: %s", stage, reason) - al.emitEvent( - EventKindError, - ts.eventMeta("hooks", "turn.error"), - ErrorPayload{ - Stage: "hook." + stage, - Message: err.Error(), - }, - ) - return err -} - -func hookDeniedToolContent(prefix, reason string) string { - if reason == "" { - return prefix - } - return prefix + ": " + reason -} - func (al *AgentLoop) logEvent(evt Event) { fields := map[string]any{ "event_kind": evt.Kind.String(), @@ -1432,8 +1324,8 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) al.OnUserMessage() } - // Expand fork-specific /skill and /plan commands - expansionCompact := al.expandForkCommands(&msg) + // Expand fork-specific /skill, /use, and /plan commands + expansionCompact, forcedSkills := al.expandForkCommands(&msg) // Check for commands (using default agent, before routing) if response, handled := al.handleCommand(ctx, msg, al.registry.GetDefaultAgent(), msg.SessionKey); handled { @@ -1448,6 +1340,18 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) sessionKey := resolveScopeKey(route, msg.SessionKey) + // Consume armed skill from a previous /use command + armKey := msg.Channel + ":" + msg.ChatID + if val, ok := al.pendingSkills.LoadAndDelete(sessionKey); ok { + if skillName, ok := val.(string); ok { + forcedSkills = append(forcedSkills, skillName) + } + } else if val, ok := al.pendingSkills.LoadAndDelete(armKey); ok { + if skillName, ok := val.(string); ok { + forcedSkills = append(forcedSkills, skillName) + } + } + return al.runAgentLoop(ctx, agent, processOptions{ SessionKey: sessionKey, Channel: msg.Channel, @@ -1455,6 +1359,7 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) SenderID: msg.SenderID, SenderDisplayName: msg.Sender.DisplayName, UserMessage: msg.Content, + ForcedSkills: forcedSkills, Media: msg.Media, HistoryMessage: expansionCompact, DefaultResponse: defaultResponse, @@ -1503,6 +1408,7 @@ func (al *AgentLoop) callLLMWithRetry( activeModel string, onChunk func(string, string), iteration int, + scope ...turnEventScope, ) (*providers.LLMResponse, error) { llmOpts := map[string]any{ "max_tokens": agent.MaxTokens, @@ -1597,6 +1503,15 @@ func (al *AgentLoop) callLLMWithRetry( strings.Contains(errMsg, "prompt is too long") || strings.Contains(errMsg, "request too large")) + // Helper to emit events if scope was provided + emitRetryEvent := func(kind EventKind, payload any) { + if len(scope) > 0 { + al.emitEvent(kind, + EventMeta{AgentID: agent.ID, TurnID: scope[0].turnID, SessionKey: opts.SessionKey, Iteration: iteration}, + payload) + } + } + if isTimeoutError && retry < maxRetries { backoff := time.Duration(retry+1) * 5 * time.Second logger.WarnCF("agent", "Timeout error, retrying after backoff", map[string]any{ @@ -1604,6 +1519,10 @@ func (al *AgentLoop) callLLMWithRetry( "retry": retry, "backoff": backoff.String(), }) + emitRetryEvent(EventKindLLMRetry, LLMRetryPayload{ + Attempt: retry + 1, MaxRetries: maxRetries, + Reason: "timeout", Error: err.Error(), Backoff: backoff, + }) time.Sleep(backoff) continue } @@ -1613,6 +1532,10 @@ func (al *AgentLoop) callLLMWithRetry( "error": err.Error(), "retry": retry, }) + emitRetryEvent(EventKindLLMRetry, LLMRetryPayload{ + Attempt: retry + 1, MaxRetries: maxRetries, + Reason: "context_limit", Error: err.Error(), + }) if retry == 0 && !constants.IsInternalChannel(opts.Channel) { _ = al.bus.PublishOutbound(ctx, bus.OutboundMessage{ Channel: opts.Channel, @@ -1620,6 +1543,7 @@ func (al *AgentLoop) callLLMWithRetry( Content: "Context window exceeded. Compressing history and retrying...", }) } + prevCount := len(*messages) al.forceCompression(agent, opts.SessionKey) newHistory := agent.Sessions.GetHistory(opts.SessionKey) newSummary := agent.Sessions.GetSummary(opts.SessionKey) @@ -1628,6 +1552,11 @@ func (al *AgentLoop) callLLMWithRetry( nil, opts.Channel, opts.ChatID, opts.SenderID, opts.SenderDisplayName, ) + emitRetryEvent(EventKindContextCompress, ContextCompressPayload{ + Reason: ContextCompressReasonRetry, + DroppedMessages: prevCount - len(*messages), + RemainingMessages: len(*messages), + }) continue } break diff --git a/pkg/agent/loop_commands.go b/pkg/agent/loop_commands.go index 367697487..151d11d50 100644 --- a/pkg/agent/loop_commands.go +++ b/pkg/agent/loop_commands.go @@ -46,6 +46,20 @@ func (al *AgentLoop) buildCommandsRuntime(agent *AgentInstance, sessionKey strin if agent == nil { return "", fmt.Errorf("no default agent configured") } + // Validate model exists in model_list + cfg := al.GetConfig() + if cfg != nil && len(cfg.ModelList) > 0 { + found := false + for _, m := range cfg.ModelList { + if m.ModelName == value { + found = true + break + } + } + if !found { + return "", fmt.Errorf("model %q not found in model_list or providers", value) + } + } old := agent.Model agent.Model = value return old, nil @@ -123,6 +137,26 @@ func (al *AgentLoop) handleCommand( args := parts[1:] switch cmd { + case "/use": + if len(args) == 0 { + return "Usage: /use [message]", true + } + skillName := args[0] + // If we reached here, expandForkCommands didn't expand (either skill not found or no message) + if agent != nil { + if _, found := agent.ContextBuilder.LoadSkill(skillName); found { + // Skill exists but no message → arm for next message. + // Use channel:chatID as key since session key may not be resolved yet. + armKey := msg.Channel + ":" + msg.ChatID + if sessionKey != "" { + armKey = sessionKey + } + al.pendingSkills.Store(armKey, skillName) + return fmt.Sprintf("Skill %q is armed for your next message.", skillName), true + } + } + return fmt.Sprintf("Unknown skill: %s", skillName), true + case "/session": return al.handleSessionCommand(args, msg.SessionKey), true @@ -624,29 +658,21 @@ func (al *AgentLoop) expandSkillCommand(msg bus.InboundMessage) (expanded string return "", "", false } - skillContent, found := agent.ContextBuilder.LoadSkill(skillName) + _, found := agent.ContextBuilder.LoadSkill(skillName) if !found { return "", "", false } + // If no user message, don't expand — let it fall through to /use arming handler + if userMessage == "" { + return "", "", false + } + tag := fmt.Sprintf("[Skill: %s]", skillName) - // Build expanded message: skill instructions + user message (for LLM) - - var sb strings.Builder - - sb.WriteString(tag) - - sb.WriteString("\n\n") - - sb.WriteString(skillContent) - - if userMessage != "" { - sb.WriteString("\n\n---\n\n") - - sb.WriteString(userMessage) - } + // The skill content will be injected into the system prompt via ForcedSkills/BuildMessages. + expanded = userMessage // Build compact form: skill name tag + user message only (for history) @@ -656,7 +682,7 @@ func (al *AgentLoop) expandSkillCommand(msg bus.InboundMessage) (expanded string compactForm = tag + "\n" + userMessage } - return sb.String(), compactForm, true + return expanded, compactForm, true } // handleSkillsCommand lists all available skills. diff --git a/pkg/agent/loop_ext.go b/pkg/agent/loop_ext.go index 24ad15fba..ea6fd2cc8 100644 --- a/pkg/agent/loop_ext.go +++ b/pkg/agent/loop_ext.go @@ -39,12 +39,8 @@ type loopExt struct { activeTasks sync.Map // sessionKey → *activeTask - activeRequests sync.WaitGroup // tracks in-flight LLM worker requests - done chan struct{} // closed by Close() to stop background goroutines - reloadFunc func() error // upstream compat: called by buildCommandsRuntime - saveConfig func(*config.Config) error onHeartbeatThreadUpdate func(int) @@ -269,20 +265,35 @@ func (al *AgentLoop) handleTaskIntervention(msg bus.InboundMessage) (string, boo return "Intervention sent to running task.", true } -// expandForkCommands expands fork-specific /skill and /plan commands in the message. -// Returns the modified message and the compact form for history. -func (al *AgentLoop) expandForkCommands(msg *bus.InboundMessage) string { +// expandForkCommands expands fork-specific /skill, /use, and /plan commands in the message. +// Returns the compact form for history and any forced skill names. +func (al *AgentLoop) expandForkCommands(msg *bus.InboundMessage) (compact string, forcedSkills []string) { var expansionCompact string - if expanded, compact, ok := al.expandSkillCommand(*msg); ok { + // Support both /skill and /use prefixes for skill loading + content := strings.TrimSpace(msg.Content) + originalContent := content + if strings.HasPrefix(content, "/use ") { + // Rewrite /use → /skill for the skill expander + msg.Content = "/skill " + content[5:] + } + if expanded, cpt, ok := al.expandSkillCommand(*msg); ok { msg.Content = expanded - expansionCompact = compact + expansionCompact = cpt + // Extract skill name + skillFields := strings.Fields(originalContent) + if len(skillFields) >= 2 { + forcedSkills = append(forcedSkills, skillFields[1]) + } + } else if strings.HasPrefix(originalContent, "/use ") { + // Skill expansion failed (no message or not found) — restore original /use content + msg.Content = originalContent } - if expanded, compact, ok := al.expandPlanCommand(*msg); ok { + if expanded, cpt, ok := al.expandPlanCommand(*msg); ok { msg.Content = expanded - expansionCompact = compact + expansionCompact = cpt } - return expansionCompact + return expansionCompact, forcedSkills } diff --git a/pkg/agent/loop_hooks.go b/pkg/agent/loop_hooks.go index 629b3d604..13528b39f 100644 --- a/pkg/agent/loop_hooks.go +++ b/pkg/agent/loop_hooks.go @@ -323,6 +323,31 @@ func (al *AgentLoop) buildAsyncCallback(opts processOptions, toolName string) to ChatID: fmt.Sprintf("%s:%s", opts.Channel, opts.ChatID), Content: fmt.Sprintf("Async tool '%s' completed.\n\nResult:\n%s", toolName, content), }) + + meta := EventMeta{SessionKey: opts.SessionKey} + // Try to find the turn ID from the active task + taskKey := opts.SessionKey + if opts.TaskID != "" { + taskKey = opts.TaskID + } + if val, ok := al.activeTasks.Load(taskKey); ok { + if at, ok := val.(*activeTask); ok { + meta.TurnID = at.turnID + } + } + if ts := al.getActiveTurnState(opts.SessionKey); ts != nil { + meta.AgentID = ts.agentID + if meta.TurnID == "" { + meta.TurnID = ts.turnID + } + } + al.emitEvent(EventKindFollowUpQueued, meta, + FollowUpQueuedPayload{ + SourceTool: toolName, + Channel: opts.Channel, + ChatID: opts.ChatID, + ContentLen: len(content), + }) } } diff --git a/pkg/agent/loop_run.go b/pkg/agent/loop_run.go index 269d7e73d..8e1ba1aa4 100644 --- a/pkg/agent/loop_run.go +++ b/pkg/agent/loop_run.go @@ -29,6 +29,34 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt defer al.releaseSessionLock(opts.SessionKey) + // Event scope for this turn + scope := al.newTurnEventScope(agent.ID, opts.SessionKey) + turnStart := time.Now() + + // Register a turnState so the interrupt API can find this turn + ts := &turnState{ + turnID: scope.turnID, + agentID: agent.ID, + sessionKey: opts.SessionKey, + channel: opts.Channel, + chatID: opts.ChatID, + userMessage: opts.UserMessage, + phase: TurnPhaseRunning, + startedAt: turnStart, + agent: agent, + } + al.registerActiveTurn(ts) + defer al.clearActiveTurn(ts) + + al.emitEvent(EventKindTurnStart, + EventMeta{AgentID: agent.ID, TurnID: scope.turnID, SessionKey: opts.SessionKey}, + TurnStartPayload{ + Channel: opts.Channel, + ChatID: opts.ChatID, + UserMessage: opts.UserMessage, + MediaCount: len(opts.Media), + }) + // Report session lifecycle to canvas. al.reporter().ReportSpawn(opts.SessionKey, opts.Channel, opts.UserMessage) @@ -51,6 +79,8 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt cancel: taskCancel, interrupt: make(chan string, 1), + + turnID: scope.turnID, } // Guarantee heartbeat worktree cleanup on ALL exit paths (error, panic, normal). @@ -241,6 +271,7 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt opts.ChatID, opts.SenderID, opts.SenderDisplayName, + opts.ForcedSkills..., ) // Resolve media:// refs: images→base64 data URLs, non-images→local paths in content @@ -380,7 +411,7 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt var err error - finalContent, iteration, err = al.runLLMIteration(ctx, agent, messages, opts, task, curPlanStatus) + finalContent, iteration, err = al.runLLMIteration(ctx, agent, messages, opts, task, curPlanStatus, scope) if err != nil { return "", err } @@ -592,6 +623,15 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt "final_length": len(finalContent), }) + al.emitEvent(EventKindTurnEnd, + EventMeta{AgentID: agent.ID, TurnID: scope.turnID, SessionKey: opts.SessionKey, Iteration: iteration}, + TurnEndPayload{ + Status: TurnEndStatusCompleted, + Iterations: iteration, + Duration: time.Since(turnStart), + FinalContentLen: len(finalContent), + }) + return finalContent, nil } @@ -800,6 +840,7 @@ func (al *AgentLoop) runLLMIteration( opts processOptions, task *activeTask, planSnapshot string, + scope turnEventScope, ) (string, int, error) { hooks := al.buildHooks(agent, opts, task, planSnapshot) @@ -810,6 +851,26 @@ func (al *AgentLoop) runLLMIteration( iteration := 0 var finalContent string + // Initial steering poll: inject queued steering messages before the first LLM call + if !opts.SkipInitialSteeringPoll { + steeringMsgs := al.dequeueSteeringMessagesForScopeWithFallback(opts.SessionKey) + if len(steeringMsgs) > 0 { + messages = append(messages, steeringMsgs...) + totalLen := 0 + for _, sm := range steeringMsgs { + totalLen += len(sm.Content) + } + al.emitEvent(EventKindSteeringInjected, + EventMeta{AgentID: agent.ID, TurnID: scope.turnID, SessionKey: opts.SessionKey}, + SteeringInjectedPayload{Count: len(steeringMsgs), TotalContentLen: totalLen}) + } + } + + // Also inject any initial steering messages from the process options + if len(opts.InitialSteeringMessages) > 0 { + messages = append(messages, opts.InitialSteeringMessages...) + } + for iteration < agent.MaxIterations { iteration++ @@ -855,6 +916,40 @@ func (al *AgentLoop) runLLMIteration( "tools_json": formatToolsForLog(providerToolDefs), }) + // Hook: BeforeLLM — let hooks inspect/modify the request + hookMeta := EventMeta{AgentID: agent.ID, TurnID: scope.turnID, SessionKey: opts.SessionKey, Iteration: iteration} + if al.hooks != nil { + hookReq := &LLMHookRequest{ + Meta: hookMeta, + Model: activeModel, + Tools: providerToolDefs, + Channel: opts.Channel, + ChatID: opts.ChatID, + } + modified, _ := al.hooks.BeforeLLM(ctx, hookReq) + if modified != nil { + if modified.Model != "" && modified.Model != activeModel { + activeModel = modified.Model + // Update candidates to use the hook-specified model + candidates = []providers.FallbackCandidate{{Model: modified.Model}} + } + if len(modified.Tools) > 0 { + providerToolDefs = modified.Tools + } + } + } + + // Emit LLM request event + al.emitEvent(EventKindLLMRequest, + hookMeta, + LLMRequestPayload{ + Model: activeModel, + MessagesCount: len(messages), + ToolsCount: len(providerToolDefs), + MaxTokens: agent.MaxTokens, + Temperature: agent.Temperature, + }) + // Streaming setup onChunk, streamCleanup := hooks.SetupStreaming() @@ -862,7 +957,7 @@ func (al *AgentLoop) runLLMIteration( // Call LLM with retry response, err := al.callLLMWithRetry(ctx, agent, &messages, opts, - providerToolDefs, candidates, activeModel, onChunk, iteration) + providerToolDefs, candidates, activeModel, onChunk, iteration, scope) // Streaming cleanup if streamCleanup != nil { @@ -890,6 +985,30 @@ func (al *AgentLoop) runLLMIteration( ) } + // Hook: AfterLLM — let hooks inspect/modify the response + if al.hooks != nil { + hookResp := &LLMHookResponse{ + Meta: hookMeta, + Model: activeModel, + Response: response, + Channel: opts.Channel, + ChatID: opts.ChatID, + } + modified, _ := al.hooks.AfterLLM(ctx, hookResp) + if modified != nil && modified.Response != nil { + response = modified.Response + } + } + + // Emit LLM response event + al.emitEvent(EventKindLLMResponse, + EventMeta{AgentID: agent.ID, TurnID: scope.turnID, SessionKey: opts.SessionKey, Iteration: iteration}, + LLMResponsePayload{ + ContentLen: len(response.Content), + ToolCalls: len(response.ToolCalls), + HasReasoning: response.Reasoning != "", + }) + go al.handleReasoning(ctx, response.Reasoning, opts.Channel, al.targetReasoningChannelID(opts.Channel)) logger.DebugCF("agent", "LLM response", @@ -966,7 +1085,7 @@ func (al *AgentLoop) runLLMIteration( agent.Sessions.AddFullMessage(opts.SessionKey, assistantMsg) // Execute tool calls and collect results - lastBlocker := al.executeToolCalls(ctx, agent, normalizedToolCalls, &messages, opts, hooks, iteration) + lastBlocker := al.executeToolCalls(ctx, agent, normalizedToolCalls, &messages, opts, hooks, iteration, scope) // Tick TTL-based tool expiry after execution agent.Tools.TickTTL() @@ -993,9 +1112,41 @@ func (al *AgentLoop) executeToolCalls( opts processOptions, hooks iterationHooks, iteration int, + scope turnEventScope, ) string { var lastBlocker string - for _, tc := range toolCalls { + steered := false + for i, tc := range toolCalls { + // Check for pending steering messages between tool calls + if i > 0 && !steered { + steeringMsgs := al.dequeueSteeringMessagesForScopeWithFallback(opts.SessionKey) + if len(steeringMsgs) > 0 { + steered = true + totalLen := 0 + for _, sm := range steeringMsgs { + totalLen += len(sm.Content) + } + *messages = append(*messages, steeringMsgs...) + al.emitEvent(EventKindSteeringInjected, + EventMeta{AgentID: agent.ID, TurnID: scope.turnID, SessionKey: opts.SessionKey, Iteration: iteration}, + SteeringInjectedPayload{Count: len(steeringMsgs), TotalContentLen: totalLen}) + } + } + + // Skip remaining tools if steering was injected + if steered { + al.emitEvent(EventKindToolExecSkipped, + EventMeta{AgentID: agent.ID, TurnID: scope.turnID, SessionKey: opts.SessionKey, Iteration: iteration}, + ToolExecSkippedPayload{Tool: tc.Name, Reason: "steering"}) + // Still need to add a tool result to messages for protocol correctness + *messages = append(*messages, providers.Message{ + Role: "tool", + Content: "Skipped due to queued user message.", + ToolCallID: tc.ID, + }) + continue + } + argsJSON, _ := json.Marshal(tc.Arguments) argsPreview := utils.Truncate(string(argsJSON), 200) logger.InfoCF("agent", fmt.Sprintf("Tool call: %s(%s)", tc.Name, argsPreview), @@ -1015,6 +1166,59 @@ func (al *AgentLoop) executeToolCalls( } } + toolMeta := EventMeta{AgentID: agent.ID, TurnID: scope.turnID, SessionKey: opts.SessionKey, Iteration: iteration} + + // Hook: BeforeTool — let hooks inspect/modify the tool call + toolArgs := tc.Arguments + if al.hooks != nil { + hookReq := &ToolCallHookRequest{ + Meta: toolMeta, Tool: tc.Name, Arguments: toolArgs, + Channel: opts.Channel, ChatID: opts.ChatID, + } + modified, decision := al.hooks.BeforeTool(ctx, hookReq) + switch decision.normalizedAction() { + case HookActionContinue, HookActionModify: + if modified != nil { + toolArgs = modified.Arguments + } + case HookActionDenyTool: + reason := decision.Reason + if reason == "" { + reason = "denied by hook" + } + al.emitEvent(EventKindToolExecSkipped, toolMeta, + ToolExecSkippedPayload{Tool: tc.Name, Reason: reason}) + *messages = append(*messages, providers.Message{ + Role: "tool", ToolCallID: tc.ID, + Content: fmt.Sprintf("Tool execution denied by hook: %s", reason), + }) + continue + } + + // Hook: ApproveTool + approvalReq := &ToolApprovalRequest{ + Meta: toolMeta, Tool: tc.Name, Arguments: toolArgs, + Channel: opts.Channel, ChatID: opts.ChatID, + } + approvalDec := al.hooks.ApproveTool(ctx, approvalReq) + if !approvalDec.Approved { + reason := approvalDec.Reason + if reason == "" { + reason = "blocked by approval hook" + } + denialMsg := fmt.Sprintf("Tool execution denied by approval hook: %s", reason) + al.emitEvent(EventKindToolExecSkipped, toolMeta, + ToolExecSkippedPayload{Tool: tc.Name, Reason: denialMsg}) + *messages = append(*messages, providers.Message{ + Role: "tool", ToolCallID: tc.ID, Content: denialMsg, + }) + continue + } + } + + al.emitEvent(EventKindToolExecStart, toolMeta, + ToolExecStartPayload{Tool: tc.Name, Arguments: toolArgs}) + asyncCallback := hooks.OnPreToolExec(ctx, tc) toolStart := time.Now() @@ -1025,11 +1229,34 @@ func (al *AgentLoop) executeToolCalls( } toolResult := agent.Tools.ExecuteWithContext( - toolCtx, tc.Name, tc.Arguments, + toolCtx, tc.Name, toolArgs, opts.Channel, opts.ChatID, asyncCallback, ) toolDuration := time.Since(toolStart) + // Hook: AfterTool — let hooks inspect/modify the tool result + if al.hooks != nil { + hookResult := &ToolResultHookResponse{ + Meta: toolMeta, Tool: tc.Name, Arguments: toolArgs, + Result: toolResult, Duration: toolDuration, + Channel: opts.Channel, ChatID: opts.ChatID, + } + modified, _ := al.hooks.AfterTool(ctx, hookResult) + if modified != nil && modified.Result != nil { + toolResult = modified.Result + } + } + + al.emitEvent(EventKindToolExecEnd, toolMeta, + ToolExecEndPayload{ + Tool: tc.Name, + Duration: toolDuration, + ForLLMLen: len(toolResult.ForLLM), + ForUserLen: len(toolResult.ForUser), + IsError: toolResult.IsError, + Async: toolResult.Async, + }) + hooks.OnToolExecDone(tc, toolResult, toolDuration) // Publish results to user diff --git a/pkg/agent/loop_session.go b/pkg/agent/loop_session.go index ffd9be78d..a61df3f37 100644 --- a/pkg/agent/loop_session.go +++ b/pkg/agent/loop_session.go @@ -314,6 +314,15 @@ func (al *AgentLoop) summarizeSession(agent *AgentInstance, sessionKey string) { agent.Sessions.Save(sessionKey) } + + al.emitEvent(EventKindSessionSummarize, + EventMeta{AgentID: agent.ID, SessionKey: sessionKey}, + SessionSummarizePayload{ + SummarizedMessages: len(validMessages), + KeptMessages: 4, + SummaryLen: len(finalSummary), + OmittedOversized: omitted, + }) } } diff --git a/pkg/agent/loop_task.go b/pkg/agent/loop_task.go index 2f081e6fc..97e460354 100644 --- a/pkg/agent/loop_task.go +++ b/pkg/agent/loop_task.go @@ -45,6 +45,8 @@ type activeTask struct { messageContent string // last content sent by the message tool (for inclusion in completion) + turnID string // event correlation ID for the owning turn + mu sync.Mutex } diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 780fab0cf..3932bb9fa 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -890,6 +890,11 @@ func TestProcessMessage_SwitchModelShowModelConsistency(t *testing.T) { Model: "openrouter/deepseek/deepseek-v3.2", APIBase: "https://openrouter.ai/api/v1", }, + { + ModelName: "after-switch", + Model: "openai/after-switch-model", + APIBase: "https://local.example.invalid/v1", + }, }, } cfg.WithSecurity(&config.SecurityConfig{ @@ -900,6 +905,9 @@ func TestProcessMessage_SwitchModelShowModelConsistency(t *testing.T) { "deepseek": { APIKeys: []string{"test-key"}, }, + "after-switch": { + APIKeys: []string{"test-key"}, + }, }, }) @@ -918,7 +926,7 @@ func TestProcessMessage_SwitchModelShowModelConsistency(t *testing.T) { ID: "user1", }, }) - if !strings.Contains(switchResp, "Switched model from before-switch to after-switch") { + if !strings.Contains(switchResp, "Switched model from local to after-switch") { t.Fatalf("unexpected /switch reply: %q", switchResp) } diff --git a/pkg/agent/loop_upstream_test.go b/pkg/agent/loop_upstream_test.go index a5a9ef09c..fdbf60fad 100644 --- a/pkg/agent/loop_upstream_test.go +++ b/pkg/agent/loop_upstream_test.go @@ -91,7 +91,7 @@ func TestSelectCandidates_WithRouterUsesLight(t *testing.T) { Agents: config.AgentsConfig{ Defaults: config.AgentDefaults{ Workspace: tmpDir, - ModelName: "heavy-model", + ModelName: "heavy-model", MaxTokens: 4096, MaxToolIterations: 10, }, diff --git a/pkg/agent/steering.go b/pkg/agent/steering.go index ad6613e8c..a49ce5e80 100644 --- a/pkg/agent/steering.go +++ b/pkg/agent/steering.go @@ -279,13 +279,6 @@ func (al *AgentLoop) dequeueSteeringMessagesForScopeWithFallback(scope string) [ return al.steering.dequeueScopeWithFallback(scope) } -func (al *AgentLoop) pendingSteeringCountForScope(scope string) int { - if al.steering == nil { - return 0 - } - return al.steering.lenScope(scope) -} - func (al *AgentLoop) continueWithSteeringMessages( ctx context.Context, agent *AgentInstance, diff --git a/pkg/agent/steering_test.go b/pkg/agent/steering_test.go index 265ff402c..c781b6698 100644 --- a/pkg/agent/steering_test.go +++ b/pkg/agent/steering_test.go @@ -340,7 +340,6 @@ func TestAgentLoop_Continue_WithMessages(t *testing.T) { } } - // slowTool simulates a tool that takes some time to execute. type slowTool struct { name string diff --git a/pkg/agent/subturn.go b/pkg/agent/subturn.go index e09861e15..9191fc62d 100644 --- a/pkg/agent/subturn.go +++ b/pkg/agent/subturn.go @@ -605,15 +605,15 @@ func copyAgentInstance(src *AgentInstance) AgentInstance { ThinkingLevel: src.ThinkingLevel, ContextWindow: src.ContextWindow, SummarizeMessageThreshold: src.SummarizeMessageThreshold, SummarizeTokenPercent: src.SummarizeTokenPercent, - Provider: src.Provider, Sessions: src.Sessions, + Provider: src.Provider, Sessions: src.Sessions, ContextBuilder: src.ContextBuilder, Tools: src.Tools, Subagents: src.Subagents, SkillsFilter: src.SkillsFilter, Candidates: src.Candidates, - PlanModel: src.PlanModel, PlanFallbacks: src.PlanFallbacks, + PlanModel: src.PlanModel, PlanFallbacks: src.PlanFallbacks, PlanCandidates: src.PlanCandidates, - ImageModel: src.ImageModel, ImageFallbacks: src.ImageFallbacks, + ImageModel: src.ImageModel, ImageFallbacks: src.ImageFallbacks, ImageCandidates: src.ImageCandidates, - Router: src.Router, LightCandidates: src.LightCandidates, + Router: src.Router, LightCandidates: src.LightCandidates, } } diff --git a/pkg/agent/turn.go b/pkg/agent/turn.go index c2d3d0d4d..1d29528bf 100644 --- a/pkg/agent/turn.go +++ b/pkg/agent/turn.go @@ -2,12 +2,10 @@ package agent import ( "context" - "reflect" "sync" "sync/atomic" "time" - "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/session" "github.com/sipeed/picoclaw/pkg/tools" @@ -41,8 +39,6 @@ type ActiveTurnInfo struct { type turnResult struct { finalContent string - status TurnEndStatus - followUps []bus.InboundMessage } type turnState struct { @@ -61,24 +57,16 @@ type turnState struct { userMessage string media []string - phase TurnPhase - iteration int - startedAt time.Time - finalContent string - - followUps []bus.InboundMessage + phase TurnPhase + iteration int + startedAt time.Time gracefulInterrupt bool gracefulInterruptHint string - gracefulTerminalUsed bool hardAbort bool providerCancel context.CancelFunc turnCancel context.CancelFunc - restorePointHistory []providers.Message - restorePointSummary string - persistedMessages []providers.Message - // SubTurn support (from HEAD) depth int // SubTurn depth (0 for root turn) parentTurnID string // Parent turn ID (empty for root turn) @@ -86,7 +74,7 @@ type turnState struct { pendingResults chan *tools.ToolResult // Channel for SubTurn results concurrencySem chan struct{} // Semaphore for limiting concurrent SubTurns isFinished atomic.Bool // Whether this turn has finished - session session.LegacyStore // Session store reference + session session.LegacyStore // Session store reference initialHistoryLength int // Snapshot of history length at turn start // Additional SubTurn fields @@ -215,54 +203,6 @@ func (ts *turnState) snapshot() ActiveTurnInfo { } } -func (ts *turnState) setPhase(phase TurnPhase) { - ts.mu.Lock() - defer ts.mu.Unlock() - ts.phase = phase -} - -func (ts *turnState) setIteration(iteration int) { - ts.mu.Lock() - defer ts.mu.Unlock() - ts.iteration = iteration -} - -func (ts *turnState) currentIteration() int { - ts.mu.RLock() - defer ts.mu.RUnlock() - return ts.iteration -} - -func (ts *turnState) setFinalContent(content string) { - ts.mu.Lock() - defer ts.mu.Unlock() - ts.finalContent = content -} - -func (ts *turnState) finalContentLen() int { - ts.mu.RLock() - defer ts.mu.RUnlock() - return len(ts.finalContent) -} - -func (ts *turnState) setTurnCancel(cancel context.CancelFunc) { - ts.mu.Lock() - defer ts.mu.Unlock() - ts.turnCancel = cancel -} - -func (ts *turnState) setProviderCancel(cancel context.CancelFunc) { - ts.mu.Lock() - defer ts.mu.Unlock() - ts.providerCancel = cancel -} - -func (ts *turnState) clearProviderCancel(_ context.CancelFunc) { - ts.mu.Lock() - defer ts.mu.Unlock() - ts.providerCancel = nil -} - func (ts *turnState) requestGracefulInterrupt(hint string) bool { ts.mu.Lock() defer ts.mu.Unlock() @@ -274,18 +214,6 @@ func (ts *turnState) requestGracefulInterrupt(hint string) bool { return true } -func (ts *turnState) gracefulInterruptRequested() (bool, string) { - ts.mu.RLock() - defer ts.mu.RUnlock() - return ts.gracefulInterrupt && !ts.gracefulTerminalUsed, ts.gracefulInterruptHint -} - -func (ts *turnState) markGracefulTerminalUsed() { - ts.mu.Lock() - defer ts.mu.Unlock() - ts.gracefulTerminalUsed = true -} - func (ts *turnState) requestHardAbort() bool { ts.mu.Lock() if ts.hardAbort { @@ -306,12 +234,6 @@ func (ts *turnState) requestHardAbort() bool { return true } -func (ts *turnState) hardAbortRequested() bool { - ts.mu.RLock() - defer ts.mu.RUnlock() - return ts.hardAbort -} - func (ts *turnState) eventMeta(source, tracePath string) EventMeta { snap := ts.snapshot() return EventMeta{ @@ -324,67 +246,6 @@ func (ts *turnState) eventMeta(source, tracePath string) EventMeta { } } -func (ts *turnState) captureRestorePoint(history []providers.Message, summary string) { - ts.mu.Lock() - defer ts.mu.Unlock() - ts.restorePointHistory = append([]providers.Message(nil), history...) - ts.restorePointSummary = summary -} - -func (ts *turnState) recordPersistedMessage(msg providers.Message) { - ts.mu.Lock() - defer ts.mu.Unlock() - ts.persistedMessages = append(ts.persistedMessages, msg) -} - -func (ts *turnState) refreshRestorePointFromSession(agent *AgentInstance) { - history := agent.Sessions.GetHistory(ts.sessionKey) - summary := agent.Sessions.GetSummary(ts.sessionKey) - - ts.mu.RLock() - persisted := append([]providers.Message(nil), ts.persistedMessages...) - ts.mu.RUnlock() - - if matched := matchingTurnMessageTail(history, persisted); matched > 0 { - history = append([]providers.Message(nil), history[:len(history)-matched]...) - } - - ts.captureRestorePoint(history, summary) -} - -func (ts *turnState) restoreSession(agent *AgentInstance) error { - ts.mu.RLock() - history := append([]providers.Message(nil), ts.restorePointHistory...) - summary := ts.restorePointSummary - ts.mu.RUnlock() - - agent.Sessions.SetHistory(ts.sessionKey, history) - agent.Sessions.SetSummary(ts.sessionKey, summary) - return agent.Sessions.Save(ts.sessionKey) -} - -func matchingTurnMessageTail(history, persisted []providers.Message) int { - maxMatch := min(len(history), len(persisted)) - for size := maxMatch; size > 0; size-- { - if reflect.DeepEqual(history[len(history)-size:], persisted[len(persisted)-size:]) { - return size - } - } - return 0 -} - -func (ts *turnState) interruptHintMessage() providers.Message { - _, hint := ts.gracefulInterruptRequested() - content := "Interrupt requested. Stop scheduling tools and provide a short final summary." - if hint != "" { - content += "\n\nInterrupt hint: " + hint - } - return providers.Message{ - Role: "user", - Content: content, - } -} - // SubTurn-related methods // Finish marks the turn as finished and closes the pendingResults channel diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 0acf07d60..290361dc2 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -87,8 +87,7 @@ type Manager struct { placeholders sync.Map // "channel:chatID" → placeholderID (string) typingStops sync.Map // "channel:chatID" → func() reactionUndos sync.Map // "channel:chatID" → reactionEntry - streamActive sync.Map // "channel:chatID" → true (set when streamer.Finalize sent the message) - channelHashes map[string]string // channel name → config hash + streamActive sync.Map // "channel:chatID" → true (set when streamer.Finalize sent the message) } type asyncTask struct { diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 88c1b01d9..4d3cea84e 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -403,7 +403,6 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag return err } - // SendPlaceholder implements channels.PlaceholderCapable. // It sends a placeholder message (e.g. "Thinking... 💭") that will later be // edited to the actual response via EditMessage (channels.MessageEditor). diff --git a/pkg/channels/wecom/aibot.go b/pkg/channels/wecom/aibot.go index ab82d134f..2663104e4 100644 --- a/pkg/channels/wecom/aibot.go +++ b/pkg/channels/wecom/aibot.go @@ -708,7 +708,7 @@ func (c *WeComAIBotChannel) getStreamResponse(task *streamTask, timestamp, nonce default: if time.Now().After(task.Deadline) { // Deadline reached: close the stream with a notice, then wait for agent via response_url. - content = "⏳ Processing, please wait. The results will be sent shortly." + content = c.config.ProcessingMessage finish = true closeStreamOnly = true logger.InfoCF( diff --git a/pkg/config/config.go b/pkg/config/config.go index 078f1a679..ed95cf9e3 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -296,7 +296,6 @@ type SubTurnConfig struct { ConcurrencyTimeoutSec int `json:"concurrency_timeout_sec" env:"PICOCLAW_AGENTS_DEFAULTS_SUBTURN_CONCURRENCY_TIMEOUT_SEC"` } - type AgentDefaults struct { Workspace string `json:"workspace" env:"PICOCLAW_AGENTS_DEFAULTS_WORKSPACE"` RestrictToWorkspace bool `json:"restrict_to_workspace" env:"PICOCLAW_AGENTS_DEFAULTS_RESTRICT_TO_WORKSPACE"` @@ -1806,14 +1805,6 @@ func (c *Config) migrateChannelConfigs() { } func SaveConfig(path string, cfg *Config) error { - if cfg.security == nil { - logger.Errorf("config %#v", *cfg) - if len(cfg.ModelList) > 0 { - logger.Errorf("model[0] %#v", cfg.ModelList[0]) - } - logger.ErrorC("config", "security is nil") - return fmt.Errorf("security is nil") - } cfg.security = normalizeSecurityConfig(cfg.security) // Ensure version is always set when saving if cfg.Version == 0 { @@ -2056,7 +2047,6 @@ func (c *Config) FindModelConfigByRef(protocol, modelID string) *ModelConfig { return nil } -// HasProvidersConfig checks if any provider in the old providers config has configuration. // ValidateModelList validates all ModelConfig entries in the model_list. // It checks that each model config is valid. // Note: Multiple entries with the same model_name are allowed for load balancing. diff --git a/pkg/heartbeat/service.go b/pkg/heartbeat/service.go index 49b375d8f..86bd0ef35 100644 --- a/pkg/heartbeat/service.go +++ b/pkg/heartbeat/service.go @@ -7,7 +7,6 @@ package heartbeat import ( - "context" "fmt" "os" "path/filepath" @@ -490,43 +489,6 @@ func heartbeatHasUserTasks(content string) bool { return false } -// sendResponse sends the heartbeat response to the last channel -func (hs *HeartbeatService) sendResponse(response string) { - hs.mu.RLock() - msgBus := hs.bus - hs.mu.RUnlock() - - if msgBus == nil { - hs.logInfof("No message bus configured, heartbeat result not sent") - return - } - - // Get last channel from state - lastChannel := hs.state.GetLastChannel() - if lastChannel == "" { - hs.logInfof("No last channel recorded, heartbeat result not sent") - return - } - - platform, userID := hs.parseLastChannel(lastChannel) - - // Skip internal channels that can't receive messages - if platform == "" || userID == "" { - return - } - - pubCtx, pubCancel := context.WithTimeout(context.Background(), 5*time.Second) - defer pubCancel() - msgBus.PublishOutbound(pubCtx, bus.OutboundMessage{ - Channel: platform, - ChatID: userID, - Content: response, - }) - - hs.logInfof("Heartbeat result sent to %s", platform) -} - - // parseLastChannel parses the last channel string into platform and userID. // Returns empty strings for invalid or internal channels. func (hs *HeartbeatService) parseLastChannel(lastChannel string) (platform, userID string) { diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go index abec3d3d7..d6206c7a4 100644 --- a/pkg/logger/logger.go +++ b/pkg/logger/logger.go @@ -526,4 +526,3 @@ func Unsubscribe(sub *LogSubscriber) { logSubsMu.Unlock() close(sub.Ch) } - diff --git a/pkg/providers/openai_compat/provider.go b/pkg/providers/openai_compat/provider.go index 57429000f..09501d3a8 100644 --- a/pkg/providers/openai_compat/provider.go +++ b/pkg/providers/openai_compat/provider.go @@ -37,7 +37,7 @@ type Provider struct { maxTokensField string // Field name for max tokens (e.g., "max_completion_tokens" for o1/glm models) stream bool // Use SSE streaming internally (accumulates into a single LLMResponse) httpClient *http.Client - extraBody map[string]any // Additional fields to inject into request body + extraBody map[string]any // Additional fields to inject into request body // Rate limiting: minimum interval between consecutive API requests. // Shared across all goroutines using this provider instance. @@ -759,8 +759,6 @@ func cloneOpenAIToolArgs(src map[string]any) map[string]any { return dst } - - func normalizeModel(model, apiBase string) string { before, after, ok := strings.Cut(model, "/") if !ok { diff --git a/pkg/session/legacy_adapter.go b/pkg/session/legacy_adapter.go index a149edf85..b79fa71c3 100644 --- a/pkg/session/legacy_adapter.go +++ b/pkg/session/legacy_adapter.go @@ -225,11 +225,7 @@ func (la *LegacyAdapter) SetHistory(key string, history []providers.Message) { defer la.mu.Unlock() - c, ok := la.cache[key] - - if !ok { - return - } + c := la.getOrLoad(key) msgs := make([]providers.Message, len(history)) diff --git a/pkg/tools/send_file.go b/pkg/tools/send_file.go index 97a896468..3d7497bec 100644 --- a/pkg/tools/send_file.go +++ b/pkg/tools/send_file.go @@ -236,7 +236,7 @@ func downloadToTemp(ctx context.Context, rawURL string) (downloadResult, error) // Generate a random temp filename to avoid collisions var randBytes [8]byte - if _, err := rand.Read(randBytes[:]); err != nil { + if _, err = rand.Read(randBytes[:]); err != nil { return downloadResult{}, err } tmpPath := filepath.Join(dir, fmt.Sprintf("dl_%x%s", randBytes, ext)) @@ -319,21 +319,21 @@ func preferredExtension(mediaType string) string { // mime.ExtensionsByType returns multiple options in undefined order; // hardcode the most common ones for determinism. preferred := map[string]string{ - "image/png": ".png", - "image/jpeg": ".jpg", - "image/gif": ".gif", - "image/webp": ".webp", - "image/svg+xml": ".svg", - "image/bmp": ".bmp", - "image/tiff": ".tiff", - "video/mp4": ".mp4", - "video/webm": ".webm", - "audio/mpeg": ".mp3", - "audio/ogg": ".ogg", - "application/pdf": ".pdf", - "application/zip": ".zip", - "text/plain": ".txt", - "text/html": ".html", + "image/png": ".png", + "image/jpeg": ".jpg", + "image/gif": ".gif", + "image/webp": ".webp", + "image/svg+xml": ".svg", + "image/bmp": ".bmp", + "image/tiff": ".tiff", + "video/mp4": ".mp4", + "video/webm": ".webm", + "audio/mpeg": ".mp3", + "audio/ogg": ".ogg", + "application/pdf": ".pdf", + "application/zip": ".zip", + "text/plain": ".txt", + "text/html": ".html", "application/json": ".json", } if ext, ok := preferred[mediaType]; ok { diff --git a/pkg/tools/send_file_test.go b/pkg/tools/send_file_test.go index fd254e683..ac448762f 100644 --- a/pkg/tools/send_file_test.go +++ b/pkg/tools/send_file_test.go @@ -378,8 +378,14 @@ func TestFilenameForDownload(t *testing.T) { {"url no ext no content-type", "https://example.com/mcp/photos/42", "", "", "42"}, {"root path", "https://example.com/", "", "image/jpeg", "download.jpg"}, {"root no content-type", "https://example.com", "", "", "download"}, - {"content-disposition wins", "https://example.com/mcp/photos/42", `attachment; filename="photo.png"`, "image/jpeg", "photo.png"}, - {"content-disposition inline", "https://example.com/x", `inline; filename="report.pdf"`, "application/pdf", "report.pdf"}, + { + "content-disposition wins", "https://example.com/mcp/photos/42", + `attachment; filename="photo.png"`, "image/jpeg", "photo.png", + }, + { + "content-disposition inline", "https://example.com/x", + `inline; filename="report.pdf"`, "application/pdf", "report.pdf", + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { diff --git a/pkg/tools/spawn_test.go b/pkg/tools/spawn_test.go index e5ceb0047..8e1bf17fd 100644 --- a/pkg/tools/spawn_test.go +++ b/pkg/tools/spawn_test.go @@ -6,24 +6,6 @@ import ( "testing" ) -// mockSpawner implements SubTurnSpawner for testing -type mockSpawner struct{} - -func (m *mockSpawner) SpawnSubTurn(ctx context.Context, cfg SubTurnConfig) (*ToolResult, error) { - // Extract task from system prompt for response - task := cfg.SystemPrompt - if strings.Contains(task, "Task: ") { - parts := strings.Split(task, "Task: ") - if len(parts) > 1 { - task = parts[1] - } - } - return &ToolResult{ - ForLLM: "Task completed: " + task, - ForUser: "Task completed", - }, nil -} - func TestSpawnTool_Execute_EmptyTask(t *testing.T) { provider := &MockLLMProvider{} manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil, nil, WebSearchToolOptions{}) diff --git a/pkg/tools/subagent_tool_test.go b/pkg/tools/subagent_tool_test.go index c2aad2612..be48d3852 100644 --- a/pkg/tools/subagent_tool_test.go +++ b/pkg/tools/subagent_tool_test.go @@ -145,8 +145,14 @@ func TestSubagentTool_Execute_Success(t *testing.T) { provider := &MockLLMProvider{} manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil, nil, WebSearchToolOptions{}) tool := NewSubagentTool(manager) - manager.SetSpawner(func(ctx context.Context, task, label, agentID string, tools *ToolRegistry, maxTokens int, temperature float64, hasMaxTokens, hasTemperature bool) (*ToolResult, error) { - return &ToolResult{ForLLM: "Completed: " + task, ForUser: "Completed: " + task}, nil + manager.SetSpawner(func( + ctx context.Context, task, label, agentID string, + tools *ToolRegistry, maxTokens int, temperature float64, + hasMaxTokens, hasTemperature bool, + ) (*ToolResult, error) { + return &ToolResult{ + ForLLM: "Completed: " + task, ForUser: "Completed: " + task, + }, nil }) ctx := WithToolContext(context.Background(), "telegram", "chat-123") @@ -202,8 +208,14 @@ func TestSubagentTool_Execute_NoLabel(t *testing.T) { provider := &MockLLMProvider{} manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil, nil, WebSearchToolOptions{}) tool := NewSubagentTool(manager) - manager.SetSpawner(func(ctx context.Context, task, label, agentID string, tools *ToolRegistry, maxTokens int, temperature float64, hasMaxTokens, hasTemperature bool) (*ToolResult, error) { - return &ToolResult{ForLLM: "Completed: " + task, ForUser: "Completed: " + task}, nil + manager.SetSpawner(func( + ctx context.Context, task, label, agentID string, + tools *ToolRegistry, maxTokens int, temperature float64, + hasMaxTokens, hasTemperature bool, + ) (*ToolResult, error) { + return &ToolResult{ + ForLLM: "Completed: " + task, ForUser: "Completed: " + task, + }, nil }) ctx := context.Background() @@ -278,8 +290,14 @@ func TestSubagentTool_Execute_ContextPassing(t *testing.T) { provider := &MockLLMProvider{} manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil, nil, WebSearchToolOptions{}) tool := NewSubagentTool(manager) - manager.SetSpawner(func(ctx context.Context, task, label, agentID string, tools *ToolRegistry, maxTokens int, temperature float64, hasMaxTokens, hasTemperature bool) (*ToolResult, error) { - return &ToolResult{ForLLM: "Completed: " + task, ForUser: "Completed: " + task}, nil + manager.SetSpawner(func( + ctx context.Context, task, label, agentID string, + tools *ToolRegistry, maxTokens int, temperature float64, + hasMaxTokens, hasTemperature bool, + ) (*ToolResult, error) { + return &ToolResult{ + ForLLM: "Completed: " + task, ForUser: "Completed: " + task, + }, nil }) channel := "test-channel" @@ -306,8 +324,14 @@ func TestSubagentTool_ForUserTruncation(t *testing.T) { provider := &MockLLMProvider{} manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil, nil, WebSearchToolOptions{}) tool := NewSubagentTool(manager) - manager.SetSpawner(func(ctx context.Context, task, label, agentID string, tools *ToolRegistry, maxTokens int, temperature float64, hasMaxTokens, hasTemperature bool) (*ToolResult, error) { - return &ToolResult{ForLLM: "Completed: " + task, ForUser: "Completed: " + task}, nil + manager.SetSpawner(func( + ctx context.Context, task, label, agentID string, + tools *ToolRegistry, maxTokens int, temperature float64, + hasMaxTokens, hasTemperature bool, + ) (*ToolResult, error) { + return &ToolResult{ + ForLLM: "Completed: " + task, ForUser: "Completed: " + task, + }, nil }) ctx := context.Background()