fix(agent): harden routing and retry guards
This commit is contained in:
parent
60f5f70f2c
commit
f803e03751
1 changed files with 44 additions and 4 deletions
|
|
@ -404,6 +404,9 @@ func (al *AgentLoop) ProcessDirectWithChannel(
|
||||||
// Each heartbeat is independent and doesn't accumulate context.
|
// Each heartbeat is independent and doesn't accumulate context.
|
||||||
func (al *AgentLoop) ProcessHeartbeat(ctx context.Context, content, channel, chatID string) (string, error) {
|
func (al *AgentLoop) ProcessHeartbeat(ctx context.Context, content, channel, chatID string) (string, error) {
|
||||||
agent := al.registry.GetDefaultAgent()
|
agent := al.registry.GetDefaultAgent()
|
||||||
|
if agent == nil {
|
||||||
|
return "", fmt.Errorf("no default agent for heartbeat")
|
||||||
|
}
|
||||||
return al.runAgentLoop(ctx, agent, processOptions{
|
return al.runAgentLoop(ctx, agent, processOptions{
|
||||||
SessionKey: "heartbeat",
|
SessionKey: "heartbeat",
|
||||||
Channel: channel,
|
Channel: channel,
|
||||||
|
|
@ -468,6 +471,9 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
|
||||||
if !ok {
|
if !ok {
|
||||||
agent = al.registry.GetDefaultAgent()
|
agent = al.registry.GetDefaultAgent()
|
||||||
}
|
}
|
||||||
|
if agent == nil {
|
||||||
|
return "", fmt.Errorf("no agent available for route (agent_id=%s)", route.AgentID)
|
||||||
|
}
|
||||||
|
|
||||||
// Use routed session key, but honor pre-set agent-scoped keys (for ProcessDirect/cron)
|
// Use routed session key, but honor pre-set agent-scoped keys (for ProcessDirect/cron)
|
||||||
sessionKey := route.SessionKey
|
sessionKey := route.SessionKey
|
||||||
|
|
@ -534,6 +540,9 @@ func (al *AgentLoop) processSystemMessage(ctx context.Context, msg bus.InboundMe
|
||||||
|
|
||||||
// Use default agent for system messages
|
// Use default agent for system messages
|
||||||
agent := al.registry.GetDefaultAgent()
|
agent := al.registry.GetDefaultAgent()
|
||||||
|
if agent == nil {
|
||||||
|
return "", fmt.Errorf("no default agent for system message")
|
||||||
|
}
|
||||||
|
|
||||||
// Use the origin session for context
|
// Use the origin session for context
|
||||||
sessionKey := routing.BuildAgentMainSessionKey(agent.ID)
|
sessionKey := routing.BuildAgentMainSessionKey(agent.ID)
|
||||||
|
|
@ -790,10 +799,35 @@ func (al *AgentLoop) runLLMIteration(
|
||||||
}
|
}
|
||||||
|
|
||||||
errMsg := strings.ToLower(err.Error())
|
errMsg := strings.ToLower(err.Error())
|
||||||
isContextError := strings.Contains(errMsg, "token") ||
|
|
||||||
strings.Contains(errMsg, "context") ||
|
// Check if this is a network/HTTP timeout — not a context window error.
|
||||||
|
isTimeoutError := errors.Is(err, context.DeadlineExceeded) ||
|
||||||
|
strings.Contains(errMsg, "deadline exceeded") ||
|
||||||
|
strings.Contains(errMsg, "client.timeout") ||
|
||||||
|
strings.Contains(errMsg, "timed out") ||
|
||||||
|
strings.Contains(errMsg, "timeout exceeded")
|
||||||
|
|
||||||
|
// Detect real context window / token limit errors, excluding network timeouts.
|
||||||
|
isContextError := !isTimeoutError && (strings.Contains(errMsg, "context_length_exceeded") ||
|
||||||
|
strings.Contains(errMsg, "context window") ||
|
||||||
|
strings.Contains(errMsg, "maximum context length") ||
|
||||||
|
strings.Contains(errMsg, "token limit") ||
|
||||||
|
strings.Contains(errMsg, "too many tokens") ||
|
||||||
|
strings.Contains(errMsg, "max_tokens") ||
|
||||||
strings.Contains(errMsg, "invalidparameter") ||
|
strings.Contains(errMsg, "invalidparameter") ||
|
||||||
strings.Contains(errMsg, "length")
|
strings.Contains(errMsg, "prompt is too long") ||
|
||||||
|
strings.Contains(errMsg, "request too large"))
|
||||||
|
|
||||||
|
if isTimeoutError && retry < maxRetries {
|
||||||
|
backoff := time.Duration(retry+1) * 5 * time.Second
|
||||||
|
logger.WarnCF("agent", "Timeout error, retrying after backoff", map[string]any{
|
||||||
|
"error": err.Error(),
|
||||||
|
"retry": retry,
|
||||||
|
"backoff": backoff.String(),
|
||||||
|
})
|
||||||
|
time.Sleep(backoff)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
if isContextError && retry < maxRetries {
|
if isContextError && retry < maxRetries {
|
||||||
logger.WarnCF("agent", "Context window error detected, attempting compression", map[string]any{
|
logger.WarnCF("agent", "Context window error detected, attempting compression", map[string]any{
|
||||||
|
|
@ -1493,9 +1527,15 @@ func (al *AgentLoop) handleCommand(ctx context.Context, msg bus.InboundMessage)
|
||||||
func extractPeer(msg bus.InboundMessage) *routing.RoutePeer {
|
func extractPeer(msg bus.InboundMessage) *routing.RoutePeer {
|
||||||
peerKind := msg.Metadata["peer_kind"]
|
peerKind := msg.Metadata["peer_kind"]
|
||||||
if peerKind == "" {
|
if peerKind == "" {
|
||||||
return nil
|
peerKind = msg.Peer.Kind
|
||||||
}
|
}
|
||||||
peerID := msg.Metadata["peer_id"]
|
peerID := msg.Metadata["peer_id"]
|
||||||
|
if peerID == "" {
|
||||||
|
peerID = msg.Peer.ID
|
||||||
|
}
|
||||||
|
if peerKind == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
if peerID == "" {
|
if peerID == "" {
|
||||||
if peerKind == "direct" {
|
if peerKind == "direct" {
|
||||||
peerID = msg.SenderID
|
peerID = msg.SenderID
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue