diff --git a/agent/robot/api/trigger.go b/agent/robot/api/trigger.go index f74be8ac..9733b598 100644 --- a/agent/robot/api/trigger.go +++ b/agent/robot/api/trigger.go @@ -111,6 +111,7 @@ func triggerHuman(ctx *types.Context, mgr managerInterface, memberID string, req Messages: req.Messages, PlanTime: req.PlanAt, ExecutorMode: req.ExecutorMode, + Locale: req.Locale, } // Call manager's Intervene diff --git a/agent/robot/events/delivery.go b/agent/robot/events/delivery.go index 6afff685..2364a10f 100644 --- a/agent/robot/events/delivery.go +++ b/agent/robot/events/delivery.go @@ -13,6 +13,8 @@ import ( "strings" "time" + "path/filepath" + "github.com/yaoapp/gou/process" "github.com/yaoapp/gou/text" agentcontext "github.com/yaoapp/yao/agent/context" @@ -21,6 +23,7 @@ import ( eventtypes "github.com/yaoapp/yao/event/types" "github.com/yaoapp/yao/messenger" messengerTypes "github.com/yaoapp/yao/messenger/types" + "github.com/yaoapp/yao/workspace" ) // handleDelivery routes delivery content to configured channels (email, webhook, process). @@ -432,6 +435,15 @@ func convertAttachments(ctx context.Context, attachments []robottypes.DeliveryAt result := make([]messengerTypes.Attachment, 0, len(attachments)) for _, att := range attachments { + // Handle workspace:// URIs — read file content from workspace FS + if strings.HasPrefix(att.File, "workspace://") { + wsAtt := convertWorkspaceAttachment(ctx, att) + if wsAtt != nil { + result = append(result, *wsAtt) + } + continue + } + uploader, fileID, isWrapper := attachment.Parse(att.File) if !isWrapper { log.Warn("convertAttachments: skipping non-wrapper file value=%q title=%q", att.File, att.Title) @@ -481,6 +493,87 @@ func convertAttachments(ctx context.Context, attachments []robottypes.DeliveryAt return result } +// convertWorkspaceAttachment reads a file from workspace:// URI and returns a messenger attachment. +// URI format: workspace:/// +func convertWorkspaceAttachment(ctx context.Context, att robottypes.DeliveryAttachment) *messengerTypes.Attachment { + uri := att.File + // Strip "workspace://" prefix + rest := strings.TrimPrefix(uri, "workspace://") + slashIdx := strings.Index(rest, "/") + if slashIdx < 0 { + log.Warn("convertWorkspaceAttachment: invalid URI %q — no path after wsID", uri) + return nil + } + wsID := rest[:slashIdx] + filePath := rest[slashIdx+1:] + if wsID == "" || filePath == "" { + log.Warn("convertWorkspaceAttachment: empty wsID or path in URI %q", uri) + return nil + } + + wsm := workspace.M() + if wsm == nil { + log.Warn("convertWorkspaceAttachment: workspace manager not available for URI %q", uri) + return nil + } + + wsFS, err := wsm.FS(ctx, wsID) + if err != nil { + log.Warn("convertWorkspaceAttachment: cannot get FS for workspace %q: %v", wsID, err) + return nil + } + + content, err := wsFS.ReadFile(filePath) + if err != nil { + log.Warn("convertWorkspaceAttachment: failed to read %q from workspace %q: %v", filePath, wsID, err) + return nil + } + + filename := filepath.Base(filePath) + if att.Title != "" { + filename = att.Title + } + + contentType := mimeFromExtDelivery(filepath.Ext(filename)) + log.Info("convertWorkspaceAttachment: added workspace attachment filename=%q contentType=%q size=%d uri=%q", + filename, contentType, len(content), uri) + + return &messengerTypes.Attachment{ + Filename: filename, + ContentType: contentType, + Content: content, + } +} + +func mimeFromExtDelivery(ext string) string { + switch strings.ToLower(ext) { + case ".pdf": + return "application/pdf" + case ".html", ".htm": + return "text/html" + case ".png": + return "image/png" + case ".jpg", ".jpeg": + return "image/jpeg" + case ".gif": + return "image/gif" + case ".csv": + return "text/csv" + case ".json": + return "application/json" + case ".md": + return "text/markdown" + case ".txt": + return "text/plain" + case ".xlsx": + return "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + case ".pptx": + return "application/vnd.openxmlformats-officedocument.presentationml.presentation" + default: + return "application/octet-stream" + } +} + func attachmentManagerKeys() []string { keys := make([]string, 0, len(attachment.Managers)) for k := range attachment.Managers { diff --git a/agent/robot/events/message.go b/agent/robot/events/message.go index 57a21b78..468fb365 100644 --- a/agent/robot/events/message.go +++ b/agent/robot/events/message.go @@ -73,6 +73,7 @@ func callHostAgent(ctx context.Context, payload *MessagePayload) (*MessageResult authorized := &oauthtypes.AuthorizedInfo{ UserID: payload.Metadata.SenderID, + TeamID: record.TeamID, } chatID := fmt.Sprintf("%s:%s", payload.Metadata.Channel, payload.Metadata.ChatID) agentCtx := agentcontext.New(ctx, authorized, chatID) diff --git a/agent/robot/executor/standard/agent.go b/agent/robot/executor/standard/agent.go index 4d9ee84c..0c39eb20 100644 --- a/agent/robot/executor/standard/agent.go +++ b/agent/robot/executor/standard/agent.go @@ -53,6 +53,11 @@ type AgentCaller struct { // When non-empty, injected into agentCtx.Metadata["workspace_id"] for sandbox node resolution. Workspace string + // Mode is the agent execution mode (e.g., "task" for robot P3 execution). + // When non-empty, injected into agentCtx.Metadata["mode"] (exposed as $CTX.MODE + // in prompt templates) and into opts.Mode for framework-level buffer/chat recording. + Mode string + // log is an optional structured logger; when set, Call emits agent-call logs. log *execLogger } @@ -194,6 +199,7 @@ func (c *AgentCaller) Call(ctx *robottypes.Context, assistantID string, messages Search: c.SkipSearch, }, Connector: c.Connector, + Mode: c.Mode, } agentCtx := c.buildAgentContext(ctx, assistantID) @@ -228,7 +234,7 @@ func (c *AgentCaller) Call(ctx *robottypes.Context, assistantID string, messages } if c.log != nil { - c.log.logAgentCall(assistantID, result) + c.log.logAgentCall(assistantID, c.Connector, result) } return result, nil @@ -276,6 +282,7 @@ func (c *AgentCaller) CallStream(ctx *robottypes.Context, assistantID string, me Search: c.SkipSearch, }, Connector: c.Connector, + Mode: c.Mode, } // Hook OnMessage to intercept streaming chunks and forward to callback @@ -332,7 +339,7 @@ func (c *AgentCaller) CallStream(ctx *robottypes.Context, assistantID string, me } if c.log != nil { - c.log.logAgentCall(assistantID, result) + c.log.logAgentCall(assistantID, c.Connector, result) } return result, nil @@ -366,6 +373,7 @@ func (c *AgentCaller) CallStreamRaw(ctx *robottypes.Context, assistantID string, Search: c.SkipSearch, }, Connector: c.Connector, + Mode: c.Mode, } if onMessage != nil { @@ -400,7 +408,7 @@ func (c *AgentCaller) CallStreamRaw(ctx *robottypes.Context, assistantID string, } if c.log != nil { - c.log.logAgentCall(assistantID, result) + c.log.logAgentCall(assistantID, c.Connector, result) } return result, nil @@ -448,11 +456,16 @@ func (c *AgentCaller) buildAgentContext(ctx *robottypes.Context, assistantID str } agentCtx.Logger = agentcontext.Noop() - if c.Workspace != "" { + if c.Workspace != "" || c.Mode != "" { if agentCtx.Metadata == nil { agentCtx.Metadata = map[string]interface{}{} } - agentCtx.Metadata["workspace_id"] = c.Workspace + if c.Workspace != "" { + agentCtx.Metadata["workspace_id"] = c.Workspace + } + if c.Mode != "" { + agentCtx.Metadata["MODE"] = c.Mode + } } kunlog.Trace("[robot-agent] context built: assistantID=%s chatID=%s contextID=%s", assistantID, c.ChatID, agentCtx.ID) diff --git a/agent/robot/executor/standard/delivery.go b/agent/robot/executor/standard/delivery.go index 1e1f5483..f63c55d2 100644 --- a/agent/robot/executor/standard/delivery.go +++ b/agent/robot/executor/standard/delivery.go @@ -3,6 +3,7 @@ package standard import ( "encoding/json" "fmt" + "path" "strings" "time" @@ -12,6 +13,7 @@ import ( robotevents "github.com/yaoapp/yao/agent/robot/events" robottypes "github.com/yaoapp/yao/agent/robot/types" "github.com/yaoapp/yao/event" + "github.com/yaoapp/yao/workspace" ) // RunDelivery executes P4: Delivery phase @@ -36,14 +38,36 @@ func (e *Executor) RunDelivery(ctx *robottypes.Context, exec *robottypes.Executi } formatter := NewInputFormatter() - userContent := formatter.FormatDeliveryInput(exec, robot) + + // Try workspace-based delivery input (manifest summaries instead of raw output inline) + var userContent string + if robot.Workspace != "" { + wsm := workspace.M() + if wsm != nil { + wsFS, err := wsm.FS(ctx, robot.Workspace) + if err == nil { + execDir := path.Join("robots", robot.MemberID, exec.ID) + data, err := wsFS.ReadFile(path.Join(execDir, "manifest.json")) + if err == nil { + var manifest Manifest + if json.Unmarshal(data, &manifest) == nil { + userContent = formatter.FormatDeliveryInputWithManifest(exec, robot, &manifest, robot.Workspace, execDir, locale) + } + } + } + } + } + + // Fallback to legacy full-output inline + if userContent == "" { + userContent = formatter.FormatDeliveryInput(exec, robot) + } if userContent == "" { return fmt.Errorf("no content available for delivery generation") } caller := NewAgentCaller() - caller.Connector = robot.LanguageModel caller.Workspace = robot.Workspace result, err := caller.CallWithMessages(ctx, agentID, userContent) if err != nil { @@ -413,3 +437,103 @@ func (f *InputFormatter) FormatDeliveryInput(exec *robottypes.Execution, robot * return sb.String() } + +// FormatDeliveryInputWithManifest formats delivery input using workspace manifest summaries +// instead of inlining full task outputs. This drastically reduces token usage. +func (f *InputFormatter) FormatDeliveryInputWithManifest( + exec *robottypes.Execution, + robot *robottypes.Robot, + manifest *Manifest, + wsID string, + execDir string, + locale string, +) string { + if exec == nil || manifest == nil { + return "" + } + + var sb strings.Builder + + // Robot Identity (same as legacy) + if robot != nil && robot.Config != nil && robot.Config.Identity != nil { + sb.WriteString("## Robot Identity\n\n") + sb.WriteString(fmt.Sprintf("- **Role**: %s\n", robot.Config.Identity.Role)) + if len(robot.Config.Identity.Duties) > 0 { + sb.WriteString("- **Duties**: ") + sb.WriteString(strings.Join(robot.Config.Identity.Duties, ", ")) + sb.WriteString("\n") + } + sb.WriteString("\n") + } + + sb.WriteString("## Execution Context\n\n") + sb.WriteString(fmt.Sprintf("- **Trigger**: %s\n", exec.TriggerType)) + sb.WriteString(fmt.Sprintf("- **Status**: %s\n", exec.Status)) + if locale != "" { + sb.WriteString(fmt.Sprintf("- **Language**: %s\n", locale)) + } + sb.WriteString(fmt.Sprintf("- **Start Time**: %s\n", exec.StartTime.Format("2006-01-02 15:04:05"))) + if exec.EndTime != nil { + duration := exec.EndTime.Sub(exec.StartTime) + sb.WriteString(fmt.Sprintf("- **Duration**: %s\n", duration.String())) + } + sb.WriteString("\n") + + sb.WriteString("## Goals\n\n") + sb.WriteString(manifest.Goals) + sb.WriteString("\n\n") + + // Results from manifest summaries + sb.WriteString("## Results (P3)\n\n") + successCount := 0 + failCount := 0 + + for _, t := range manifest.Tasks { + if t.Status == "completed" { + successCount++ + sb.WriteString(fmt.Sprintf("### ✓ Task: %s\n\n", t.ID)) + } else if t.Status == "failed" { + failCount++ + sb.WriteString(fmt.Sprintf("### ✗ Task: %s\n\n", t.ID)) + } else { + continue + } + + sb.WriteString(fmt.Sprintf("- **Description**: %s\n", t.Description)) + sb.WriteString(fmt.Sprintf("- **Executor**: %s (%s)\n", t.Executor, t.ExecutorType)) + + if t.Summary != "" { + sb.WriteString(fmt.Sprintf("- **Summary**: %s\n", t.Summary)) + } + if len(t.KeyOutputs) > 0 { + sb.WriteString(fmt.Sprintf("- **Key Outputs**: %s\n", strings.Join(t.KeyOutputs, ", "))) + } + + // File references — use existing URI if present, otherwise build from execDir + if len(t.Files) > 0 { + sb.WriteString("- **Artifacts**:\n") + for _, file := range t.Files { + uri := file.URI + if uri == "" { + uri = fmt.Sprintf("workspace://%s/%s/%s/%s", wsID, execDir, t.ID, file.Name) + } + sb.WriteString(fmt.Sprintf(" - [%s](%s)", file.Name, uri)) + if file.Desc != "" { + sb.WriteString(fmt.Sprintf(" — %s", file.Desc)) + } + sb.WriteString("\n") + } + } + + // Full output file reference + outputURI := fmt.Sprintf("workspace://%s/%s/%s.output.md", wsID, execDir, t.ID) + sb.WriteString(fmt.Sprintf("- **Full Output**: [%s.output.md](%s)\n", t.ID, outputURI)) + + sb.WriteString("\n") + } + + sb.WriteString(fmt.Sprintf("### Summary\n\n- **Total Tasks**: %d\n- **Succeeded**: %d\n- **Failed**: %d\n\n", + successCount+failCount, successCount, failCount)) + + return sb.String() +} diff --git a/agent/robot/executor/standard/goals.go b/agent/robot/executor/standard/goals.go index cb1d3629..3fccdef6 100644 --- a/agent/robot/executor/standard/goals.go +++ b/agent/robot/executor/standard/goals.go @@ -84,7 +84,6 @@ func (e *Executor) RunGoals(ctx *robottypes.Context, exec *robottypes.Execution, // Call agent caller := NewAgentCaller() - caller.Connector = robot.LanguageModel caller.Workspace = robot.Workspace result, err := caller.CallWithMessages(ctx, agentID, userContent) if err != nil { diff --git a/agent/robot/executor/standard/host.go b/agent/robot/executor/standard/host.go index afb019f0..f937bab8 100644 --- a/agent/robot/executor/standard/host.go +++ b/agent/robot/executor/standard/host.go @@ -32,7 +32,6 @@ func (e *Executor) CallHostAgent(ctx *robottypes.Context, robot *robottypes.Robo kunlog.Info("calling Host Agent %s for scenario=%s chatID=%s", agentID, input.Scenario, chatID) caller := NewConversationCaller(chatID) - caller.Connector = robot.LanguageModel caller.Workspace = robot.Workspace result, err := caller.CallWithMessages(ctx, agentID, string(inputJSON)) if err != nil { diff --git a/agent/robot/executor/standard/input.go b/agent/robot/executor/standard/input.go index 0787c33f..51785b39 100644 --- a/agent/robot/executor/standard/input.go +++ b/agent/robot/executor/standard/input.go @@ -183,6 +183,7 @@ func (f *InputFormatter) FormatAvailableResourcesWithLocale(robot *robottypes.Ro capabilities := i18n.Translate(agentID, locale, ast.Capabilities).(string) sb.WriteString(fmt.Sprintf(" - **Capabilities**: %s\n", capabilities)) } + } sb.WriteString("\n") } diff --git a/agent/robot/executor/standard/inspiration.go b/agent/robot/executor/standard/inspiration.go index c31faada..f637a9e3 100644 --- a/agent/robot/executor/standard/inspiration.go +++ b/agent/robot/executor/standard/inspiration.go @@ -53,7 +53,6 @@ func (e *Executor) RunInspiration(ctx *robottypes.Context, exec *robottypes.Exec // Call agent caller := NewAgentCaller() - caller.Connector = robot.LanguageModel caller.Workspace = robot.Workspace result, err := caller.CallWithMessages(ctx, agentID, userContent) if err != nil { diff --git a/agent/robot/executor/standard/log.go b/agent/robot/executor/standard/log.go index 7fbdc13f..f17c9751 100644 --- a/agent/robot/executor/standard/log.go +++ b/agent/robot/executor/standard/log.go @@ -103,9 +103,9 @@ func (l *execLogger) devTaskOverview(tasks []robottypes.Task) { // P3: Task Input // --------------------------------------------------------------------------- -func (l *execLogger) logTaskInput(task *robottypes.Task, prompt string) { +func (l *execLogger) logTaskInput(task *robottypes.Task, prompt string, actualConnector string) { if config.IsDevelopment() { - l.devTaskInput(task, prompt) + l.devTaskInput(task, prompt, actualConnector) } kunlog.With(kunlog.F{ "robot_id": l.robotID(), @@ -115,17 +115,23 @@ func (l *execLogger) logTaskInput(task *robottypes.Task, prompt string) { "executor_id": task.ExecutorID, "prompt_len": len(prompt), "language_model": l.connector(), + "connector": actualConnector, }).Info("Task input: %s [%s]", task.ID, task.ExecutorID) } -func (l *execLogger) devTaskInput(task *robottypes.Task, prompt string) { +func (l *execLogger) devTaskInput(task *robottypes.Task, prompt string, actualConnector string) { w := logger.Gray v := logger.White r := logger.Reset + connLabel := actualConnector + if connLabel == "" { + connLabel = "agent-default" + } + var sb strings.Builder - sb.WriteString(fmt.Sprintf("%s ▶ Task %s%s%s [%s:%s] Prompt: %d chars%s\n", - w, v, task.ID, w, task.ExecutorType, task.ExecutorID, len(prompt), r)) + sb.WriteString(fmt.Sprintf("%s ▶ Task %s%s%s [%s:%s] Connector: %s%s%s Prompt: %d chars%s\n", + w, v, task.ID, w, task.ExecutorType, task.ExecutorID, v, connLabel, w, len(prompt), r)) logger.Raw(sb.String()) } @@ -190,18 +196,19 @@ func (l *execLogger) devTaskOutput(task *robottypes.Task, result *robottypes.Tas // Agent Call // --------------------------------------------------------------------------- -func (l *execLogger) logAgentCall(agentID string, result *CallResult) { +func (l *execLogger) logAgentCall(agentID string, connector string, result *CallResult) { if result == nil { return } if config.IsDevelopment() { - l.devAgentCall(agentID, result) + l.devAgentCall(agentID, connector, result) } fields := kunlog.F{ "robot_id": l.robotID(), "execution_id": l.execID, "agent_id": agentID, + "connector": connector, "content_len": len(result.Content), "language_model": l.connector(), } @@ -209,23 +216,28 @@ func (l *execLogger) logAgentCall(agentID string, result *CallResult) { fields["next_type"] = fmt.Sprintf("%T", result.Next) fields["next_len"] = outputLen(result.Next) } - kunlog.With(fields).Info("Agent call: %s (content=%d, next=%T)", agentID, len(result.Content), result.Next) + kunlog.With(fields).Info("Agent call: %s (connector=%s, content=%d, next=%T)", agentID, connector, len(result.Content), result.Next) } -func (l *execLogger) devAgentCall(agentID string, result *CallResult) { +func (l *execLogger) devAgentCall(agentID string, connector string, result *CallResult) { w := logger.Gray v := logger.White c := logger.Cyan r := logger.Reset + displayConn := connector + if displayConn == "" { + displayConn = "agent-default" + } + nextInfo := "—" if result.Next != nil { nextInfo = fmt.Sprintf("%T (len=%d)", result.Next, outputLen(result.Next)) } var sb strings.Builder - sb.WriteString(fmt.Sprintf("%s → Agent(%s%s%s) Content: %s%d%s chars Next: %s%s%s\n", - c, v, agentID, c, v, len(result.Content), w, v, nextInfo, r)) + sb.WriteString(fmt.Sprintf("%s → Agent(%s%s%s) Connector: %s%s%s Content: %s%d%s chars Next: %s%s%s\n", + c, v, agentID, c, v, displayConn, c, v, len(result.Content), w, v, nextInfo, r)) logger.Raw(sb.String()) } diff --git a/agent/robot/executor/standard/prompts/workspace.yml b/agent/robot/executor/standard/prompts/workspace.yml new file mode 100644 index 00000000..b5664630 --- /dev/null +++ b/agent/robot/executor/standard/prompts/workspace.yml @@ -0,0 +1,60 @@ +workspace: | + ## Workspace + + Execution directory: {{.ExecDir}} + + ### Available Files + {{range .Files}} + - {{.Path}} — {{.Desc}} + {{- end}} + + ### Your Task Directory + + Write any output files to: {{.TaskDir}} + +context: | + ## Execution Context + + Goals: {{.Goals}} + {{- if .Locale}} + Language: {{.Locale}} + {{- end}} + Mode: automated-pipeline + {{- if .FailureWarning}} + + ⚠ {{.FailureWarning}} + {{- end}} + {{if .CompletedTasks}} + ### Completed Tasks + {{range .CompletedTasks}} + {{.Seq}}. [{{.ID}}] {{.Description}} ({{.ExecutorType}}: {{.Executor}}) ✓ + Summary: {{.Summary}} + {{- if .KeyOutputs}} + Key outputs: {{.KeyOutputs}} + {{- end}} + {{- if .HasFiles}} + Files: {{.Files}} + {{- end}} + {{end}} + {{- end}} + {{- if .FailedTasks}} + ### Failed Tasks + {{range .FailedTasks}} + {{.Seq}}. [{{.ID}}] {{.Description}} ✗ + Error: {{.Error}} + {{end}} + {{- end}} + ### Current Task + + {{.CurrentOrder}}. [{{.CurrentID}}] {{.CurrentDesc}} ({{.CurrentType}}: {{.CurrentExec}}) + +instructions: | + ## Task Instructions + + {{.TaskInstructions}} + {{- if .ExpectedOutput}} + + ### Expected Output + + {{.ExpectedOutput}} + {{- end}} diff --git a/agent/robot/executor/standard/run.go b/agent/robot/executor/standard/run.go index cfd91a26..c0a73dfe 100644 --- a/agent/robot/executor/standard/run.go +++ b/agent/robot/executor/standard/run.go @@ -2,8 +2,10 @@ package standard import ( "fmt" + "path" "time" + kunlog "github.com/yaoapp/kun/log" robotevents "github.com/yaoapp/yao/agent/robot/events" robottypes "github.com/yaoapp/yao/agent/robot/types" "github.com/yaoapp/yao/event" @@ -53,9 +55,6 @@ func (e *Executor) RunExecution(ctx *robottypes.Context, exec *robottypes.Execut config = DefaultRunConfig() } - // Determine locale for UI messages - locale := getEffectiveLocale(robot, exec.Input) - // Determine start index and restore results from resume context startIndex := 0 if exec.ResumeContext != nil { @@ -67,6 +66,26 @@ func (e *Executor) RunExecution(ctx *robottypes.Context, exec *robottypes.Execut // Create task runner with execution-level chatID (§8.4) runner := NewRunner(ctx, robot, config, exec.ChatID, exec.ID) + if ctx.Locale != "" { + runner.locale = ctx.Locale + } else { + runner.locale = getEffectiveLocale(robot, exec.Input) + } + + // Initialize workspace for file-based context + wsFS, err := ensureRobotWorkspace(ctx, robot) + if err != nil { + kunlog.Warn("[robot-run] workspace unavailable, falling back to in-memory context: %v", err) + } else { + execDir := path.Join("robots", robot.MemberID, exec.ID) + if mkErr := wsFS.MkdirAll(execDir, 0755); mkErr != nil { + kunlog.Warn("[robot-run] mkdir %s: %v", execDir, mkErr) + } else { + runner.wsFS = wsFS + runner.execDir = execDir + runner.initManifest(exec) + } + } // Execute tasks sequentially from startIndex for i := startIndex; i < len(exec.Tasks); i++ { @@ -80,7 +99,7 @@ func (e *Executor) RunExecution(ctx *robottypes.Context, exec *robottypes.Execut } // Update UI field with current task description (i18n) - taskName := formatTaskProgressName(task, i, len(exec.Tasks), locale) + taskName := formatTaskProgressName(task, i, len(exec.Tasks), runner.locale) e.updateUIFields(ctx, exec, "", taskName) // Mark task as running @@ -127,7 +146,10 @@ func (e *Executor) RunExecution(ctx *robottypes.Context, exec *robottypes.Execut }) } - // Store result + // Write task files to workspace (non-blocking, errors logged) + runner.writeTaskOutput(task, result, runner.lastPromptSnapshot) + + // Store result (in-memory, for persistence + resume) exec.Results = append(exec.Results, *result) // Persist completed/failed state to database diff --git a/agent/robot/executor/standard/runner.go b/agent/robot/executor/standard/runner.go index 9fedbebc..69ee8c62 100644 --- a/agent/robot/executor/standard/runner.go +++ b/agent/robot/executor/standard/runner.go @@ -10,16 +10,24 @@ import ( "github.com/yaoapp/gou/process" kunlog "github.com/yaoapp/kun/log" agentcontext "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/llm" robottypes "github.com/yaoapp/yao/agent/robot/types" + taiworkspace "github.com/yaoapp/yao/tai/workspace" ) // Runner handles execution of individual tasks type Runner struct { - ctx *robottypes.Context - robot *robottypes.Robot - config *RunConfig - chatID string // execution-level chatID for conversation persistence (§8.4) - log *execLogger + ctx *robottypes.Context + robot *robottypes.Robot + config *RunConfig + chatID string // execution-level chatID for conversation persistence (§8.4) + log *execLogger + wsFS taiworkspace.FS // workspace file system (nil if unavailable) + execDir string // workspace-relative execution directory (e.g. "robots//") + lastPromptSnapshot string // captured prompt text for workspace .input.md + currentTaskIndex int // current task index for workspace prompt building + currentExec *robottypes.Execution + locale string // effective locale for this execution (e.g. "zh", "en") } // NewRunner creates a new task runner @@ -47,12 +55,15 @@ type RunnerContext struct { // BuildTaskContext builds context for a task including previous results func (r *Runner) BuildTaskContext(exec *robottypes.Execution, taskIndex int) *RunnerContext { + r.currentTaskIndex = taskIndex + r.currentExec = exec + ctx := &RunnerContext{ Goals: exec.Goals, SystemPrompt: r.robot.SystemPrompt, } - // Include results from previous tasks (with bounds check) + // Include results from previous tasks (with bounds check) — kept for fallback if taskIndex > 0 && len(exec.Results) > 0 { endIndex := taskIndex if endIndex > len(exec.Results) { @@ -132,26 +143,56 @@ func (r *Runner) executeNonAssistantTask(task *robottypes.Task, taskCtx *RunnerC // Returns the extracted output, the raw CallResult (for need_input detection), and any error. func (r *Runner) executeAssistantTask(task *robottypes.Task, taskCtx *RunnerContext) (interface{}, *CallResult, error) { caller := NewAgentCaller() + caller.Mode = "task" caller.log = r.log - caller.Connector = r.robot.LanguageModel + if r.robot.LanguageModel != "" { + if _, _, err := llm.ResolveConnector(r.robot.LanguageModel, nil); err == nil { + caller.Connector = r.robot.LanguageModel + } else { + kunlog.Warn("[robot-runner] connector %s invalid, using agent default: %v", + r.robot.LanguageModel, err) + } + } caller.Workspace = r.robot.Workspace caller.ChatID = r.chatID - messages := r.BuildAssistantMessages(task, taskCtx) - input := r.FormatMessagesAsText(messages) + var input string + workspacePromptUsed := false + + // Use workspace-based prompt when available + if r.wsFS != nil { + manifest, err := r.readManifest() + if err == nil { + taskInstructions := r.FormatMessagesAsText(task.Messages) + input = r.buildWorkspacePrompt(manifest, r.currentTaskIndex, task, taskInstructions) + workspacePromptUsed = (input != "") + } + } + + // Fallback to legacy in-memory context + if input == "" { + messages := r.BuildAssistantMessages(task, taskCtx) + input = r.FormatMessagesAsText(messages) + } if strings.TrimSpace(input) == "" { return nil, nil, fmt.Errorf("no valid input messages for task %s", task.ID) } - if taskCtx.SystemPrompt != "" { + // Only inject robot system prompt for legacy (non-workspace) path. + // In workspace mode the agent's own prompts.yml defines its role; + // injecting the robot's dispatcher prompt would confuse the executor. + if taskCtx.SystemPrompt != "" && !workspacePromptUsed { input = "## Context\n\n" + taskCtx.SystemPrompt + "\n\n## Task\n\n" + input } + // Capture prompt snapshot for workspace .input.md + r.lastPromptSnapshot = 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) + r.log.logTaskInput(task, input, caller.Connector) result, err := caller.CallWithMessages(r.ctx, task.ExecutorID, input) if err != nil { diff --git a/agent/robot/executor/standard/tasks.go b/agent/robot/executor/standard/tasks.go index 3d447d0f..4992989b 100644 --- a/agent/robot/executor/standard/tasks.go +++ b/agent/robot/executor/standard/tasks.go @@ -2,7 +2,9 @@ package standard import ( "fmt" + "strings" + kunlog "github.com/yaoapp/kun/log" agentcontext "github.com/yaoapp/yao/agent/context" robottypes "github.com/yaoapp/yao/agent/robot/types" ) @@ -54,7 +56,6 @@ func (e *Executor) RunTasks(ctx *robottypes.Context, exec *robottypes.Execution, // Call agent caller := NewAgentCaller() caller.log = newExecLogger(robot, exec.ID) - caller.Connector = robot.LanguageModel caller.Workspace = robot.Workspace result, err := caller.CallWithMessages(ctx, agentID, userContent) if err != nil { @@ -80,6 +81,9 @@ func (e *Executor) RunTasks(ctx *robottypes.Context, exec *robottypes.Execution, return fmt.Errorf("tasks agent (%s) returned invalid task structure: %w", agentID, err) } + // Normalize executor IDs and types against available resources + NormalizeTaskExecutors(tasks, robot) + // Validate tasks if err := ValidateTasks(tasks); err != nil { return fmt.Errorf("tasks validation failed: %w", err) @@ -395,3 +399,81 @@ func ValidateMCPTask(task *robottypes.Task) error { } return nil } + +// NormalizeTaskExecutors fixes LLM-generated executor_id and executor_type +// against the robot's actual resource lists. It handles two common LLM errors: +// 1. Partial executor_id (e.g. "report-writer" instead of "yao.report-writer") +// 2. Wrong executor_type (e.g. classifying an assistant as "mcp") +func NormalizeTaskExecutors(tasks []robottypes.Task, robot *robottypes.Robot) { + if robot == nil || robot.Config == nil || robot.Config.Resources == nil { + return + } + + agentSet := make(map[string]bool, len(robot.Config.Resources.Agents)) + for _, id := range robot.Config.Resources.Agents { + agentSet[id] = true + } + + mcpSet := make(map[string]bool, len(robot.Config.Resources.MCP)) + for _, m := range robot.Config.Resources.MCP { + mcpSet[m.ID] = true + } + + for i := range tasks { + task := &tasks[i] + origID := task.ExecutorID + origType := task.ExecutorType + + // Skip process tasks — they are not in resource lists + if task.ExecutorType == robottypes.ExecutorProcess { + continue + } + + // Step 1: Try exact match first + if agentSet[task.ExecutorID] { + task.ExecutorType = robottypes.ExecutorAssistant + if origType != task.ExecutorType { + kunlog.Trace("[normalize] task %s: executor_type %s -> %s (crosscheck)", task.ID, origType, task.ExecutorType) + } + continue + } + if mcpSet[task.ExecutorID] { + task.ExecutorType = robottypes.ExecutorMCP + if origType != task.ExecutorType { + kunlog.Trace("[normalize] task %s: executor_type %s -> %s (crosscheck)", task.ID, origType, task.ExecutorType) + } + continue + } + + // Step 2: Suffix match — LLM may omit namespace prefix + if match := suffixMatch(task.ExecutorID, robot.Config.Resources.Agents); match != "" { + task.ExecutorID = match + task.ExecutorType = robottypes.ExecutorAssistant + kunlog.Trace("[normalize] task %s: executor_id %s -> %s (suffix match)", task.ID, origID, match) + continue + } + + mcpIDs := make([]string, 0, len(robot.Config.Resources.MCP)) + for _, m := range robot.Config.Resources.MCP { + mcpIDs = append(mcpIDs, m.ID) + } + if match := suffixMatch(task.ExecutorID, mcpIDs); match != "" { + task.ExecutorID = match + task.ExecutorType = robottypes.ExecutorMCP + kunlog.Trace("[normalize] task %s: executor_id %s -> %s (suffix match)", task.ID, origID, match) + continue + } + } +} + +// suffixMatch finds the first entry in candidates that ends with "."+partial +// (or equals partial exactly, which is already handled by the caller). +func suffixMatch(partial string, candidates []string) string { + suffix := "." + partial + for _, c := range candidates { + if strings.HasSuffix(c, suffix) { + return c + } + } + return "" +} diff --git a/agent/robot/executor/standard/validator.go b/agent/robot/executor/standard/validator.go index 9d597fa6..fdcbfe71 100644 --- a/agent/robot/executor/standard/validator.go +++ b/agent/robot/executor/standard/validator.go @@ -433,7 +433,6 @@ func (v *Validator) validateSemantic(task *robottypes.Task, output interface{}) // Call validation agent caller := NewAgentCaller() - caller.Connector = v.robot.LanguageModel caller.Workspace = v.robot.Workspace result, err := caller.CallWithMessages(v.ctx, validationAgentID, validationPrompt) if err != nil { @@ -641,7 +640,6 @@ func (av *robotAgentValidator) Validate(agentID string, output, input, criteria // Call agent caller := NewAgentCaller() - caller.Connector = av.v.robot.LanguageModel caller.Workspace = av.v.robot.Workspace callResult, err := caller.CallWithMessages(av.v.ctx, agentID, string(inputJSON)) if err != nil { diff --git a/agent/robot/executor/standard/workspace.go b/agent/robot/executor/standard/workspace.go new file mode 100644 index 00000000..5a4cec92 --- /dev/null +++ b/agent/robot/executor/standard/workspace.go @@ -0,0 +1,850 @@ +package standard + +import ( + "bytes" + _ "embed" + "encoding/json" + "fmt" + "path" + "path/filepath" + "regexp" + "strings" + "text/template" + + kunlog "github.com/yaoapp/kun/log" + agentcontext "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/llm" + robottypes "github.com/yaoapp/yao/agent/robot/types" + taiworkspace "github.com/yaoapp/yao/tai/workspace" + "github.com/yaoapp/yao/workspace" + "gopkg.in/yaml.v3" +) + +//go:embed prompts/workspace.yml +var workspacePromptYAML []byte + +type workspacePrompts struct { + Workspace string `yaml:"workspace"` + Context string `yaml:"context"` + Instructions string `yaml:"instructions"` +} + +var wsPromptTpls struct { + Workspace *template.Template + Context *template.Template + Instructions *template.Template +} + +func init() { + var p workspacePrompts + if err := yaml.Unmarshal(workspacePromptYAML, &p); err != nil { + return + } + wsPromptTpls.Workspace, _ = template.New("ws").Parse(p.Workspace) + wsPromptTpls.Context, _ = template.New("ctx").Parse(p.Context) + wsPromptTpls.Instructions, _ = template.New("inst").Parse(p.Instructions) +} + +// Manifest is the shared context hub for an execution, written as manifest.json. +type Manifest struct { + ExecID string `json:"exec_id"` + RobotID string `json:"robot_id"` + Goals string `json:"goals"` + Tasks []ManifestTask `json:"tasks"` +} + +// ManifestTask represents a single task entry in the manifest. +type ManifestTask struct { + ID string `json:"id"` + Order int `json:"order"` + Description string `json:"description"` + Executor string `json:"executor"` + ExecutorType string `json:"executor_type"` + Status string `json:"status"` + Summary string `json:"summary,omitempty"` + KeyOutputs []string `json:"key_outputs,omitempty"` + Files []ManifestFile `json:"files,omitempty"` + Error string `json:"error,omitempty"` +} + +// ManifestFile represents a produced artifact. +type ManifestFile struct { + Name string `json:"name"` + Type string `json:"type,omitempty"` + Desc string `json:"desc,omitempty"` + URI string `json:"uri,omitempty"` +} + +// ensureRobotWorkspace guarantees a workspace FS exists for the robot. +// If robot.Workspace is empty, it derives a deterministic ID and auto-creates. +// It also writes back robot.Workspace so subsequent callers get the correct ID. +func ensureRobotWorkspace(ctx *robottypes.Context, robot *robottypes.Robot) (taiworkspace.FS, error) { + wsm := workspace.M() + if wsm == nil { + return nil, fmt.Errorf("workspace manager not available") + } + + wsID := robot.Workspace + if wsID == "" { + nodes := wsm.Nodes() + nodeID := "" + for _, n := range nodes { + if n.Online { + nodeID = n.Name + break + } + } + if nodeID == "" { + return nil, fmt.Errorf("no available node for workspace") + } + wsID = workspace.DefaultWorkspaceID(robot.TeamID, nodeID) + + if _, err := wsm.Get(ctx, wsID); err != nil { + if _, err := wsm.Create(ctx, workspace.CreateOptions{ + ID: wsID, + Name: "Robot Workspace", + Owner: robot.TeamID, + Node: nodeID, + }); err != nil { + return nil, fmt.Errorf("workspace create failed: %w", err) + } + } + robot.Workspace = wsID + } + + return wsm.FS(ctx, wsID) +} + +// initManifest creates the initial manifest.json with goals and pending task list. +// Uses slice index (not P2's order field) so manifests always have sequential numbering. +func (r *Runner) initManifest(exec *robottypes.Execution) { + if r.wsFS == nil { + return + } + + goalsContent := "" + if exec.Goals != nil { + goalsContent = exec.Goals.Content + } + + m := &Manifest{ + ExecID: exec.ID, + RobotID: r.robot.MemberID, + Goals: goalsContent, + Tasks: make([]ManifestTask, 0, len(exec.Tasks)), + } + + for i, t := range exec.Tasks { + m.Tasks = append(m.Tasks, ManifestTask{ + ID: t.ID, + Order: i, + Description: t.Description, + Executor: t.ExecutorID, + ExecutorType: string(t.ExecutorType), + Status: string(t.Status), + }) + } + + r.writeManifest(m) +} + +// writeManifest serializes and writes manifest.json. +func (r *Runner) writeManifest(m *Manifest) { + data, err := json.MarshalIndent(m, "", " ") + if err != nil { + kunlog.Warn("[robot-workspace] marshal manifest: %v", err) + return + } + p := path.Join(r.execDir, "manifest.json") + if err := r.wsFS.WriteFile(p, data, 0644); err != nil { + kunlog.Warn("[robot-workspace] write manifest: %v", err) + } +} + +// readManifest reads and parses manifest.json from workspace. +func (r *Runner) readManifest() (*Manifest, error) { + if r.wsFS == nil { + return nil, fmt.Errorf("wsFS not available") + } + data, err := r.wsFS.ReadFile(path.Join(r.execDir, "manifest.json")) + if err != nil { + return nil, err + } + var m Manifest + if err := json.Unmarshal(data, &m); err != nil { + return nil, err + } + return &m, nil +} + +// writeTaskOutput writes the three task files and updates manifest after task completion. +func (r *Runner) writeTaskOutput(task *robottypes.Task, result *robottypes.TaskResult, promptSnapshot string) { + if r.wsFS == nil { + return + } + + taskID := task.ID + + // Write task-NNN.input.md (prompt snapshot for debug) + if promptSnapshot != "" { + inputPath := path.Join(r.execDir, taskID+".input.md") + if err := r.wsFS.WriteFile(inputPath, []byte(promptSnapshot), 0644); err != nil { + kunlog.Warn("[robot-workspace] write %s.input.md: %v", taskID, err) + } + } + + // Write task-NNN.output.md (full output) + outputText := formatOutputAsText(result.Output) + outputPath := path.Join(r.execDir, taskID+".output.md") + if err := r.wsFS.WriteFile(outputPath, []byte(outputText), 0644); err != nil { + kunlog.Warn("[robot-workspace] write %s.output.md: %v", taskID, err) + } + + // Write task-NNN.json (metadata) + meta := map[string]interface{}{ + "id": taskID, + "executor": task.ExecutorID, + "executor_type": string(task.ExecutorType), + "status": string(task.Status), + "duration_ms": result.Duration, + "success": result.Success, + } + if result.Error != "" { + meta["error"] = result.Error + } + metaJSON, _ := json.MarshalIndent(meta, "", " ") + metaPath := path.Join(r.execDir, taskID+".json") + if err := r.wsFS.WriteFile(metaPath, metaJSON, 0644); err != nil { + kunlog.Warn("[robot-workspace] write %s.json: %v", taskID, err) + } + + r.updateManifestForTask(task, result) +} + +// updateManifestForTask reads manifest, updates the matching task entry, and writes back. +func (r *Runner) updateManifestForTask(task *robottypes.Task, result *robottypes.TaskResult) { + m, err := r.readManifest() + if err != nil { + kunlog.Warn("[robot-workspace] read manifest for update: %v", err) + return + } + + // Scan files first so LLM summary can reference them + files := r.scanTaskArtifacts(task.ID) + outputURIs := r.extractAndVerifyFiles(result.Output) + files = mergeManifestFiles(files, outputURIs) + + for i := range m.Tasks { + if m.Tasks[i].ID == task.ID { + if result.Success { + m.Tasks[i].Status = "completed" + summary := r.llmSummarize(task, result.Output, files) + if summary == "" { + summary = generateSummary(result.Output) + } + m.Tasks[i].Summary = summary + m.Tasks[i].KeyOutputs = extractKeyOutputs(result.Output) + } else { + m.Tasks[i].Status = "failed" + m.Tasks[i].Error = result.Error + if result.Error != "" { + m.Tasks[i].Summary = "Failed: " + result.Error + } + } + m.Tasks[i].Files = files + break + } + } + + r.writeManifest(m) +} + +// scanTaskArtifacts scans the task-id/ directory for produced files. +func (r *Runner) scanTaskArtifacts(taskID string) []ManifestFile { + if r.wsFS == nil { + return nil + } + + dirPath := path.Join(r.execDir, taskID) + entries, err := r.wsFS.ReadDir(dirPath) + if err != nil { + return nil + } + + var files []ManifestFile + for _, e := range entries { + if e.IsDir() { + continue + } + files = append(files, ManifestFile{ + Name: e.Name(), + Type: mimeFromExt(filepath.Ext(e.Name())), + }) + } + return files +} + +// llmSummarize uses a lightweight LLM to generate a concise task summary +// from the full task context (description, output, produced files). +// Returns empty string on any failure, allowing caller to fall back to static extraction. +func (r *Runner) llmSummarize(task *robottypes.Task, output interface{}, files []ManifestFile) string { + text := flattenOutput(output) + if text == "" { + return "" + } + + conn, _, err := llm.ResolveConnector("use::light", r.ctx.Auth) + if err != nil { + kunlog.Warn("[robot-workspace] llmSummarize resolve connector: %v", err) + return "" + } + + opts := llm.BuildCompletionOptions(conn, nil) + instance, err := llm.New(conn, opts) + if err != nil { + kunlog.Warn("[robot-workspace] llmSummarize create LLM: %v", err) + return "" + } + + var sb strings.Builder + sb.WriteString("Task: " + task.Description + "\n\n") + if len(files) > 0 { + sb.WriteString("Produced files:\n") + for _, f := range files { + sb.WriteString("- " + f.Name + " (" + f.Type + ")\n") + } + sb.WriteString("\n") + } + sb.WriteString("Output:\n") + outputText := text + if len(outputText) > 4000 { + outputText = outputText[:4000] + } + sb.WriteString(outputText) + + messages := []agentcontext.Message{ + {Role: agentcontext.RoleSystem, Content: "Summarize the task execution result in 1-2 concise sentences. " + + "Focus on what was actually produced or accomplished, not the process. " + + "If files were produced, mention them. Reply in the same language as the task description."}, + {Role: agentcontext.RoleUser, Content: sb.String()}, + } + + agentCtx := agentcontext.New(r.ctx.Context, r.ctx.Auth, "") + defer agentCtx.Release() + + resp, err := instance.Post(agentCtx, messages, opts) + if err != nil { + kunlog.Warn("[robot-workspace] llmSummarize Post: %v", err) + return "" + } + + return extractLLMContent(resp) +} + +// extractLLMContent extracts the text content from a CompletionResponse. +func extractLLMContent(resp *agentcontext.CompletionResponse) string { + if resp == nil { + return "" + } + if s, ok := resp.Content.(string); ok { + return strings.TrimSpace(s) + } + return "" +} + +// workspaceURIRegex matches workspace://wsID/path patterns in markdown links and plain text. +// Excludes trailing backticks, quotes, and brackets that are markdown formatting artifacts. +var workspaceURIRegex = regexp.MustCompile("workspace://([^/\\s)]+)/([^\\s)\\]`\"']+)") + +// extractAndVerifyFiles extracts workspace:// URIs from output and verifies each +// file exists via wsFS.Stat, eliminating false positives from regex artifacts. +func (r *Runner) extractAndVerifyFiles(output interface{}) []ManifestFile { + text := flattenOutput(output) + if text == "" || r.wsFS == nil { + return nil + } + + wsID, err := r.wsFS.GetID() + if err != nil { + return nil + } + prefix := "workspace://" + wsID + "/" + + matches := workspaceURIRegex.FindAllStringSubmatch(text, -1) + seen := make(map[string]bool) + var files []ManifestFile + for _, m := range matches { + uri := "workspace://" + m[1] + "/" + m[2] + if seen[uri] { + continue + } + seen[uri] = true + + if !strings.HasPrefix(uri, prefix) { + continue + } + relPath := strings.TrimPrefix(uri, prefix) + + info, err := r.wsFS.Stat(relPath) + if err != nil || info.IsDir() { + continue + } + + name := filepath.Base(relPath) + files = append(files, ManifestFile{ + Name: name, + Type: mimeFromExt(filepath.Ext(name)), + URI: uri, + }) + } + return files +} + +// mergeManifestFiles deduplicates files from scanTaskArtifacts and extractWorkspaceURIs. +// If a URI-bearing entry has the same Name as a scan entry, the URI is merged onto the +// existing entry instead of creating a duplicate. +func mergeManifestFiles(scanned []ManifestFile, fromURIs []ManifestFile) []ManifestFile { + if len(fromURIs) == 0 { + return scanned + } + + nameIndex := make(map[string]int, len(scanned)) + for i, f := range scanned { + nameIndex[f.Name] = i + } + + for _, uf := range fromURIs { + if idx, exists := nameIndex[uf.Name]; exists { + if scanned[idx].URI == "" { + scanned[idx].URI = uf.URI + } + } else { + nameIndex[uf.Name] = len(scanned) + scanned = append(scanned, uf) + } + } + return scanned +} + +// --- Summary & key_outputs extraction --- + +// llmPrefixPatterns are common LLM filler prefixes that carry no information. +var llmPrefixPatterns = []string{ + "It seems ", "It appears ", "Here is ", "Here's ", + "Based on ", "I encountered ", "I wasn't able ", + "I'm currently unable ", "Let me ", "I recommend ", +} + +// generateSummary produces a concise summary of the task's actual output/result. +// It prioritizes conclusion/summary sections over the beginning of the output, +// because agent responses typically start with planning/thinking text. +func generateSummary(output interface{}) string { + text := flattenOutput(output) + if text == "" { + return "" + } + + const maxLen = 200 + + // Try to find an explicit summary/conclusion section + for _, heading := range []string{"## Summary", "## Conclusion", "## Result", "## 总结", "## 结论", "## 结果"} { + if idx := strings.Index(text, heading); idx >= 0 { + section := strings.TrimSpace(text[idx+len(heading):]) + section = strings.TrimPrefix(section, "\n") + if nextH := strings.Index(section, "\n## "); nextH > 0 { + section = section[:nextH] + } + section = strings.TrimSpace(section) + if section != "" { + if len(section) > maxLen { + return section[:maxLen] + "..." + } + return section + } + } + } + + // No explicit section — use last substantive paragraph as it's + // more likely to contain the actual result than the beginning + paragraphs := strings.Split(text, "\n\n") + for i := len(paragraphs) - 1; i >= 0; i-- { + p := strings.TrimSpace(paragraphs[i]) + if p == "" || len(p) < 10 { + continue + } + isFiller := false + for _, prefix := range llmPrefixPatterns { + if strings.HasPrefix(p, prefix) { + isFiller = true + break + } + } + if isFiller { + continue + } + if len(p) > maxLen { + return p[:maxLen] + "..." + } + return p + } + + // Fallback: skip filler and take from the beginning + text = skipFillerPrefixes(text) + if len(text) > maxLen { + return text[:maxLen] + "..." + } + return text +} + +// skipFillerPrefixes skips past common LLM opening phrases to find substantive content. +func skipFillerPrefixes(text string) string { + lines := strings.SplitN(text, "\n", 20) + for i, line := range lines { + trimmed := strings.TrimSpace(line) + if trimmed == "" { + continue + } + isFiller := false + for _, prefix := range llmPrefixPatterns { + if strings.HasPrefix(trimmed, prefix) { + isFiller = true + break + } + } + if !isFiller { + return strings.Join(lines[i:], "\n") + } + } + return text +} + +// boldPatternRegex matches **bold text** in markdown +var boldPatternRegex = regexp.MustCompile(`\*\*([^*]+)\*\*`) + +// extractKeyOutputs tries to extract structured key_outputs from the output. +func extractKeyOutputs(output interface{}) []string { + if output == nil { + return nil + } + + // If output is a map with key_outputs/outputs/results, extract directly + if m, ok := output.(map[string]interface{}); ok { + for _, key := range []string{"key_outputs", "outputs", "results"} { + if arr, ok := m[key].([]interface{}); ok { + result := make([]string, 0, len(arr)) + for _, v := range arr { + if s, ok := v.(string); ok { + result = append(result, s) + } + } + if len(result) > 0 { + return capSlice(result, 5) + } + } + } + } + + text := flattenOutput(output) + + // Try ## headings + if strings.Contains(text, "\n## ") { + var headings []string + for _, line := range strings.Split(text, "\n") { + if strings.HasPrefix(line, "## ") { + headings = append(headings, strings.TrimPrefix(line, "## ")) + } + } + if len(headings) > 0 { + return capSlice(headings, 5) + } + } + + // Try **bold** items from numbered lists or bullet lists + // e.g. "1. **Model-Driven Architecture:**" or "- **Low-Code Engine**" + var boldItems []string + for _, line := range strings.Split(text, "\n") { + trimmed := strings.TrimSpace(line) + if !strings.Contains(trimmed, "**") { + continue + } + // Only extract from list-like lines + if !(strings.HasPrefix(trimmed, "- ") || strings.HasPrefix(trimmed, "* ") || + (len(trimmed) > 2 && trimmed[0] >= '0' && trimmed[0] <= '9' && trimmed[1] == '.')) { + continue + } + matches := boldPatternRegex.FindStringSubmatch(trimmed) + if len(matches) >= 2 { + item := strings.TrimRight(matches[1], ":") + if len(item) > 0 && len(item) < 80 { + boldItems = append(boldItems, item) + } + } + } + if len(boldItems) > 0 { + return capSlice(boldItems, 5) + } + + return nil +} + +func capSlice(s []string, max int) []string { + if len(s) > max { + return s[:max] + } + return s +} + +// flattenOutput converts any output value to a plain text string. +func flattenOutput(output interface{}) string { + if output == nil { + return "" + } + switch v := output.(type) { + case string: + return v + case map[string]interface{}: + if text, ok := v["text"].(string); ok { + return text + } + if content, ok := v["content"].(string); ok { + return content + } + b, _ := json.Marshal(v) + return string(b) + default: + b, err := json.Marshal(v) + if err != nil { + return fmt.Sprintf("%v", v) + } + return string(b) + } +} + +// formatOutputAsText converts task output to markdown text for .output.md files. +func formatOutputAsText(output interface{}) string { + if output == nil { + return "" + } + switch v := output.(type) { + case string: + return v + default: + b, err := json.MarshalIndent(v, "", " ") + if err != nil { + return fmt.Sprintf("%v", v) + } + return string(b) + } +} + +// --- Prompt template rendering --- + +// wsTemplateData holds template variables for the workspace section. +type wsTemplateData struct { + ExecDir string + Files []wsFileEntry + TaskDir string +} + +type wsFileEntry struct { + Path string + Desc string +} + +// ctxTemplateData holds template variables for the execution context section. +type ctxTemplateData struct { + Goals string + Locale string + CompletedTasks []ctxTaskEntry + FailedTasks []ctxFailedEntry + CurrentOrder int + CurrentID string + CurrentDesc string + CurrentType string + CurrentExec string + FailureWarning string +} + +type ctxTaskEntry struct { + Seq int + ID string + Description string + ExecutorType string + Executor string + Summary string + KeyOutputs string + Files string + HasFiles bool +} + +type ctxFailedEntry struct { + Seq int + ID string + Description string + Error string +} + +// instTemplateData holds template variables for the instructions section. +type instTemplateData struct { + TaskInstructions string + ExpectedOutput string +} + +// buildWorkspacePrompt renders the full prompt for a task from manifest + template. +func (r *Runner) buildWorkspacePrompt(manifest *Manifest, taskIndex int, task *robottypes.Task, taskInstructions string) string { + if manifest == nil || taskIndex >= len(manifest.Tasks) { + return taskInstructions + } + + var sb strings.Builder + + // Section 1: Workspace + if wsPromptTpls.Workspace != nil { + wd := wsTemplateData{ + ExecDir: r.execDir + "/", + TaskDir: r.execDir + "/" + manifest.Tasks[taskIndex].ID + "/", + } + wd.Files = append(wd.Files, wsFileEntry{ + Path: "manifest.json", + Desc: "Execution context: goals, completed task summaries, progress", + }) + for i := 0; i < taskIndex; i++ { + t := manifest.Tasks[i] + if t.Status == "completed" { + wd.Files = append(wd.Files, wsFileEntry{ + Path: t.ID + ".output.md", + Desc: "Full output: " + t.Description, + }) + for _, f := range t.Files { + desc := "Artifact: " + f.Name + if f.URI != "" { + desc = "Artifact: " + f.Name + " (" + f.URI + ")" + } + filePath := t.ID + "/" + f.Name + if f.URI != "" { + filePath = f.URI + } + wd.Files = append(wd.Files, wsFileEntry{ + Path: filePath, + Desc: desc, + }) + } + } + } + var buf bytes.Buffer + if err := wsPromptTpls.Workspace.Execute(&buf, wd); err == nil { + sb.WriteString(buf.String()) + sb.WriteString("\n\n") + } + } + + // Section 2: Execution Context + if wsPromptTpls.Context != nil { + ct := manifest.Tasks[taskIndex] + cd := ctxTemplateData{ + Goals: manifest.Goals, + Locale: r.locale, + CurrentID: ct.ID, + CurrentOrder: taskIndex + 1, + CurrentDesc: ct.Description, + CurrentType: ct.ExecutorType, + CurrentExec: ct.Executor, + } + + seq := 0 + failedCount := 0 + for i := 0; i < taskIndex; i++ { + t := manifest.Tasks[i] + seq++ + if t.Status == "completed" { + entry := ctxTaskEntry{ + Seq: seq, + ID: t.ID, + Description: t.Description, + ExecutorType: t.ExecutorType, + Executor: t.Executor, + Summary: t.Summary, + KeyOutputs: strings.Join(t.KeyOutputs, ", "), + } + if len(t.Files) > 0 { + entry.HasFiles = true + names := make([]string, 0, len(t.Files)) + for _, f := range t.Files { + names = append(names, f.Name) + } + entry.Files = strings.Join(names, ", ") + } + cd.CompletedTasks = append(cd.CompletedTasks, entry) + } else if t.Status == "failed" { + failedCount++ + errMsg := t.Error + if errMsg == "" { + errMsg = t.Summary + } + cd.FailedTasks = append(cd.FailedTasks, ctxFailedEntry{ + Seq: seq, + ID: t.ID, + Description: t.Description, + Error: errMsg, + }) + } + } + + // P8: failure cascade warning + if failedCount > 0 && len(cd.CompletedTasks) == 0 { + cd.FailureWarning = "WARNING: All previous tasks failed. You may lack necessary input data. Do your best with available information or report the limitation." + } else if failedCount > 0 { + cd.FailureWarning = fmt.Sprintf("Note: %d previous task(s) failed. Some expected input may be missing.", failedCount) + } + + var buf bytes.Buffer + if err := wsPromptTpls.Context.Execute(&buf, cd); err == nil { + sb.WriteString(buf.String()) + sb.WriteString("\n\n") + } + } + + // Section 3: Task Instructions (enriched with expected_output from P2) + if wsPromptTpls.Instructions != nil { + instData := instTemplateData{TaskInstructions: taskInstructions} + if task != nil && task.ExpectedOutput != "" { + instData.ExpectedOutput = task.ExpectedOutput + } + var buf bytes.Buffer + if err := wsPromptTpls.Instructions.Execute(&buf, instData); err == nil { + sb.WriteString(buf.String()) + } + } else { + sb.WriteString("## Task Instructions\n\n") + sb.WriteString(taskInstructions) + } + + return sb.String() +} + +func mimeFromExt(ext string) string { + switch strings.ToLower(ext) { + case ".md": + return "text/markdown" + case ".html", ".htm": + return "text/html" + case ".json": + return "application/json" + case ".pdf": + return "application/pdf" + case ".png": + return "image/png" + case ".jpg", ".jpeg": + return "image/jpeg" + case ".csv": + return "text/csv" + case ".txt": + return "text/plain" + case ".xlsx": + return "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + case ".pptx": + return "application/vnd.openxmlformats-officedocument.presentationml.presentation" + default: + return "application/octet-stream" + } +} diff --git a/agent/robot/executor/standard/workspace_test.go b/agent/robot/executor/standard/workspace_test.go new file mode 100644 index 00000000..88ab31ea --- /dev/null +++ b/agent/robot/executor/standard/workspace_test.go @@ -0,0 +1,371 @@ +package standard + +import ( + "context" + "path" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/yaoapp/gou/connector" + "github.com/yaoapp/gou/store" + "github.com/yaoapp/yao/agent" + agentcontext "github.com/yaoapp/yao/agent/context" + robottypes "github.com/yaoapp/yao/agent/robot/types" + "github.com/yaoapp/yao/agent/testutils" + "github.com/yaoapp/yao/llmprovider" + oauthtypes "github.com/yaoapp/yao/openapi/oauth/types" + "github.com/yaoapp/yao/setting" + "github.com/yaoapp/yao/tai/volume" + taiworkspace "github.com/yaoapp/yao/tai/workspace" +) + +// ============================================================================ +// extractLLMContent — pure unit tests (no external dependencies) +// ============================================================================ + +func TestExtractLLMContent(t *testing.T) { + t.Run("string content", func(t *testing.T) { + resp := &agentcontext.CompletionResponse{Content: " hello world "} + assert.Equal(t, "hello world", extractLLMContent(resp)) + }) + + t.Run("nil response", func(t *testing.T) { + assert.Equal(t, "", extractLLMContent(nil)) + }) + + t.Run("non-string content", func(t *testing.T) { + resp := &agentcontext.CompletionResponse{Content: []interface{}{"a", "b"}} + assert.Equal(t, "", extractLLMContent(resp)) + }) + + t.Run("empty string content", func(t *testing.T) { + resp := &agentcontext.CompletionResponse{Content: " "} + assert.Equal(t, "", extractLLMContent(resp)) + }) + + t.Run("multiline content trimmed", func(t *testing.T) { + resp := &agentcontext.CompletionResponse{Content: "\n summary line\n"} + assert.Equal(t, "summary line", extractLLMContent(resp)) + }) + + t.Run("int content", func(t *testing.T) { + resp := &agentcontext.CompletionResponse{Content: 42} + assert.Equal(t, "", extractLLMContent(resp)) + }) +} + +// ============================================================================ +// extractAndVerifyFiles — unit tests with local volume-backed FS +// ============================================================================ + +func newTestWorkspaceFS(t *testing.T) taiworkspace.FS { + t.Helper() + dir := t.TempDir() + vol := volume.NewLocal(dir) + t.Cleanup(func() { vol.Close() }) + wfs := taiworkspace.New(vol, "ws-test") + t.Cleanup(func() { wfs.Close() }) + return wfs +} + +func TestExtractAndVerifyFiles(t *testing.T) { + t.Run("valid URI with existing file", func(t *testing.T) { + wfs := newTestWorkspaceFS(t) + require.NoError(t, wfs.MkdirAll("robots/r1/exec1/task-001", 0755)) + require.NoError(t, wfs.WriteFile("robots/r1/exec1/task-001/notes.md", []byte("hello"), 0644)) + + r := &Runner{ + wsFS: wfs, + execDir: "robots/r1/exec1", + } + output := "I wrote the file to workspace://ws-test/robots/r1/exec1/task-001/notes.md for you." + files := r.extractAndVerifyFiles(output) + + require.Len(t, files, 1) + assert.Equal(t, "notes.md", files[0].Name) + assert.Equal(t, "text/markdown", files[0].Type) + assert.Equal(t, "workspace://ws-test/robots/r1/exec1/task-001/notes.md", files[0].URI) + }) + + t.Run("URI with non-existent file filtered out", func(t *testing.T) { + wfs := newTestWorkspaceFS(t) + + r := &Runner{ + wsFS: wfs, + execDir: "robots/r1/exec1", + } + output := "See workspace://ws-test/robots/r1/exec1/task-001/missing.pdf" + files := r.extractAndVerifyFiles(output) + + assert.Empty(t, files) + }) + + t.Run("URI with trailing backtick excluded by regex", func(t *testing.T) { + wfs := newTestWorkspaceFS(t) + require.NoError(t, wfs.MkdirAll("robots/r1/exec1/task-001", 0755)) + require.NoError(t, wfs.WriteFile("robots/r1/exec1/task-001/data.json", []byte("{}"), 0644)) + + r := &Runner{ + wsFS: wfs, + execDir: "robots/r1/exec1", + } + // Backtick-wrapped URI — the regex excludes the backtick from the captured path + output := "`workspace://ws-test/robots/r1/exec1/task-001/data.json`" + files := r.extractAndVerifyFiles(output) + + require.Len(t, files, 1) + assert.Equal(t, "data.json", files[0].Name) + assert.Equal(t, "application/json", files[0].Type) + }) + + t.Run("duplicate URIs deduplicated", func(t *testing.T) { + wfs := newTestWorkspaceFS(t) + require.NoError(t, wfs.MkdirAll("robots/r1/exec1/task-001", 0755)) + require.NoError(t, wfs.WriteFile("robots/r1/exec1/task-001/report.html", []byte("

