fix: resolve lint errors and wire upstream event/hook system
- Remove unused code from merge (40 unused issues) - Fix gci import ordering (11 files) - Fix golines formatting (3 files) - Fix govet shadow in send_file.go - Fix godoclint in config.go - Wire upstream event emissions into fork's loop - Wire upstream hook system (BeforeLLM/AfterLLM/BeforeTool/AfterTool/ApproveTool) - Fix /use command and skill arming - Fix server auth_config tests - Add node_modules/ to .gitignore Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
dj-oyu
parent
0a657fd640
commit
ad24cddd57
27 changed files with 461 additions and 410 deletions
3
.gitignore
vendored
3
.gitignore
vendored
|
|
@ -30,6 +30,9 @@ config/config.json
|
|||
coverage.txt
|
||||
coverage.html
|
||||
|
||||
# Dependencies
|
||||
node_modules/
|
||||
|
||||
# OS
|
||||
.DS_Store
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
)
|
||||
},
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -58,7 +58,7 @@ type AgentLoop struct {
|
|||
mcp mcpRuntime
|
||||
hookRuntime hookRuntime
|
||||
steering *steeringQueue
|
||||
pendingSkills sync.Map
|
||||
pendingSkills sync.Map // sessionKey → skillName (armed by /use <skill>)
|
||||
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 <skill> 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
|
||||
|
|
|
|||
|
|
@ -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 <skill> [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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
@ -64,21 +60,13 @@ type turnState struct {
|
|||
phase TurnPhase
|
||||
iteration int
|
||||
startedAt time.Time
|
||||
finalContent string
|
||||
|
||||
followUps []bus.InboundMessage
|
||||
|
||||
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)
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -88,7 +88,6 @@ type Manager struct {
|
|||
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
|
||||
}
|
||||
|
||||
type asyncTask struct {
|
||||
|
|
|
|||
|
|
@ -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).
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -526,4 +526,3 @@ func Unsubscribe(sub *LogSubscriber) {
|
|||
logSubsMu.Unlock()
|
||||
close(sub.Ch)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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{})
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue