mirror of
https://github.com/sipeed/picoclaw.git
synced 2026-06-12 18:08:54 +00:00
sky5454
12d5421c26
Break up the monolithic 4384-line loop.go into 12 focused files: - loop.go: core AgentLoop struct and main Run loop - loop_turn.go: turn execution logic (runTurn, askSideQuestion, etc.) - loop_utils.go: pure utility functions (formatters, helpers) - loop_init.go: constructor and tool registration - loop_message.go: message handling (processMessage, routing) - loop_command.go: command processing (/use, /btw, etc.) - loop_mcp.go: MCP runtime management - loop_event.go: event/hook system helpers - loop_media.go: media resolution and artifact handling - loop_outbound.go: response publishing - loop_transcribe.go: audio transcription - loop_steering.go: steering queue and continuation - loop_inject.go: setter injection methods No functional changes - pure code movement with updated imports. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
97 lines
2.6 KiB
Go
97 lines
2.6 KiB
Go
// PicoClaw - Ultra-lightweight personal AI agent
|
|
|
|
package agent
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/sipeed/picoclaw/pkg/bus"
|
|
"github.com/sipeed/picoclaw/pkg/logger"
|
|
)
|
|
|
|
func (al *AgentLoop) processMessageSync(ctx context.Context, msg bus.InboundMessage) {
|
|
if al.channelManager != nil {
|
|
defer al.channelManager.InvokeTypingStop(msg.Channel, msg.ChatID)
|
|
}
|
|
|
|
response, err := al.processMessage(ctx, msg)
|
|
al.publishResponseOrError(ctx, msg.Channel, msg.ChatID, msg.SessionKey, response, err)
|
|
}
|
|
|
|
func (al *AgentLoop) runTurnWithSteering(ctx context.Context, initialMsg bus.InboundMessage) {
|
|
// Process the initial message
|
|
response, err := al.processMessage(ctx, initialMsg)
|
|
if err != nil {
|
|
if !al.maybePublishError(ctx, initialMsg.Channel, initialMsg.ChatID, initialMsg.SessionKey, err) {
|
|
return // context canceled
|
|
}
|
|
response = ""
|
|
}
|
|
finalResponse := response
|
|
|
|
// Build continuation target
|
|
target, targetErr := al.buildContinuationTarget(initialMsg)
|
|
if targetErr != nil {
|
|
logger.WarnCF("agent", "Failed to build steering continuation target",
|
|
map[string]any{
|
|
"channel": initialMsg.Channel,
|
|
"error": targetErr.Error(),
|
|
})
|
|
return
|
|
}
|
|
if target == nil {
|
|
// System message or non-routable, response already published
|
|
return
|
|
}
|
|
|
|
// Drain steering queue using existing Continue mechanism
|
|
for al.pendingSteeringCountForScope(target.SessionKey) > 0 {
|
|
// Check for context cancellation between iterations
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
|
|
logger.InfoCF("agent", "Continuing queued steering after turn end",
|
|
map[string]any{
|
|
"channel": target.Channel,
|
|
"chat_id": target.ChatID,
|
|
"session_key": target.SessionKey,
|
|
"queue_depth": al.pendingSteeringCountForScope(target.SessionKey),
|
|
})
|
|
|
|
continued, continueErr := al.Continue(ctx, target.SessionKey, target.Channel, target.ChatID)
|
|
if continueErr != nil {
|
|
logger.WarnCF("agent", "Failed to continue queued steering",
|
|
map[string]any{
|
|
"channel": target.Channel,
|
|
"chat_id": target.ChatID,
|
|
"error": continueErr.Error(),
|
|
})
|
|
break
|
|
}
|
|
if continued == "" {
|
|
break
|
|
}
|
|
finalResponse = continued
|
|
}
|
|
|
|
// Publish final response
|
|
if finalResponse != "" {
|
|
al.PublishResponseIfNeeded(ctx, target.Channel, target.ChatID, target.SessionKey, finalResponse)
|
|
}
|
|
}
|
|
|
|
func (al *AgentLoop) resolveSteeringTarget(msg bus.InboundMessage) (string, string, bool) {
|
|
if msg.Channel == "system" {
|
|
return "", "", false
|
|
}
|
|
|
|
route, agent, err := al.resolveMessageRoute(msg)
|
|
if err != nil || agent == nil {
|
|
return "", "", false
|
|
}
|
|
allocation := al.allocateRouteSession(route, msg)
|
|
|
|
return resolveScopeKey(allocation.SessionKey, msg.SessionKey), agent.ID, true
|
|
}
|