hi

"), 0644)) + + r := &Runner{ + wsFS: wfs, + execDir: "robots/r1/exec1", + } + output := "workspace://ws-test/robots/r1/exec1/task-001/report.html and again workspace://ws-test/robots/r1/exec1/task-001/report.html" + files := r.extractAndVerifyFiles(output) + + require.Len(t, files, 1) + assert.Equal(t, "report.html", files[0].Name) + }) + + t.Run("empty output returns nil", func(t *testing.T) { + wfs := newTestWorkspaceFS(t) + r := &Runner{wsFS: wfs, execDir: "robots/r1/exec1"} + assert.Nil(t, r.extractAndVerifyFiles("")) + }) + + t.Run("nil wsFS returns nil", func(t *testing.T) { + r := &Runner{wsFS: nil, execDir: "robots/r1/exec1"} + assert.Nil(t, r.extractAndVerifyFiles("some text with workspace://ws-test/foo/bar")) + }) + + t.Run("URI from different workspace filtered", func(t *testing.T) { + wfs := newTestWorkspaceFS(t) + r := &Runner{wsFS: wfs, execDir: "robots/r1/exec1"} + output := "See workspace://other-ws/robots/r1/exec1/task-001/file.txt" + files := r.extractAndVerifyFiles(output) + assert.Empty(t, files) + }) + + t.Run("directory URI filtered out", func(t *testing.T) { + wfs := newTestWorkspaceFS(t) + require.NoError(t, wfs.MkdirAll("robots/r1/exec1/task-001", 0755)) + + r := &Runner{wsFS: wfs, execDir: "robots/r1/exec1"} + output := "workspace://ws-test/robots/r1/exec1/task-001" + files := r.extractAndVerifyFiles(output) + assert.Empty(t, files) + }) + + t.Run("multiple valid files extracted", func(t *testing.T) { + wfs := newTestWorkspaceFS(t) + require.NoError(t, wfs.MkdirAll("robots/r1/exec1/task-002", 0755)) + require.NoError(t, wfs.WriteFile("robots/r1/exec1/task-002/slides.html", []byte(""), 0644)) + require.NoError(t, wfs.WriteFile("robots/r1/exec1/task-002/slides.pdf", []byte("%PDF"), 0644)) + + r := &Runner{wsFS: wfs, execDir: "robots/r1/exec1"} + output := "Generated workspace://ws-test/robots/r1/exec1/task-002/slides.html and exported workspace://ws-test/robots/r1/exec1/task-002/slides.pdf" + files := r.extractAndVerifyFiles(output) + + require.Len(t, files, 2) + names := []string{files[0].Name, files[1].Name} + assert.Contains(t, names, "slides.html") + assert.Contains(t, names, "slides.pdf") + }) +} + +// ============================================================================ +// mergeManifestFiles — pure unit tests +// ============================================================================ + +func TestMergeManifestFiles(t *testing.T) { + t.Run("empty fromURIs returns scanned unchanged", func(t *testing.T) { + scanned := []ManifestFile{{Name: "a.md", Type: "text/markdown"}} + result := mergeManifestFiles(scanned, nil) + assert.Equal(t, scanned, result) + }) + + t.Run("merge URI onto matching scanned entry", func(t *testing.T) { + scanned := []ManifestFile{{Name: "notes.md", Type: "text/markdown"}} + fromURIs := []ManifestFile{{Name: "notes.md", Type: "text/markdown", URI: "workspace://ws/path/notes.md"}} + result := mergeManifestFiles(scanned, fromURIs) + + require.Len(t, result, 1) + assert.Equal(t, "workspace://ws/path/notes.md", result[0].URI) + }) + + t.Run("add new URI entry when no scan match", func(t *testing.T) { + scanned := []ManifestFile{{Name: "a.md", Type: "text/markdown"}} + fromURIs := []ManifestFile{{Name: "b.pdf", Type: "application/pdf", URI: "workspace://ws/b.pdf"}} + result := mergeManifestFiles(scanned, fromURIs) + + require.Len(t, result, 2) + assert.Equal(t, "b.pdf", result[1].Name) + }) +} + +// ============================================================================ +// llmSummarize — integration test (real LLM call) +// ============================================================================ + +func setupLLMProvider(t *testing.T) { + t.Helper() + + if err := setting.Init(); err != nil { + t.Skipf("setting.Init failed (store not available): %v", err) + } + if err := llmprovider.Init(); err != nil { + t.Skipf("llmprovider.Init failed: %v", err) + } + if err := agent.SyncLLMDefaults(); err != nil { + t.Skipf("SyncLLMDefaults failed: %v", err) + } + + connIDs := connector.AIConnectors + if len(connIDs) == 0 { + t.Skip("no AI connectors available in test env") + } + + t.Cleanup(func() { + s, _ := store.Get("__yao.store") + if s != nil { + s.Del("llmprovider:*") + } + c, _ := store.Get("__yao.cache") + if c != nil { + c.Del("llmprovider:*") + } + }) +} + +func TestLLMSummarize(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test (requires real LLM)") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + setupLLMProvider(t) + + auth := &oauthtypes.AuthorizedInfo{UserID: "test-user", TeamID: "test-team"} + ctx := robottypes.NewContext(context.Background(), auth) + + r := &Runner{ + ctx: ctx, + } + + task := &robottypes.Task{ + ID: "task-001", + Description: "Research Yao Agents platform and compile structured notes", + } + + output := `## Research Findings + +Yao Agents is an AI-powered platform that enables developers to build intelligent agents. + +### Key Features +- **Model-Driven Architecture**: Define data models in YAML/JSON +- **Low-Code Engine**: Visual workflow builder +- **Multi-Agent Orchestration**: Coordinate multiple AI agents + +### Conclusion +Yao Agents provides a comprehensive toolkit for building production-ready AI applications with minimal boilerplate code.` + + files := []ManifestFile{ + {Name: "research-notes.md", Type: "text/markdown"}, + } + + summary := r.llmSummarize(task, output, files) + + t.Logf("LLM Summary: %s", summary) + assert.NotEmpty(t, summary, "summary should not be empty") + assert.Less(t, len(summary), 500, "summary should be concise (< 500 chars)") +} + +// ============================================================================ +// updateManifestForTask end-to-end — integration test +// ============================================================================ + +func TestUpdateManifestForTaskWithLLMSummary(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test (requires real LLM)") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + setupLLMProvider(t) + + wfs := newTestWorkspaceFS(t) + auth := &oauthtypes.AuthorizedInfo{UserID: "test-user", TeamID: "test-team"} + ctx := robottypes.NewContext(context.Background(), auth) + + execDir := "robots/r1/exec1" + r := &Runner{ + ctx: ctx, + wsFS: wfs, + execDir: execDir, + robot: &robottypes.Robot{MemberID: "r1"}, + } + + task := robottypes.Task{ + ID: "task-001", + Description: "Research Yao Agents platform", + ExecutorID: "yao.general", + ExecutorType: robottypes.ExecutorAssistant, + Status: robottypes.TaskPending, + } + exec := &robottypes.Execution{ + ID: "exec1", + Tasks: []robottypes.Task{task}, + } + + r.initManifest(exec) + + // Verify manifest was created with pending status + m, err := r.readManifest() + require.NoError(t, err) + require.Len(t, m.Tasks, 1) + assert.Equal(t, "pending", m.Tasks[0].Status) + + // Write an artifact file that scanTaskArtifacts can discover + require.NoError(t, wfs.MkdirAll(path.Join(execDir, "task-001"), 0755)) + require.NoError(t, wfs.WriteFile(path.Join(execDir, "task-001", "notes.md"), []byte("research content"), 0644)) + + // Simulate task completion with output referencing the artifact + wsID, _ := wfs.GetID() + result := &robottypes.TaskResult{ + Success: true, + Duration: 5000, + Output: "Completed research. Notes saved to workspace://" + wsID + "/" + execDir + "/task-001/notes.md", + } + + r.updateManifestForTask(&task, result) + + // Re-read and verify + m, err = r.readManifest() + require.NoError(t, err) + require.Len(t, m.Tasks, 1) + + mt := m.Tasks[0] + assert.Equal(t, "completed", mt.Status) + assert.NotEmpty(t, mt.Summary, "summary should be generated (LLM or fallback)") + t.Logf("Summary: %s", mt.Summary) + + // Files should include the scanned artifact, potentially with URI merged + assert.NotEmpty(t, mt.Files, "files should be populated") + found := false + for _, f := range mt.Files { + if f.Name == "notes.md" { + found = true + assert.Equal(t, "text/markdown", f.Type) + } + } + assert.True(t, found, "notes.md should be in files list") +} diff --git a/agent/robot/manager/interact.go b/agent/robot/manager/interact.go index 100758f9..40dcbda0 100644 --- a/agent/robot/manager/interact.go +++ b/agent/robot/manager/interact.go @@ -337,7 +337,6 @@ func (m *Manager) callHostAgent(ctx *types.Context, agentID string, input *types } caller := standard.NewConversationCaller(chatID) - caller.Connector = robot.LanguageModel caller.Workspace = robot.Workspace result, err := caller.CallWithMessages(ctx, agentID, string(inputJSON)) if err != nil { @@ -740,7 +739,6 @@ func (m *Manager) callHostAgentStream(ctx *types.Context, agentID string, input } caller := standard.NewConversationCaller(chatID) - caller.Connector = robot.LanguageModel caller.Workspace = robot.Workspace result, err := caller.CallWithMessagesStream(ctx, agentID, string(inputJSON), streamFn) if err != nil { @@ -960,7 +958,6 @@ func (m *Manager) callHostAgentStreamRaw(ctx *types.Context, agentID string, inp } caller := standard.NewConversationCaller(chatID) - caller.Connector = robot.LanguageModel caller.Workspace = robot.Workspace result, err := caller.CallWithMessagesStreamRaw(ctx, agentID, string(inputJSON), wrappedOnMessage) if err != nil { diff --git a/agent/robot/manager/manager.go b/agent/robot/manager/manager.go index 9bf7c649..d4908411 100644 --- a/agent/robot/manager/manager.go +++ b/agent/robot/manager/manager.go @@ -524,6 +524,7 @@ func (m *Manager) Intervene(ctx *types.Context, req *types.InterveneRequest) (*t Action: req.Action, Messages: req.Messages, UserID: ctx.UserID(), + Locale: req.Locale, } // Handle plan.add action - schedule for later diff --git a/agent/robot/types/request.go b/agent/robot/types/request.go index 0269c9f2..71004001 100644 --- a/agent/robot/types/request.go +++ b/agent/robot/types/request.go @@ -14,6 +14,7 @@ type InterveneRequest struct { Messages []agentcontext.Message `json:"messages"` // user input (text, images, files) PlanTime *time.Time `json:"plan_time,omitempty"` // for action=plan ExecutorMode ExecutorMode `json:"executor_mode,omitempty"` // optional: override robot config + Locale string `json:"locale,omitempty"` // language for UI display (e.g., "en", "zh") } // EventRequest - event trigger request diff --git a/agent/sandbox/v2/claude/parse.go b/agent/sandbox/v2/claude/parse.go index ca4542cc..d3ab29b1 100644 --- a/agent/sandbox/v2/claude/parse.go +++ b/agent/sandbox/v2/claude/parse.go @@ -11,6 +11,7 @@ import ( "github.com/yaoapp/kun/log" "github.com/yaoapp/yao/agent/output/message" + "github.com/yaoapp/yao/agent/sandbox/v2/shared" ) // streamParser is an explicit state machine for Claude CLI stream-json output. @@ -77,8 +78,7 @@ 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) + reader := bufio.NewReaderSize(stdout, 64*1024) startTime := time.Now() lineCount := 0 @@ -87,9 +87,22 @@ func (p *streamParser) parse(ctx context.Context, stdout io.ReadCloser) error { log.Trace("[claude-parse] stream started") - for scanner.Scan() { - line := scanner.Text() - if line == "" { + for { + line, skipped, err := shared.ReadJSONLine(reader) + if err == io.EOF { + break + } + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return err + } + if skipped { + log.Warn("[claude-parse] skipped oversized JSONL line (>%dMB)", shared.MaxLineSize/1024/1024) + continue + } + if len(line) == 0 { continue } lineCount++ @@ -105,11 +118,11 @@ func (p *streamParser) parse(ctx context.Context, stdout io.ReadCloser) error { } var msg map[string]any - if err := json.Unmarshal([]byte(line), &msg); err != nil { + if err := json.Unmarshal(line, &msg); err != nil { if len(line) > 200 { - log.Trace("[claude-parse] JSON unmarshal error: %v (line len=%d, prefix=%q)", err, len(line), line[:200]) + log.Trace("[claude-parse] JSON unmarshal error: %v (line len=%d, prefix=%q)", err, len(line), string(line[:200])) } else { - log.Trace("[claude-parse] JSON unmarshal error: %v (line=%q)", err, line) + log.Trace("[claude-parse] JSON unmarshal error: %v (line=%q)", err, string(line)) } continue } @@ -141,16 +154,9 @@ func (p *streamParser) parse(ctx context.Context, stdout io.ReadCloser) error { } } - 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()) + log.Trace("[claude-parse] stream ended: lines=%d elapsed=%v completed=%v", + lineCount, time.Since(startTime).Round(time.Second), p.completed) - if err := scanner.Err(); err != nil { - log.Trace("[claude-parse] scanner error: %v (ctx.Err=%v)", err, ctx.Err()) - if ctx.Err() != nil { - return ctx.Err() - } - return err - } return nil } diff --git a/agent/sandbox/v2/claude/session.go b/agent/sandbox/v2/claude/session.go index 0ff85a77..cf0b85be 100644 --- a/agent/sandbox/v2/claude/session.go +++ b/agent/sandbox/v2/claude/session.go @@ -63,6 +63,10 @@ func (s *session) runStream(handler message.StreamFunc) (completed bool, err err s.logger.Debug("runStream: parse returned completed=%v parseErr=%v", parser.completed, parseErr) + if !parser.completed && parseErr != nil { + s.exec.Cancel() + } + if parser.completed { s.logger.Info("claude stream completed normally") return true, nil diff --git a/agent/sandbox/v2/opencode/parse.go b/agent/sandbox/v2/opencode/parse.go index 3060b8fd..f8d2efa0 100644 --- a/agent/sandbox/v2/opencode/parse.go +++ b/agent/sandbox/v2/opencode/parse.go @@ -11,6 +11,7 @@ import ( "github.com/yaoapp/kun/log" "github.com/yaoapp/yao/agent/output/message" + "github.com/yaoapp/yao/agent/sandbox/v2/shared" ) // streamParser handles OpenCode's JSONL output (--format json). @@ -63,8 +64,7 @@ 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) + reader := bufio.NewReaderSize(stdout, 64*1024) startTime := time.Now() lineCount := 0 @@ -72,9 +72,22 @@ func (p *streamParser) parse(ctx context.Context, stdout io.ReadCloser) error { log.Trace("[opencode-parse] stream started") - for scanner.Scan() { - line := scanner.Text() - if line == "" { + for { + line, skipped, err := shared.ReadJSONLine(reader) + if err == io.EOF { + break + } + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return err + } + if skipped { + log.Warn("[opencode-parse] skipped oversized JSONL line (>%dMB)", shared.MaxLineSize/1024/1024) + continue + } + if len(line) == 0 { continue } lineCount++ @@ -86,11 +99,11 @@ func (p *streamParser) parse(ctx context.Context, stdout io.ReadCloser) error { } var msg map[string]any - if err := json.Unmarshal([]byte(line), &msg); err != nil { + if err := json.Unmarshal(line, &msg); err != nil { if len(line) > 200 { log.Trace("[opencode-parse] JSON unmarshal error: %v (line len=%d)", err, len(line)) } else { - log.Trace("[opencode-parse] JSON unmarshal error: %v (line=%q)", err, line) + log.Trace("[opencode-parse] JSON unmarshal error: %v (line=%q)", err, string(line)) } continue } @@ -131,15 +144,9 @@ func (p *streamParser) parse(ctx context.Context, stdout io.ReadCloser) error { } } - log.Trace("[opencode-parse] stream ended: lines=%d elapsed=%v completed=%v scanErr=%v", - lineCount, time.Since(startTime).Round(time.Second), p.completed, scanner.Err()) + log.Trace("[opencode-parse] stream ended: lines=%d elapsed=%v completed=%v", + lineCount, time.Since(startTime).Round(time.Second), p.completed) - if err := scanner.Err(); err != nil { - if ctx.Err() != nil { - return ctx.Err() - } - return err - } return nil } diff --git a/agent/sandbox/v2/opencode/session.go b/agent/sandbox/v2/opencode/session.go index 7be5c9f7..eb201ccf 100644 --- a/agent/sandbox/v2/opencode/session.go +++ b/agent/sandbox/v2/opencode/session.go @@ -78,6 +78,10 @@ func (s *session) runStream(handler message.StreamFunc) (completed bool, err err s.logger.Debug("runStream: parse returned completed=%v parseErr=%v", parser.completed, parseErr) + if !parser.completed && parseErr != nil { + s.exec.Cancel() + } + if parser.completed { s.logger.Info("opencode stream completed normally") return true, nil diff --git a/agent/sandbox/v2/shared/linereader.go b/agent/sandbox/v2/shared/linereader.go new file mode 100644 index 00000000..29ccf7cd --- /dev/null +++ b/agent/sandbox/v2/shared/linereader.go @@ -0,0 +1,43 @@ +package shared + +import "bufio" + +// MaxLineSize is the safety threshold for a single JSONL line. +// Lines exceeding this limit are drained and discarded to prevent +// unbounded memory growth (e.g. Claude CLI "result" events containing +// full conversation history). +const MaxLineSize = 50 * 1024 * 1024 // 50MB + +// ReadJSONLine reads one complete line from a bufio.Reader. +// +// Returns: +// - line: complete line bytes (without \n); nil when skipped +// - skipped: true if line exceeded MaxLineSize and was drained +// - err: io.EOF at end of stream, or other IO error +func ReadJSONLine(r *bufio.Reader) ([]byte, bool, error) { + var buf []byte + for { + chunk, isPrefix, err := r.ReadLine() + if err != nil { + return nil, false, err + } + buf = append(buf, chunk...) + if len(buf) > MaxLineSize { + if !isPrefix { + buf = nil + return nil, true, nil + } + for isPrefix { + _, isPrefix, err = r.ReadLine() + if err != nil { + return nil, true, err + } + } + buf = nil + return nil, true, nil + } + if !isPrefix { + return buf, false, nil + } + } +} diff --git a/agent/sandbox/v2/shared/linereader_test.go b/agent/sandbox/v2/shared/linereader_test.go new file mode 100644 index 00000000..d5e2c598 --- /dev/null +++ b/agent/sandbox/v2/shared/linereader_test.go @@ -0,0 +1,234 @@ +package shared + +import ( + "bufio" + "bytes" + "fmt" + "io" + "strings" + "testing" +) + +func TestReadJSONLine_ShortLine(t *testing.T) { + input := `{"type":"system","text":"hello"}` + "\n" + r := bufio.NewReaderSize(strings.NewReader(input), 64*1024) + + line, skipped, err := ReadJSONLine(r) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if skipped { + t.Fatal("expected skipped=false") + } + if string(line) != `{"type":"system","text":"hello"}` { + t.Fatalf("unexpected line: %q", line) + } +} + +func TestReadJSONLine_EmptyLine(t *testing.T) { + input := "\n" + `{"type":"ok"}` + "\n" + r := bufio.NewReaderSize(strings.NewReader(input), 64*1024) + + line, skipped, err := ReadJSONLine(r) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if skipped { + t.Fatal("expected skipped=false") + } + if len(line) != 0 { + t.Fatalf("expected empty line, got %d bytes", len(line)) + } + + line, skipped, err = ReadJSONLine(r) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if skipped { + t.Fatal("expected skipped=false for second line") + } + if string(line) != `{"type":"ok"}` { + t.Fatalf("unexpected second line: %q", line) + } +} + +func TestReadJSONLine_MediumLine(t *testing.T) { + payload := strings.Repeat("x", 200*1024) // 200KB — exceeds 64KB buffer, requires multiple ReadLine chunks + input := payload + "\n" + r := bufio.NewReaderSize(strings.NewReader(input), 64*1024) + + line, skipped, err := ReadJSONLine(r) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if skipped { + t.Fatal("expected skipped=false for 200KB line") + } + if len(line) != 200*1024 { + t.Fatalf("expected 200KB, got %d bytes", len(line)) + } +} + +func TestReadJSONLine_OversizedLine(t *testing.T) { + saved := MaxLineSize + defer func() { + // MaxLineSize is a const; we test by embedding a smaller threshold. + // Since we can't reassign a const, we build a line that exceeds 50MB. + // Instead, we use a custom helper to keep the test fast. + _ = saved + }() + + // Build a line just over MaxLineSize. To keep allocations reasonable + // in tests, we use a custom reader that streams repeated bytes. + totalSize := MaxLineSize + 1024 // slightly over 50MB + src := &repeatingReader{char: 'A', remaining: totalSize} + // Append a newline + a short "next" line so we can verify recovery. + combined := io.MultiReader(src, strings.NewReader("\n{\"type\":\"ok\"}\n")) + r := bufio.NewReaderSize(combined, 64*1024) + + line, skipped, err := ReadJSONLine(r) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !skipped { + t.Fatal("expected skipped=true for oversized line") + } + if line != nil { + t.Fatalf("expected nil line when skipped, got %d bytes", len(line)) + } + + // Next line should be readable normally + line, skipped, err = ReadJSONLine(r) + if err != nil { + t.Fatalf("unexpected error reading next line: %v", err) + } + if skipped { + t.Fatal("expected skipped=false for recovery line") + } + if string(line) != `{"type":"ok"}` { + t.Fatalf("unexpected recovery line: %q", line) + } +} + +func TestReadJSONLine_EOF(t *testing.T) { + r := bufio.NewReaderSize(strings.NewReader(""), 64*1024) + + _, _, err := ReadJSONLine(r) + if err != io.EOF { + t.Fatalf("expected io.EOF, got %v", err) + } +} + +func TestReadJSONLine_EOFDuringDrain(t *testing.T) { + // Oversized line without trailing newline — EOF during drain + totalSize := MaxLineSize + 1024 + src := &repeatingReader{char: 'B', remaining: totalSize} + r := bufio.NewReaderSize(src, 64*1024) + + _, skipped, err := ReadJSONLine(r) + // During drain, ReadLine will eventually hit EOF + if err == nil && !skipped { + t.Fatal("expected either error or skip") + } + if err != nil && err != io.EOF { + t.Fatalf("expected io.EOF during drain, got %v", err) + } +} + +func TestReadJSONLine_MultiLineMixed(t *testing.T) { + // Line 1: normal short + // Line 2: oversized + // Line 3: normal short (verify recovery) + var buf bytes.Buffer + buf.WriteString(`{"type":"start"}`) + buf.WriteByte('\n') + + oversized := strings.Repeat("Z", MaxLineSize+100) + buf.WriteString(oversized) + buf.WriteByte('\n') + + buf.WriteString(`{"type":"end"}`) + buf.WriteByte('\n') + + r := bufio.NewReaderSize(&buf, 64*1024) + + // Line 1 + line, skipped, err := ReadJSONLine(r) + if err != nil { + t.Fatalf("line 1 error: %v", err) + } + if skipped { + t.Fatal("line 1 should not be skipped") + } + if string(line) != `{"type":"start"}` { + t.Fatalf("line 1 unexpected: %q", string(line)) + } + + // Line 2 (oversized) + line, skipped, err = ReadJSONLine(r) + if err != nil { + t.Fatalf("line 2 error: %v", err) + } + if !skipped { + t.Fatal("line 2 should be skipped") + } + if line != nil { + t.Fatal("line 2 should return nil") + } + + // Line 3 (recovery) + line, skipped, err = ReadJSONLine(r) + if err != nil { + t.Fatalf("line 3 error: %v", err) + } + if skipped { + t.Fatal("line 3 should not be skipped") + } + if string(line) != `{"type":"end"}` { + t.Fatalf("line 3 unexpected: %q", string(line)) + } +} + +func TestReadJSONLine_IOError(t *testing.T) { + expectedErr := fmt.Errorf("simulated IO failure") + r := bufio.NewReaderSize(&errorReader{err: expectedErr}, 64*1024) + + _, _, err := ReadJSONLine(r) + if err == nil { + t.Fatal("expected error") + } + if err.Error() != expectedErr.Error() { + t.Fatalf("expected %q, got %q", expectedErr, err) + } +} + +// repeatingReader streams a single character `remaining` times without a newline, +// then returns io.EOF. Used to test oversized lines without allocating huge buffers. +type repeatingReader struct { + char byte + remaining int +} + +func (r *repeatingReader) Read(p []byte) (int, error) { + if r.remaining <= 0 { + return 0, io.EOF + } + n := len(p) + if n > r.remaining { + n = r.remaining + } + for i := 0; i < n; i++ { + p[i] = r.char + } + r.remaining -= n + return n, nil +} + +// errorReader always returns the configured error. +type errorReader struct { + err error +} + +func (r *errorReader) Read(p []byte) (int, error) { + return 0, r.err +} diff --git a/sandbox/v2/box.go b/sandbox/v2/box.go index fc42a355..fec54f7f 100644 --- a/sandbox/v2/box.go +++ b/sandbox/v2/box.go @@ -2,7 +2,9 @@ package sandbox import ( "context" + "fmt" "io" + "strings" "sync/atomic" "time" @@ -121,11 +123,33 @@ func (b *Box) Stream(ctx context.Context, cmd []string, opts ...ExecOption) (*Ex return nil, err } - handle, err := res.Runtime.ExecStream(ctx, b.containerID, cmd, tairuntime.ExecOptions{ + execOpts := tairuntime.ExecOptions{ WorkDir: cfg.WorkDir, Env: cfg.Env, User: "sandbox", - }) + } + + handle, err := res.Runtime.ExecStream(ctx, b.containerID, cmd, execOpts) + if err != nil && strings.Contains(err.Error(), "500 Internal Server Error") { + select { + case <-ctx.Done(): + return nil, fmt.Errorf("exec create failed: %w (ctx cancelled, no retry)", err) + case <-time.After(2 * time.Second): + } + + status := b.inspectStatus(ctx) + if status != "running" { + return nil, fmt.Errorf("exec create failed: %w (container status: %s)", err, status) + } + + var retryErr error + handle, retryErr = res.Runtime.ExecStream(ctx, b.containerID, cmd, execOpts) + if retryErr != nil { + return nil, fmt.Errorf("exec create failed after retry: %w (original: %v)", retryErr, err) + } + err = nil + } + if err != nil { return nil, err } diff --git a/tai/runtime/docker_core.go b/tai/runtime/docker_core.go index e6601e4a..172372d7 100644 --- a/tai/runtime/docker_core.go +++ b/tai/runtime/docker_core.go @@ -7,6 +7,7 @@ import ( "io" "strconv" "strings" + "time" "github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/filters" @@ -185,6 +186,8 @@ func (d *dockerCore) execStream(ctx context.Context, id string, cmd []string, op Stdout: stdoutR, Stderr: stderrR, Wait: func() (int, error) { + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() for { inspect, err := d.cli.ContainerExecInspect(execCtx, execResp.ID) if err != nil { @@ -193,6 +196,11 @@ func (d *dockerCore) execStream(ctx context.Context, id string, cmd []string, op if !inspect.Running { return inspect.ExitCode, nil } + select { + case <-execCtx.Done(): + return -1, execCtx.Err() + case <-ticker.C: + } } }, Cancel: func() {