From 14426327dd1ac6a7da065d2a5fb758c0bb488376 Mon Sep 17 00:00:00 2001 From: Max Date: Wed, 25 Mar 2026 22:00:58 +0800 Subject: [PATCH] chore(logging): enhance logging and context management in agent and sandbox components - Updated logging in various methods to include detailed traces for better monitoring of execution flow, including assistantID and chatID. - Improved context management by ensuring proper release of resources and tracking of execution durations in agent calls. - Added heartbeat logging in the Claude parser to monitor stream processing and prevent potential issues with long-running tasks. - Refined the .gitignore file to exclude additional markdown files in the sandbox directory. --- .gitignore | 1 + agent/assistant/handlers/stream.go | 2 -- agent/robot/executor/standard/agent.go | 41 ++++++++++++++++++++----- agent/robot/executor/standard/runner.go | 6 ++++ agent/sandbox/v2/claude/parse.go | 36 +++++++++++++++++++++- agent/sandbox/v2/claude/runner.go | 13 +++++++- agent/sandbox/v2/claude/session.go | 6 ++-- 7 files changed, 90 insertions(+), 15 deletions(-) diff --git a/.gitignore b/.gitignore index 8cbd4399..ba97bb83 100644 --- a/.gitignore +++ b/.gitignore @@ -81,3 +81,4 @@ agent/robot/ROBOT-WATCHER-IMPROVEMENT.md agent/robot/ROBOT-IM-INTEGRATION-IMPROVEMENT.md agent/robot/ROBOT-CACHE-IMPROVEMENT.md sandbox/v2/PID-KILL-UPGRADE.md +sandbox/v2/*.md diff --git a/agent/assistant/handlers/stream.go b/agent/assistant/handlers/stream.go index e62c252e..51ed071c 100644 --- a/agent/assistant/handlers/stream.go +++ b/agent/assistant/handlers/stream.go @@ -117,7 +117,6 @@ func (s *streamState) handleMessageStart(data []byte) int { startData.ThreadID = s.ctx.Stack.ID } - // Initialize message state with the correct message ID s.inGroup = true s.currentGroupID = messageID s.buffer = []byte{} @@ -381,7 +380,6 @@ func (s *streamState) handleMessageEnd(data []byte) int { return 0 } - // Calculate duration durationMs := time.Since(s.groupStartTime).Milliseconds() // Use the tracked message type (thinking, text, tool_call, etc.) diff --git a/agent/robot/executor/standard/agent.go b/agent/robot/executor/standard/agent.go index df84b757..25834364 100644 --- a/agent/robot/executor/standard/agent.go +++ b/agent/robot/executor/standard/agent.go @@ -2,8 +2,10 @@ package standard import ( "fmt" + "time" "github.com/yaoapp/gou/text" + kunlog "github.com/yaoapp/kun/log" "github.com/yaoapp/yao/agent/assistant" agentcontext "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/output/message" @@ -190,27 +192,31 @@ func (c *AgentCaller) Call(ctx *robottypes.Context, assistantID string, messages Connector: c.Connector, } - // Convert robot context to agent context agentCtx := c.buildAgentContext(ctx, assistantID) - defer agentCtx.Release() // IMPORTANT: Release agent context to prevent resource leaks + defer func() { + kunlog.Trace("[robot-agent] releasing context: assistantID=%s chatID=%s", assistantID, c.ChatID) + agentCtx.Release() + }() + + callStart := time.Now() + kunlog.Trace("[robot-agent] Call started: assistantID=%s chatID=%s", assistantID, c.ChatID) - // Call assistant with streaming response, err := ast.Stream(agentCtx, messages, opts) if err != nil { + kunlog.Trace("[robot-agent] Call failed: assistantID=%s elapsed=%v err=%v", assistantID, time.Since(callStart).Round(time.Second), err) return nil, fmt.Errorf("assistant call failed: %w", err) } - // Build result + kunlog.Trace("[robot-agent] Call completed: assistantID=%s elapsed=%v", assistantID, time.Since(callStart).Round(time.Second)) + result := &CallResult{ Response: response, } - // Extract Next hook data if response.Next != nil { result.Next = response.Next } - // Extract Content from Completion if response.Completion != nil { if content, ok := response.Completion.Content.(string); ok { result.Content = content @@ -295,13 +301,22 @@ func (c *AgentCaller) CallStream(ctx *robottypes.Context, assistantID string, me } agentCtx := c.buildAgentContext(ctx, assistantID) - defer agentCtx.Release() + defer func() { + kunlog.Trace("[robot-agent] releasing context (CallStream): assistantID=%s chatID=%s", assistantID, c.ChatID) + agentCtx.Release() + }() + + callStart := time.Now() + kunlog.Trace("[robot-agent] CallStream started: assistantID=%s chatID=%s", assistantID, c.ChatID) response, err := ast.Stream(agentCtx, messages, opts) if err != nil { + kunlog.Trace("[robot-agent] CallStream failed: assistantID=%s elapsed=%v err=%v", assistantID, time.Since(callStart).Round(time.Second), err) return nil, fmt.Errorf("assistant call failed: %w", err) } + kunlog.Trace("[robot-agent] CallStream completed: assistantID=%s elapsed=%v", assistantID, time.Since(callStart).Round(time.Second)) + result := &CallResult{Response: response} if response.Next != nil { result.Next = response.Next @@ -354,13 +369,22 @@ func (c *AgentCaller) CallStreamRaw(ctx *robottypes.Context, assistantID string, } agentCtx := c.buildAgentContext(ctx, assistantID) - defer agentCtx.Release() + defer func() { + kunlog.Trace("[robot-agent] releasing context (CallStreamRaw): assistantID=%s chatID=%s", assistantID, c.ChatID) + agentCtx.Release() + }() + + callStart := time.Now() + kunlog.Trace("[robot-agent] CallStreamRaw started: assistantID=%s chatID=%s", assistantID, c.ChatID) response, err := ast.Stream(agentCtx, messages, opts) if err != nil { + kunlog.Trace("[robot-agent] CallStreamRaw failed: assistantID=%s elapsed=%v err=%v", assistantID, time.Since(callStart).Round(time.Second), err) return nil, fmt.Errorf("assistant call failed: %w", err) } + kunlog.Trace("[robot-agent] CallStreamRaw completed: assistantID=%s elapsed=%v", assistantID, time.Since(callStart).Round(time.Second)) + result := &CallResult{Response: response} if response.Next != nil { result.Next = response.Next @@ -420,6 +444,7 @@ func (c *AgentCaller) buildAgentContext(ctx *robottypes.Context, assistantID str } agentCtx.Logger = agentcontext.Noop() + kunlog.Trace("[robot-agent] context built: assistantID=%s chatID=%s contextID=%s", assistantID, c.ChatID, agentCtx.ID) return agentCtx } diff --git a/agent/robot/executor/standard/runner.go b/agent/robot/executor/standard/runner.go index 7e7b639d..d78a3041 100644 --- a/agent/robot/executor/standard/runner.go +++ b/agent/robot/executor/standard/runner.go @@ -8,6 +8,7 @@ import ( "github.com/yaoapp/gou/mcp" "github.com/yaoapp/gou/process" + kunlog "github.com/yaoapp/kun/log" agentcontext "github.com/yaoapp/yao/agent/context" robottypes "github.com/yaoapp/yao/agent/robot/types" ) @@ -146,6 +147,9 @@ func (r *Runner) executeAssistantTask(task *robottypes.Task, taskCtx *RunnerCont input = "## Context\n\n" + taskCtx.SystemPrompt + "\n\n## Task\n\n" + input } + kunlog.Trace("[robot-runner] executeAssistantTask: task=%s assistant=%s promptLen=%d prevResults=%d", + task.ID, task.ExecutorID, len(input), len(taskCtx.PreviousResults)) + r.log.logTaskInput(task, input) result, err := caller.CallWithMessages(r.ctx, task.ExecutorID, input) @@ -338,5 +342,7 @@ func (r *Runner) FormatPreviousResultsAsContext(results []robottypes.TaskResult) sb.WriteString("\n") } + contextLen := sb.Len() + kunlog.Trace("[robot-runner] FormatPreviousResultsAsContext: results=%d totalLen=%d", len(results), contextLen) return sb.String() } diff --git a/agent/sandbox/v2/claude/parse.go b/agent/sandbox/v2/claude/parse.go index 5714a368..52b89a4b 100644 --- a/agent/sandbox/v2/claude/parse.go +++ b/agent/sandbox/v2/claude/parse.go @@ -68,11 +68,29 @@ func (p *streamParser) parse(ctx context.Context, stdout io.ReadCloser) error { scanner := bufio.NewScanner(stdout) scanner.Buffer(make([]byte, 0, 64*1024), 10*1024*1024) + startTime := time.Now() + lineCount := 0 + lastHeartbeat := time.Now() + lastEventType := "" + + log.Trace("[claude-parse] stream started") + for scanner.Scan() { line := scanner.Text() if line == "" { continue } + lineCount++ + + if time.Since(lastHeartbeat) > 30*time.Second { + builderLen := 0 + if p.curTool != nil { + builderLen = p.curTool.inputJSON.Len() + } + log.Trace("[claude-parse] heartbeat: lines=%d elapsed=%v lastEvent=%s toolBuilderLen=%d", + lineCount, time.Since(startTime).Round(time.Second), lastEventType, builderLen) + lastHeartbeat = time.Now() + } var msg map[string]any if err := json.Unmarshal([]byte(line), &msg); err != nil { @@ -85,6 +103,7 @@ func (p *streamParser) parse(ctx context.Context, stdout io.ReadCloser) error { } msgType, _ := msg["type"].(string) + lastEventType = msgType var stopped bool switch msgType { @@ -97,16 +116,22 @@ func (p *streamParser) parse(ctx context.Context, stdout io.ReadCloser) error { case "user": stopped = p.handleUser(msg) case "result": + log.Trace("[claude-parse] stream ended: lines=%d elapsed=%v completed=true", lineCount, time.Since(startTime).Round(time.Second)) return p.handleResult(msg) case "error": + log.Trace("[claude-parse] stream ended with error: lines=%d elapsed=%v", lineCount, time.Since(startTime).Round(time.Second)) return p.handleError(msg) } if stopped { + log.Trace("[claude-parse] stream stopped by handler: lines=%d elapsed=%v", lineCount, time.Since(startTime).Round(time.Second)) return nil } } + log.Trace("[claude-parse] stream ended: lines=%d elapsed=%v completed=%v scanErr=%v", + lineCount, time.Since(startTime).Round(time.Second), p.completed, scanner.Err()) + if err := scanner.Err(); err != nil { log.Trace("[claude-parse] scanner error: %v (ctx.Err=%v)", err, ctx.Err()) if ctx.Err() != nil { @@ -346,6 +371,10 @@ func (p *streamParser) onContentBlockDelta(event map[string]any) (stopped bool) return false } p.curTool.inputJSON.WriteString(partial) + builderLen := p.curTool.inputJSON.Len() + if builderLen > 0 && builderLen%100000 < len(partial) { + log.Trace("[claude-parse] WARN: tool %s inputJSON growing: %d bytes", p.curTool.name, builderLen) + } if p.handler != nil { return p.emitExecute(map[string]any{ "input_delta": p.curTool.inputJSON.String(), @@ -381,10 +410,15 @@ func (p *streamParser) handleAssistant(msg map[string]any) (stopped bool) { itemType, _ := ci["type"].(string) if itemType == "tool_use" && p.handler != nil { + toolID, _ := ci["id"].(string) + + if _, alreadyStreamed := p.toolNames[toolID]; alreadyStreamed && toolID != "" { + continue + } + p.closeTextMessage() toolName, _ := ci["name"].(string) - toolID, _ := ci["id"].(string) if toolID == "" { toolID = fmt.Sprintf("tool_%d_%d", p.toolIndex, time.Now().UnixNano()) } diff --git a/agent/sandbox/v2/claude/runner.go b/agent/sandbox/v2/claude/runner.go index 61cf6df2..98eb8175 100644 --- a/agent/sandbox/v2/claude/runner.go +++ b/agent/sandbox/v2/claude/runner.go @@ -111,15 +111,24 @@ func (r *ClaudeRunner) Stream(ctx context.Context, req *types.StreamRequest, han chatID := req.ChatID r.lastChatID = chatID + assistantID := "" + if req.Config != nil { + assistantID = req.Config.ID + } + + log.Trace("[claude-runner] Stream started: assistantID=%s chatID=%s promptLen=%d", assistantID, chatID, len(cmd.shell)) sess, err := startSession(ctx, computer, p, cmd, chatID, r.logger) if err != nil { return err } + streamStart := time.Now() completed, err := sess.runStream(handler) r.lastCompleted = completed - r.logger.Debug("Stream: runStream returned completed=%v err=%v", completed, err) + elapsed := time.Since(streamStart).Round(time.Second) + log.Trace("[claude-runner] Stream finished: assistantID=%s chatID=%s completed=%v elapsed=%v err=%v", assistantID, chatID, completed, elapsed, err) + r.logger.Debug("Stream: runStream returned completed=%v err=%v elapsed=%v", completed, err, elapsed) if completed { sess.shutdown() if chatID != "" { @@ -142,6 +151,8 @@ func (r *ClaudeRunner) Cleanup(ctx context.Context, computer infra.Computer) err return nil } + log.Trace("[claude-runner] Cleanup: chatID=%s lastCompleted=%v", r.lastChatID, r.lastCompleted) + if r.lastCompleted { if r.logger != nil { r.logger.Info("cleanup: stream completed normally, preserving child processes") diff --git a/agent/sandbox/v2/claude/session.go b/agent/sandbox/v2/claude/session.go index 377e081d..0ff85a77 100644 --- a/agent/sandbox/v2/claude/session.go +++ b/agent/sandbox/v2/claude/session.go @@ -31,8 +31,8 @@ func startSession(ctx context.Context, computer infra.Computer, p platform, cmd opts = append(opts, infra.WithStdin(cmd.stdin)) } - logger.Info("claude session starting: cmd=%s workDir=%s platform=%s stdinLen=%d", - cmd.shell, cmd.workDir, p.OS(), len(cmd.stdin)) + logger.Info("claude session starting: cmd=%s workDir=%s platform=%s stdinLen=%d chatID=%s", + cmd.shell, cmd.workDir, p.OS(), len(cmd.stdin), chatID) execStream, err := computer.Stream(ctx, cmd.shell, opts...) if err != nil { @@ -155,7 +155,7 @@ func (s *session) watchCancel() func() { // would actively terminate child processes (web servers, etc.). Those children // survive because they run in separate process groups/sessions. func (s *session) shutdown() { - s.logger.Info("shutting down completed claude exec session") + s.logger.Info("shutting down completed claude exec session: chatID=%s", s.chatID) killCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel()