diff --git a/agent/sandbox/v2/claude/attachments.go b/agent/sandbox/v2/claude/attachments.go index 5e38438e..b08b5618 100644 --- a/agent/sandbox/v2/claude/attachments.go +++ b/agent/sandbox/v2/claude/attachments.go @@ -2,224 +2,19 @@ package claude import ( "context" - "fmt" - "path/filepath" - "strings" agentContext "github.com/yaoapp/yao/agent/context" - "github.com/yaoapp/yao/attachment" + "github.com/yaoapp/yao/agent/sandbox/v2/shared" workspace "github.com/yaoapp/yao/tai/workspace" ) // prepareAttachments resolves __yao.attachment:// URLs in messages, // copies actual files into the workspace .attachments/{chatID}/ directory via ws.Copy, // and replaces multimodal content parts with text references. +// +// Delegates to shared.PrepareAttachments; the returned text-replaced messages +// are used directly by the Claude CLI (which reads local files via text refs). func prepareAttachments(ctx context.Context, messages []agentContext.Message, chatID string, ws workspace.FS) ([]agentContext.Message, error) { - usedNames := make(map[string]int) - attachDir := ".attachments/" + chatID - - result := make([]agentContext.Message, len(messages)) - copy(result, messages) - - for i, msg := range result { - if msg.Role != "user" { - continue - } - - parts, ok := msg.Content.([]interface{}) - if !ok { - if typedParts, ok := msg.Content.([]agentContext.ContentPart); ok { - iparts := make([]interface{}, len(typedParts)) - for j, p := range typedParts { - m := map[string]interface{}{"type": string(p.Type)} - if p.Text != "" { - m["text"] = p.Text - } - if p.ImageURL != nil { - m["image_url"] = map[string]interface{}{ - "url": p.ImageURL.URL, - "detail": string(p.ImageURL.Detail), - } - } - if p.File != nil { - m["file"] = map[string]interface{}{ - "url": p.File.URL, - "filename": p.File.Filename, - } - } - iparts[j] = m - } - parts = iparts - } else { - continue - } - } - - if len(parts) == 0 { - continue - } - - var textParts []string - - for _, item := range parts { - m, ok := item.(map[string]interface{}) - if !ok { - continue - } - partType, _ := m["type"].(string) - - switch partType { - case "text": - if text, ok := m["text"].(string); ok && text != "" { - textParts = append(textParts, text) - } - - case "image_url": - imgData, _ := m["image_url"].(map[string]interface{}) - if imgData == nil { - continue - } - url, _ := imgData["url"].(string) - if url == "" { - continue - } - uploaderName, fileID, isWrapper := attachment.Parse(url) - if !isWrapper { - textParts = append(textParts, fmt.Sprintf("[Image: %s]", url)) - continue - } - ref, err := resolveAttachment(ctx, uploaderName, fileID, "", attachDir, usedNames, ws) - if err != nil { - textParts = append(textParts, "[Attached image: failed to load]") - continue - } - textParts = append(textParts, ref) - - case "file": - fileData, _ := m["file"].(map[string]interface{}) - if fileData == nil { - continue - } - url, _ := fileData["url"].(string) - hintName, _ := fileData["filename"].(string) - if url == "" { - continue - } - uploaderName, fileID, isWrapper := attachment.Parse(url) - if !isWrapper { - textParts = append(textParts, fmt.Sprintf("[File: %s]", url)) - continue - } - ref, err := resolveAttachment(ctx, uploaderName, fileID, hintName, attachDir, usedNames, ws) - if err != nil { - textParts = append(textParts, "[Attached file: failed to load]") - continue - } - textParts = append(textParts, ref) - } - } - - if len(textParts) > 0 { - newMsg := result[i] - newMsg.Content = strings.Join(textParts, "\n\n") - result[i] = newMsg - } - } - - return result, nil -} - -// resolveAttachment gets the local path of an attachment and copies it into -// the workspace via ws.Copy("local:///abs/path", ".attachments/{chatID}/filename"). -func resolveAttachment( - ctx context.Context, - uploaderName, fileID, hintName, attachDir string, - usedNames map[string]int, - ws workspace.FS, -) (string, error) { - manager, exists := attachment.Managers[uploaderName] - if !exists { - return "", fmt.Errorf("attachment manager not found: %s", uploaderName) - } - - fileInfo, err := manager.Info(ctx, fileID) - if err != nil { - return "", fmt.Errorf("failed to get file info: %w", err) - } - - absPath, _, err := manager.LocalPath(ctx, fileID) - if err != nil { - return "", fmt.Errorf("failed to get local path: %w", err) - } - - filename := fileInfo.Filename - if filename == "" && hintName != "" { - filename = hintName - } - if filename == "" { - ext := extensionFromContentType(fileInfo.ContentType) - filename = fileID + ext - } - - baseName := filename - if count, exists := usedNames[baseName]; exists { - ext := filepath.Ext(filename) - name := strings.TrimSuffix(filename, ext) - filename = fmt.Sprintf("%s_%d%s", name, count+1, ext) - usedNames[baseName] = count + 1 - } else { - usedNames[baseName] = 0 - } - - dstPath := attachDir + "/" + filename - src := "local:///" + absPath - - if _, err := ws.Copy(src, dstPath); err != nil { - return "", fmt.Errorf("failed to copy attachment to workspace: %w", err) - } - - sizeStr := formatFileSize(fileInfo.Bytes) - return fmt.Sprintf("[Attached file: %s (%s, %s)]", dstPath, fileInfo.ContentType, sizeStr), nil -} - -func extensionFromContentType(contentType string) string { - switch contentType { - case "image/png": - return ".png" - case "image/jpeg": - return ".jpg" - case "image/gif": - return ".gif" - case "image/webp": - return ".webp" - case "image/svg+xml": - return ".svg" - case "application/pdf": - return ".pdf" - case "text/plain": - return ".txt" - case "text/html": - return ".html" - case "text/css": - return ".css" - case "text/javascript", "application/javascript": - return ".js" - case "application/json": - return ".json" - case "application/zip": - return ".zip" - default: - return "" - } -} - -func formatFileSize(bytes int) string { - switch { - case bytes >= 1024*1024: - return fmt.Sprintf("%.1fMB", float64(bytes)/(1024*1024)) - case bytes >= 1024: - return fmt.Sprintf("%.1fKB", float64(bytes)/1024) - default: - return fmt.Sprintf("%dB", bytes) - } + processed, _, err := shared.PrepareAttachments(ctx, messages, chatID, ws) + return processed, err } diff --git a/agent/sandbox/v2/init.go b/agent/sandbox/v2/init.go index cd13d530..4e9aaee0 100644 --- a/agent/sandbox/v2/init.go +++ b/agent/sandbox/v2/init.go @@ -2,6 +2,7 @@ package sandboxv2 import ( "github.com/yaoapp/yao/agent/sandbox/v2/claude" + "github.com/yaoapp/yao/agent/sandbox/v2/opencode" "github.com/yaoapp/yao/agent/sandbox/v2/types" yaorunner "github.com/yaoapp/yao/agent/sandbox/v2/yao" ) @@ -9,5 +10,7 @@ import ( func init() { Register("claude", func() types.Runner { return claude.New() }) Register("claude/cli", func() types.Runner { return claude.New() }) + Register("opencode", func() types.Runner { return opencode.New() }) + Register("opencode/cli", func() types.Runner { return opencode.New() }) Register("yao", func() types.Runner { return yaorunner.New() }) } diff --git a/agent/sandbox/v2/opencode/command.go b/agent/sandbox/v2/opencode/command.go new file mode 100644 index 00000000..63f600bd --- /dev/null +++ b/agent/sandbox/v2/opencode/command.go @@ -0,0 +1,374 @@ +package opencode + +import ( + "fmt" + "regexp" + "strings" + "time" + + "github.com/google/uuid" + "github.com/yaoapp/gou/connector" + "github.com/yaoapp/gou/store" + "github.com/yaoapp/kun/str" + agentContext "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/sandbox/v2/types" +) + +var ( + yaoSessionNS = uuid.MustParse("e37bc21a-72dd-4a8f-b567-1f02c3d4e590") + safeNameRe = regexp.MustCompile(`[^a-zA-Z0-9_\-.]`) +) + +type command struct { + shell []string + env map[string]string + stdin string + workDir string +} + +func chatIDToSessionID(assistantID, chatID string) string { + return uuid.NewSHA1(yaoSessionNS, []byte(assistantID+":"+chatID)).String() +} + +func sanitizeSessionName(chatID string) string { + return "yao-oc-" + safeNameRe.ReplaceAllString(chatID, "_") +} + +func chatSessionExists(storeKey string) bool { + s, err := store.Get("__yao.store") + if err != nil { + return false + } + return s.Has(storeKey) +} + +func markChatSession(storeKey, sessionID string, ttl time.Duration) { + s, err := store.Get("__yao.store") + if err != nil { + return + } + s.Set(storeKey, sessionID, ttl) +} + +func (r *Runner) buildCommand(req *types.StreamRequest, p platform, attachmentPaths []string) command { + workDir := req.Computer.GetWorkDir() + assistantID := req.AssistantID + chatID := req.ChatID + + var isContinuation bool + if chatID != "" { + storeKey := "opencode-session:" + assistantID + ":" + chatID + isContinuation = chatSessionExists(storeKey) + } + + env := buildEnv(req, p) + args := buildArgs(req, r, isContinuation, chatID) + + stdinMsg := buildStdinMessage(req.Messages, attachmentPaths) + + script := shellQuote("opencode", args...) + + return command{ + shell: p.ShellCmd(script), + env: env, + stdin: stdinMsg, + workDir: workDir, + } +} + +func buildEnv(req *types.StreamRequest, p platform) map[string]string { + env := make(map[string]string) + workDir := req.Computer.GetWorkDir() + + ws := req.Computer.Workplace() + if ws != nil { + workspaceID, err := ws.GetID() + if err == nil { + env["CTX_WORKSPACE_ID"] = workspaceID + } + } + + for k, v := range p.HomeEnv(workDir) { + env[k] = v + } + env["WORKDIR"] = workDir + + assistantID := req.AssistantID + prefix := p.PathJoin(workDir, ".yao", "assistants", assistantID, "opencode") + if assistantID == "" { + prefix = p.PathJoin(workDir, ".opencode-data") + } + if assistantID != "" { + env["CTX_ASSISTANT_ID"] = assistantID + env["CTX_SKILLS_DIR"] = p.PathJoin(workDir, ".yao", "assistants", assistantID, "skills") + } + + env["OPENCODE_DATA_DIR"] = p.PathJoin(prefix, "data") + env["OPENCODE_CACHE_DIR"] = p.PathJoin(prefix, "cache") + env["OPENCODE_STATE_DIR"] = p.PathJoin(prefix, "state") + env["OPENCODE_CONFIG_DIR"] = p.PathJoin(prefix, "config") + + env["OPENCODE_DISABLE_AUTOUPDATE"] = "true" + env["OPENCODE_DISABLE_MODELS_FETCH"] = "true" + env["OPENCODE_DISABLE_LSP_DOWNLOAD"] = "true" + env["OPENCODE_DISABLE_DEFAULT_PLUGINS"] = "true" + env["OPENCODE_DISABLE_TERMINAL_TITLE"] = "true" + env["OPENCODE_DISABLE_MOUSE"] = "true" + env["OPENCODE_DISABLE_CLAUDE_CODE"] = "true" + env["OPENCODE_CLIENT"] = "cli" + + // Lower bash default timeout from 120s to 30s. Long-running commands + // like browsers should be nohup'd; this prevents accidental 2-min hangs. + env["OPENCODE_EXPERIMENTAL_BASH_DEFAULT_TIMEOUT_MS"] = "30000" + + if req.Connector != nil { + setting := req.Connector.Setting() + key, _ := setting["key"].(string) + if key != "" { + env["YAO_PROVIDER_KEY"] = key + } + + if req.Connector.Is(connector.ANTHROPIC) { + apiKey, _ := setting["key"].(string) + if apiKey != "" { + env["ANTHROPIC_API_KEY"] = apiKey + } + } + } + + injectRoleEnvVars(env, req) + + if req.Config != nil && len(req.Config.Secrets) > 0 { + for k, v := range req.Config.Secrets { + env[k] = str.EnvVar(v) + } + } + + if req.Token != nil { + if req.Token.Token != "" { + env["YAO_TOKEN"] = req.Token.Token + } + if req.Token.RefreshToken != "" { + env["YAO_REFRESH_TOKEN"] = req.Token.RefreshToken + } + } + + return env +} + +func buildArgs(req *types.StreamRequest, r *Runner, isContinuation bool, chatID string) []string { + args := []string{"run", "--format", "json"} + + permMode := "" + if req.Config != nil && req.Config.Runner.Options != nil { + if v, ok := req.Config.Runner.Options["permission_mode"]; ok { + permMode = fmt.Sprintf("%v", v) + } + } + if permMode == "bypassPermissions" { + args = append(args, "--dangerously-skip-permissions") + } + + if chatID != "" && isContinuation { + sessionID := chatIDToSessionID(req.AssistantID, chatID) + args = append(args, "--continue", "--session", sessionID) + } + + if req.Connector != nil { + if mid := connectorModelID(req.Connector); mid != "" { + args = append(args, "--model", mid) + } + } + + // User message and attachments are passed via stdin (heredoc pipe), + // NOT as positional args. This avoids shell escaping issues with + // special characters, CJK text, long messages, and --file ambiguity. + + return args +} + +// buildStdinMessage builds the text piped to `opencode run` via stdin. +// It combines the user's text message with attachment references so OpenCode +// receives everything through stdin — no positional args, no --file flags. +// This mirrors the Claude runner approach and avoids shell escaping pitfalls. +func buildStdinMessage(messages []agentContext.Message, attachmentPaths []string) string { + var parts []string + + if len(attachmentPaths) > 0 { + parts = append(parts, "The user has attached the following files — read them to understand context:") + for _, p := range attachmentPaths { + parts = append(parts, fmt.Sprintf(" - %s", p)) + } + parts = append(parts, "") + } + + text := lastUserText(messages) + if text != "" { + parts = append(parts, text) + } + + return strings.Join(parts, "\n") +} + +// lastUserText extracts the plain text from the last user message, +// handling string, []ContentPart, and []any (generic JSON) content types. +func lastUserText(messages []agentContext.Message) string { + for i := len(messages) - 1; i >= 0; i-- { + if messages[i].Role != "user" { + continue + } + switch c := messages[i].Content.(type) { + case string: + return c + case []agentContext.ContentPart: + var texts []string + for _, part := range c { + if part.Type == agentContext.ContentText && part.Text != "" { + texts = append(texts, part.Text) + } + } + return strings.Join(texts, "\n") + case []any: + var texts []string + for _, item := range c { + if m, ok := item.(map[string]any); ok { + if t, _ := m["type"].(string); t == "text" { + if text, _ := m["text"].(string); text != "" { + texts = append(texts, text) + } + } + } + } + return strings.Join(texts, "\n") + } + } + return "" +} + +func buildSandboxEnvPrompt(p platform, workDir string) string { + osName := p.OS() + if osName == "" { + osName = "linux" + } + shell := p.Shell() + if shell == "" { + shell = "bash" + } + + return fmt.Sprintf(`## Sandbox Environment + +- **Operating System**: %[2]s +- **Shell**: %[3]s +- **Working Directory**: %[1]s +- **File Access**: You have full read/write access to %[1]s + +## User Attachments + +User-uploaded files are placed in %[1]s/.attachments/{chatID}/ +Each chat session has its own subdirectory. +When the user attaches files, their paths are listed at the top of the message. +**Read these files yourself** using the Read or Bash tool — they are NOT passed as CLI arguments. +`, workDir, osName, shell) +} + +func getProviderPrefix(conn connector.Connector) string { + if conn != nil && conn.Is(connector.ANTHROPIC) { + return "anthropic" + } + return "openai" +} + +// resolveRoleConnector determines which connector to use for a given role. +func resolveRoleConnector( + role string, + roleConnectors map[string]*types.RoleConnector, + userExplicit bool, + getConnector func(id string) connector.Connector, +) connector.Connector { + rc, ok := roleConnectors[role] + if !ok || rc == nil { + return nil + } + if rc.Override == "user" && userExplicit { + return nil + } + return getConnector(rc.Connector) +} + +func getRoleConnectors(req *types.StreamRequest) map[string]*types.RoleConnector { + if req.Config == nil { + return nil + } + return req.Config.Runner.Connectors +} + +// shellQuote builds a shell-safe command string from program and args. +func shellQuote(program string, args ...string) string { + parts := make([]string, 0, 1+len(args)) + parts = append(parts, program) + for _, a := range args { + if a == "" || strings.ContainsAny(a, " \t\n\"'\\$`!#&|;(){}[]<>?*~") { + parts = append(parts, "'"+strings.ReplaceAll(a, "'", `'\''`)+"'") + } else { + parts = append(parts, a) + } + } + return strings.Join(parts, " ") +} + +// connectorModelID returns the "provider/model" string matching the +// provider ID used in opencode.json (see buildProviderConfig). +func connectorModelID(c connector.Connector) string { + setting := c.Setting() + modelName, _ := setting["model"].(string) + host, _ := setting["host"].(string) + + if c.Is(connector.ANTHROPIC) { + return "anthropic/" + modelName + } + if host == "" || isNativeOpenAI(host) { + return "openai/" + modelName + } + return "custom/" + modelName +} + +// injectRoleEnvVars adds API key, base URL, and model environment variables +// for each role connector defined in openCodeRoleMap. These env vars are +// consumed by opencode.json provider blocks (via {env:...} references) and +// by the custom read.ts tool (for vision API calls). +func injectRoleEnvVars(env map[string]string, req *types.StreamRequest) { + if req.Config == nil || req.Config.Runner.Connectors == nil { + return + } + for role, spec := range openCodeRoleMap { + if spec.EnvKeyPrefix == "" { + continue + } + rc, ok := req.Config.Runner.Connectors[role] + if !ok || rc == nil || rc.Connector == "" { + continue + } + c, exists := connector.Connectors[rc.Connector] + if !exists || c == nil { + continue + } + setting := c.Setting() + if key, _ := setting["key"].(string); key != "" { + env[spec.EnvKeyPrefix+"_KEY"] = key + } + if host, _ := setting["host"].(string); host != "" { + env[spec.EnvKeyPrefix+"_BASE_URL"] = normalizeBaseURL(host) + } + if model, _ := setting["model"].(string); model != "" { + env[spec.EnvKeyPrefix+"_MODEL"] = model + } + } +} + +func connectorHost(c connector.Connector) string { + if c == nil { + return "" + } + host, _ := c.Setting()["host"].(string) + return strings.TrimSpace(host) +} diff --git a/agent/sandbox/v2/opencode/command_test.go b/agent/sandbox/v2/opencode/command_test.go new file mode 100644 index 00000000..763bdd9d --- /dev/null +++ b/agent/sandbox/v2/opencode/command_test.go @@ -0,0 +1,145 @@ +package opencode + +import ( + "testing" + + agentContext "github.com/yaoapp/yao/agent/context" +) + +func TestChatIDToSessionID(t *testing.T) { + id1 := chatIDToSessionID("assistant-1", "chat-1") + id2 := chatIDToSessionID("assistant-1", "chat-1") + id3 := chatIDToSessionID("assistant-1", "chat-2") + + if id1 != id2 { + t.Error("same inputs should produce same session ID") + } + if id1 == id3 { + t.Error("different chatIDs should produce different session IDs") + } + if id1 == "" { + t.Error("session ID should not be empty") + } +} + +func TestSanitizeSessionName(t *testing.T) { + cases := []struct { + input, want string + }{ + {"simple-chat", "yao-oc-simple-chat"}, + {"chat with spaces", "yao-oc-chat_with_spaces"}, + {"chat/with/slashes", "yao-oc-chat_with_slashes"}, + {"chat@special#chars", "yao-oc-chat_special_chars"}, + } + for _, tc := range cases { + got := sanitizeSessionName(tc.input) + if got != tc.want { + t.Errorf("sanitizeSessionName(%q) = %q, want %q", tc.input, got, tc.want) + } + } +} + +func TestLastUserText(t *testing.T) { + cases := []struct { + name string + messages []agentContext.Message + want string + }{ + { + name: "empty", + messages: nil, + want: "", + }, + { + name: "single user message", + messages: []agentContext.Message{ + {Role: "user", Content: "hello"}, + }, + want: "hello", + }, + { + name: "last user wins", + messages: []agentContext.Message{ + {Role: "user", Content: "first"}, + {Role: "assistant", Content: "reply"}, + {Role: "user", Content: "second"}, + }, + want: "second", + }, + { + name: "no user messages", + messages: []agentContext.Message{ + {Role: "assistant", Content: "only assistant"}, + }, + want: "", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := lastUserText(tc.messages) + if got != tc.want { + t.Errorf("lastUserText() = %q, want %q", got, tc.want) + } + }) + } +} + +func TestBuildStdinMessage(t *testing.T) { + msgs := []agentContext.Message{ + {Role: "user", Content: "把这个会议纪要的关键内容提取出来"}, + } + + t.Run("no attachments", func(t *testing.T) { + got := buildStdinMessage(msgs, nil) + if got != "把这个会议纪要的关键内容提取出来" { + t.Errorf("unexpected: %q", got) + } + }) + + t.Run("with attachments", func(t *testing.T) { + got := buildStdinMessage(msgs, []string{"/workspace/.attachments/abc/test.txt"}) + if !strContains(got, "/workspace/.attachments/abc/test.txt") { + t.Error("should contain attachment path") + } + if !strContains(got, "把这个会议纪要的关键内容提取出来") { + t.Error("should contain user message") + } + }) + + t.Run("empty message", func(t *testing.T) { + got := buildStdinMessage(nil, []string{"/workspace/file.txt"}) + if !strContains(got, "/workspace/file.txt") { + t.Error("should contain attachment path even without message") + } + }) +} + +func TestBuildSandboxEnvPrompt(t *testing.T) { + p := &posixBase{os: "linux", shell: "bash"} + prompt := buildSandboxEnvPrompt(p, "/workspace") + if prompt == "" { + t.Error("prompt should not be empty") + } + if !strContains(prompt, "/workspace") { + t.Error("prompt should mention workspace path") + } + if !strContains(prompt, "linux") { + t.Error("prompt should mention OS") + } +} + +func TestGetProviderPrefix(t *testing.T) { + if p := getProviderPrefix(nil); p != "openai" { + t.Errorf("nil connector should give openai, got %s", p) + } +} + +func strContains(s, sub string) bool { + for i := 0; i <= len(s)-len(sub); i++ { + if s[i:i+len(sub)] == sub { + return true + } + } + return false +} diff --git a/agent/sandbox/v2/opencode/config.go b/agent/sandbox/v2/opencode/config.go new file mode 100644 index 00000000..06964754 --- /dev/null +++ b/agent/sandbox/v2/opencode/config.go @@ -0,0 +1,364 @@ +package opencode + +import ( + "encoding/json" + "strings" + + "github.com/yaoapp/gou/connector" + "github.com/yaoapp/yao/agent/sandbox/v2/types" +) + +type roleSpec struct { + EnvKeyPrefix string + TopLevel string + Modalities map[string][]string +} + +var openCodeRoleMap = map[string]roleSpec{ + "light": { + EnvKeyPrefix: "YAO_LIGHT", + TopLevel: "small_model", + }, + "vision": { + EnvKeyPrefix: "YAO_VISION", + Modalities: map[string][]string{ + "input": {"text", "image"}, + "output": {"text"}, + }, + }, + "heavy": { + EnvKeyPrefix: "YAO_HEAVY", + }, + "subagent": { + EnvKeyPrefix: "YAO_SUBAGENT", + }, +} + +// buildOpenCodeConfig generates the opencode.json project configuration. +// All provider configuration is direct (no a2o proxy). +func buildOpenCodeConfig(req *types.PrepareRequest, mcpServers []types.MCPServer) []byte { + cfg := map[string]any{ + "$schema": "https://opencode.ai/config.json", + "autoupdate": false, + "snapshot": false, + "share": "disabled", + "watcher": map[string]any{"ignore": []string{".yao/**", ".attachments/**"}}, + "permission": map[string]any{"*": "allow"}, + } + + if req.Connector != nil { + providerID, providerCfg, modelStr := buildProviderConfig(req.Connector) + cfg["provider"] = map[string]any{providerID: providerCfg} + cfg["model"] = modelStr + cfg["enabled_providers"] = []string{providerID} + } + + injectRoleProviders(cfg, req) + + if len(mcpServers) > 0 { + cfg["mcp"] = buildMCPConfig(mcpServers) + } + + prefix := ".yao/assistants/" + req.AssistantID + if req.AssistantID == "" { + prefix = ".opencode" + } + cfg["instructions"] = []string{prefix + "/system-prompt.md"} + + data, _ := json.MarshalIndent(cfg, "", " ") + return data +} + +// buildProviderConfig maps a Yao connector to an OpenCode provider configuration. +// Anthropic connectors map directly; OpenAI/OpenAI-compatible map to "openai". +// +// OpenCode appends its own endpoint paths (e.g. /responses) to baseURL, +// so we must NOT include /chat/completions. For native OpenAI (api.openai.com) +// we omit baseURL entirely and let OpenCode use its built-in default. +// For custom hosts (OpenAI-compatible proxies), we pass the bare host URL. +func buildProviderConfig(conn connector.Connector) (providerID string, cfg map[string]any, model string) { + setting := conn.Setting() + host, _ := setting["host"].(string) + modelName, _ := setting["model"].(string) + + opts := map[string]any{ + "apiKey": "{env:YAO_PROVIDER_KEY}", + } + + if conn.Is(connector.ANTHROPIC) { + if host != "" { + opts["baseURL"] = host + } + return "anthropic", map[string]any{"options": opts}, "anthropic/" + modelName + } + + // Native OpenAI (api.openai.com): use built-in "openai" provider which + // already knows all official models — no models declaration needed. + if host == "" || isNativeOpenAI(host) { + return "openai", map[string]any{"options": opts}, "openai/" + modelName + } + + // OpenAI-compatible provider (DeepSeek, Moonshot, etc.): must use + // @ai-sdk/openai-compatible and explicitly declare models, otherwise + // OpenCode throws ProviderModelNotFoundError. + opts["baseURL"] = normalizeBaseURL(host) + + modelCfg := map[string]any{ + "name": modelName, + } + + // DeepSeek (and similar) thinking models return reasoning_content in + // assistant messages. OpenCode must be told to preserve and replay + // this field on conversation continuation, otherwise the API returns: + // "The reasoning_content in the thinking mode must be passed back to the API." + // Adding "interleaved" is safe for non-thinking models (no-op if absent). + modelCfg["interleaved"] = map[string]any{"field": "reasoning_content"} + + // Pass through thinking configuration from the Yao connector so OpenCode + // sends it to the upstream API. DeepSeek defaults thinking to "enabled"; + // without explicitly sending {"thinking":{"type":"disabled"}}, the API + // returns reasoning_content that OpenCode (AI SDK bug) fails to replay. + modelOpts := buildModelOptions(setting) + if len(modelOpts) > 0 { + modelCfg["options"] = modelOpts + } + + return "custom", map[string]any{ + "npm": "@ai-sdk/openai-compatible", + "options": opts, + "models": map[string]any{ + modelName: modelCfg, + }, + }, "custom/" + modelName +} + +// buildModelOptions extracts connector-level model options (thinking, etc.) +// and maps them to the OpenCode model options format. +func buildModelOptions(setting map[string]any) map[string]any { + opts := map[string]any{} + + // Forward thinking configuration as-is (e.g. {"type":"disabled"}). + // DeepSeek V4 models default thinking to "enabled"; the only way to + // suppress reasoning_content is to explicitly send {"type":"disabled"}. + if thinking, ok := setting["thinking"]; ok && thinking != nil { + opts["thinking"] = thinking + } + + return opts +} + +// isNativeOpenAI returns true if host points to official OpenAI API, +// where OpenCode already knows the correct base URL. +func isNativeOpenAI(host string) bool { + h := strings.TrimRight(strings.TrimPrefix(strings.TrimPrefix(host, "https://"), "http://"), "/") + return h == "api.openai.com" || + strings.HasPrefix(h, "api.openai.com/") +} + +// normalizeBaseURL strips trailing /chat/completions or /v1/chat/completions +// that Yao connectors may include, because OpenCode appends its own paths. +func normalizeBaseURL(host string) string { + u := strings.TrimRight(host, "/") + for _, suffix := range []string{"/chat/completions", "/completions"} { + if strings.HasSuffix(u, suffix) { + u = strings.TrimSuffix(u, suffix) + break + } + } + return strings.TrimRight(u, "/") +} + +// injectRoleProviders iterates openCodeRoleMap and injects provider blocks +// for every role that has a configured connector. For the "light" role it +// also sets the top-level "small_model" field. This replaces the old +// buildSmallModel function and adds support for vision/heavy/subagent roles. +func injectRoleProviders(cfg map[string]any, req *types.PrepareRequest) { + if req.Config == nil || req.Config.Runner.Connectors == nil { + return + } + + providers, _ := cfg["provider"].(map[string]any) + if providers == nil { + providers = map[string]any{} + cfg["provider"] = providers + } + + enabledSlice, _ := cfg["enabled_providers"].([]string) + enabledSet := map[string]bool{} + for _, e := range enabledSlice { + enabledSet[e] = true + } + + primaryHost := "" + primaryType := "" + if req.Connector != nil { + primaryHost = connectorHost(req.Connector) + if req.Connector.Is(connector.ANTHROPIC) { + primaryType = "anthropic" + } else { + primaryType = "openai" + } + } + + for role, spec := range openCodeRoleMap { + rc, ok := req.Config.Runner.Connectors[role] + if !ok || rc == nil || rc.Connector == "" { + continue + } + + c, exists := connector.Connectors[rc.Connector] + if !exists || c == nil { + continue + } + + setting := c.Setting() + modelName, _ := setting["model"].(string) + if modelName == "" { + continue + } + + roleHost := connectorHost(c) + roleType := "openai" + if c.Is(connector.ANTHROPIC) { + roleType = "anthropic" + } + + sameProvider := roleType == primaryType && roleHost == primaryHost + if sameProvider && primaryHost != "" { + sameProvider = true + } else if sameProvider && primaryHost == "" && roleHost == "" { + sameProvider = true + } else if roleHost != primaryHost { + sameProvider = false + } + + var providerID string + var modelRef string + + if sameProvider { + providerID = resolveExistingProviderID(providers, primaryType) + modelRef = providerID + "/" + modelName + mergeModelIntoProvider(providers, providerID, modelName, spec.Modalities) + } else { + providerID = role + providerCfg := buildRoleProviderConfig(c, spec.EnvKeyPrefix, spec.Modalities) + providers[providerID] = providerCfg + modelRef = providerID + "/" + modelName + } + + if !enabledSet[providerID] { + enabledSlice = append(enabledSlice, providerID) + enabledSet[providerID] = true + } + + if spec.TopLevel != "" { + cfg[spec.TopLevel] = modelRef + } + } + + cfg["enabled_providers"] = enabledSlice +} + +// resolveExistingProviderID finds the actual provider ID key used in the +// providers map for a given type. For "openai" type, it could be "openai" +// or "custom" (for openai-compatible). Returns the type as fallback. +func resolveExistingProviderID(providers map[string]any, pType string) string { + if _, ok := providers[pType]; ok { + return pType + } + if pType == "openai" { + if _, ok := providers["custom"]; ok { + return "custom" + } + } + return pType +} + +// mergeModelIntoProvider adds a model entry to an existing provider block. +func mergeModelIntoProvider(providers map[string]any, providerID, modelName string, modalities map[string][]string) { + block, ok := providers[providerID].(map[string]any) + if !ok { + return + } + models, _ := block["models"].(map[string]any) + if models == nil { + models = map[string]any{} + block["models"] = models + } + modelCfg := map[string]any{"name": modelName} + if len(modalities) > 0 { + modelCfg["modalities"] = modalities + } + models[modelName] = modelCfg +} + +// buildRoleProviderConfig creates a provider configuration block for a +// non-primary role connector. Uses the role's env key prefix for API key +// and base URL references. +func buildRoleProviderConfig(conn connector.Connector, envKeyPrefix string, modalities map[string][]string) map[string]any { + setting := conn.Setting() + modelName, _ := setting["model"].(string) + host, _ := setting["host"].(string) + + opts := map[string]any{ + "apiKey": "{env:" + envKeyPrefix + "_KEY}", + } + + modelCfg := map[string]any{"name": modelName} + if len(modalities) > 0 { + modelCfg["modalities"] = modalities + } + + if conn.Is(connector.ANTHROPIC) { + if host != "" { + opts["baseURL"] = host + } + return map[string]any{ + "options": opts, + "models": map[string]any{modelName: modelCfg}, + } + } + + if host == "" || isNativeOpenAI(host) { + return map[string]any{ + "options": opts, + "models": map[string]any{modelName: modelCfg}, + } + } + + opts["baseURL"] = normalizeBaseURL(host) + modelCfg["interleaved"] = map[string]any{"field": "reasoning_content"} + + return map[string]any{ + "npm": "@ai-sdk/openai-compatible", + "options": opts, + "models": map[string]any{modelName: modelCfg}, + } +} + +// buildMCPConfig produces the "mcp" object for opencode.json. +// OpenCode uses "command" as an array (not command + args like Claude). +func buildMCPConfig(servers []types.MCPServer) map[string]any { + result := make(map[string]any, len(servers)) + for _, s := range servers { + name := s.ServerID + if name == "" { + continue + } + result[name] = map[string]any{ + "type": "local", + "command": []string{"tai", "mcp", name}, + "enabled": true, + "environment": map[string]string{"YAO_TOKEN": "{env:YAO_TOKEN}"}, + } + } + if len(result) == 0 { + result["yao"] = map[string]any{ + "type": "local", + "command": []string{"tai", "mcp"}, + "enabled": true, + "environment": map[string]string{"YAO_TOKEN": "{env:YAO_TOKEN}"}, + } + } + return result +} diff --git a/agent/sandbox/v2/opencode/config_test.go b/agent/sandbox/v2/opencode/config_test.go new file mode 100644 index 00000000..39ec1f04 --- /dev/null +++ b/agent/sandbox/v2/opencode/config_test.go @@ -0,0 +1,125 @@ +package opencode + +import ( + "encoding/json" + "testing" + + "github.com/yaoapp/yao/agent/sandbox/v2/types" +) + +func TestBuildOpenCodeConfig_Defaults(t *testing.T) { + req := &types.PrepareRequest{ + AssistantID: "test-assistant", + Config: &types.SandboxConfig{}, + } + + data := buildOpenCodeConfig(req, nil) + var cfg map[string]any + if err := json.Unmarshal(data, &cfg); err != nil { + t.Fatalf("invalid JSON: %v", err) + } + + if cfg["autoupdate"] != false { + t.Errorf("autoupdate should be false, got %v", cfg["autoupdate"]) + } + if cfg["snapshot"] != false { + t.Errorf("snapshot should be false, got %v", cfg["snapshot"]) + } + if cfg["share"] != "disabled" { + t.Errorf("share should be disabled, got %v", cfg["share"]) + } + + instructions, ok := cfg["instructions"].([]any) + if !ok || len(instructions) == 0 { + t.Fatal("instructions should be a non-empty array") + } + if instructions[0] != ".yao/assistants/test-assistant/system-prompt.md" { + t.Errorf("instructions[0] = %q, want .yao/assistants/test-assistant/system-prompt.md", instructions[0]) + } + + watcher, ok := cfg["watcher"].(map[string]any) + if !ok { + t.Fatal("watcher should be a map") + } + ignore, ok := watcher["ignore"].([]any) + if !ok || len(ignore) < 2 { + t.Errorf("watcher.ignore should have at least 2 entries, got %v", ignore) + } +} + +func TestBuildOpenCodeConfig_NoAssistantID(t *testing.T) { + req := &types.PrepareRequest{ + AssistantID: "", + Config: &types.SandboxConfig{}, + } + + data := buildOpenCodeConfig(req, nil) + var cfg map[string]any + if err := json.Unmarshal(data, &cfg); err != nil { + t.Fatalf("invalid JSON: %v", err) + } + + instructions := cfg["instructions"].([]any) + if instructions[0] != ".opencode/system-prompt.md" { + t.Errorf("instructions[0] = %q, want .opencode/system-prompt.md", instructions[0]) + } +} + +func TestBuildOpenCodeConfig_WithMCP(t *testing.T) { + req := &types.PrepareRequest{ + AssistantID: "test", + Config: &types.SandboxConfig{}, + } + servers := []types.MCPServer{ + {ServerID: "my-server"}, + {ServerID: "another-server"}, + } + + data := buildOpenCodeConfig(req, servers) + var cfg map[string]any + if err := json.Unmarshal(data, &cfg); err != nil { + t.Fatalf("invalid JSON: %v", err) + } + + mcp, ok := cfg["mcp"].(map[string]any) + if !ok { + t.Fatal("mcp should be a map") + } + if _, exists := mcp["my-server"]; !exists { + t.Error("mcp should contain my-server") + } + if _, exists := mcp["another-server"]; !exists { + t.Error("mcp should contain another-server") + } +} + +func TestBuildMCPConfig_Default(t *testing.T) { + result := buildMCPConfig(nil) + if _, ok := result["yao"]; !ok { + t.Error("empty server list should produce default 'yao' entry") + } +} + +func TestBuildMCPConfig_WithServers(t *testing.T) { + servers := []types.MCPServer{ + {ServerID: "server-a"}, + {ServerID: "server-b"}, + {ServerID: ""}, + } + result := buildMCPConfig(servers) + if _, ok := result["server-a"]; !ok { + t.Error("should contain server-a") + } + if _, ok := result["server-b"]; !ok { + t.Error("should contain server-b") + } + if len(result) != 2 { + t.Errorf("should only have 2 entries (empty ID skipped), got %d", len(result)) + } + + serverA := result["server-a"].(map[string]any) + cmd := serverA["command"].([]string) + if len(cmd) != 3 || cmd[0] != "tai" || cmd[1] != "mcp" || cmd[2] != "server-a" { + t.Errorf("command should be [tai mcp server-a], got %v", cmd) + } +} diff --git a/agent/sandbox/v2/opencode/parse.go b/agent/sandbox/v2/opencode/parse.go new file mode 100644 index 00000000..3060b8fd --- /dev/null +++ b/agent/sandbox/v2/opencode/parse.go @@ -0,0 +1,533 @@ +package opencode + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "strings" + "time" + + "github.com/yaoapp/kun/log" + "github.com/yaoapp/yao/agent/output/message" +) + +// streamParser handles OpenCode's JSONL output (--format json). +// +// OpenCode emits one JSON object per line with a "type" field. Unlike Claude's +// streaming events which provide incremental deltas, OpenCode events arrive at +// completion boundaries: +// +// step_start – agent step begins (contains sessionID, part metadata) +// text – completed text block (full text, only emitted when part.time.end is set) +// tool_use – completed tool call (status: completed|error, contains full input/output) +// step_finish – step ended (reason: stop|tool-calls; includes token/cost info) +// reasoning – thinking/reasoning output +// error – error event +// +// Because tool_use events only arrive after the tool finishes, the parser +// emits a "running" execute chunk at step_start so the frontend has immediate +// feedback that work is in progress. It also tracks intermediate tool_use +// events (status=completed) so they are shown as soon as they arrive, even +// before step_finish. +type streamParser struct { + handler message.StreamFunc + completed bool + toolIndex int + + // textActive tracks whether a text message group is currently open. + textActive bool + textMsgID string + + // pendingExec tracks tools that were announced at step_start but haven't + // received their completed tool_use event yet. Key = step ID. + pendingExec map[string]string // stepID -> msgID +} + +func newStreamParser(handler message.StreamFunc) *streamParser { + return &streamParser{ + handler: handler, + pendingExec: make(map[string]string), + } +} + +func (p *streamParser) parse(ctx context.Context, stdout io.ReadCloser) error { + doneParsing := make(chan struct{}) + defer close(doneParsing) + go func() { + select { + case <-ctx.Done(): + stdout.Close() + case <-doneParsing: + } + }() + + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 0, 64*1024), 10*1024*1024) + + startTime := time.Now() + lineCount := 0 + lastHeartbeat := time.Now() + + log.Trace("[opencode-parse] stream started") + + for scanner.Scan() { + line := scanner.Text() + if line == "" { + continue + } + lineCount++ + + if time.Since(lastHeartbeat) > 30*time.Second { + log.Trace("[opencode-parse] heartbeat: lines=%d elapsed=%v", + lineCount, time.Since(startTime).Round(time.Second)) + lastHeartbeat = time.Now() + } + + var msg map[string]any + if err := json.Unmarshal([]byte(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) + } + continue + } + + msgType, _ := msg["type"].(string) + + switch msgType { + case "step_start": + if p.handleStepStart(msg) { + return nil + } + case "text": + if p.handleText(msg) { + return nil + } + case "tool_use": + if p.handleToolUse(msg) { + return nil + } + case "step_finish": + if err := p.handleStepFinish(msg); err != nil { + return err + } + if p.completed { + log.Trace("[opencode-parse] stream completed: lines=%d elapsed=%v", + lineCount, time.Since(startTime).Round(time.Second)) + return nil + } + log.Trace("[opencode-parse] step_finish (intermediate, reason!=stop): continuing parse loop") + case "reasoning": + p.handleReasoning(msg) + case "error": + log.Trace("[opencode-parse] error event: lines=%d elapsed=%v", + lineCount, time.Since(startTime).Round(time.Second)) + return p.handleError(msg) + default: + log.Trace("[opencode-parse] unknown event type: %s", msgType) + } + } + + 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()) + + if err := scanner.Err(); err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return err + } + return nil +} + +// --- Message lifecycle helpers (mirroring Claude parser) --- + +func (p *streamParser) beginMessageWithID(id, msgType string) (stopped bool) { + startData := message.EventMessageStartData{ + MessageID: id, + Type: msgType, + Timestamp: time.Now().UnixMilli(), + } + sd, _ := json.Marshal(startData) + return p.handler != nil && p.handler(message.ChunkMessageStart, sd) != 0 +} + +func (p *streamParser) beginMessage(msgType string) (messageID string, stopped bool) { + id := fmt.Sprintf("sandbox-%s-%s", msgType, message.GenerateNanoID()) + return id, p.beginMessageWithID(id, msgType) +} + +func (p *streamParser) endMessage() { + if p.handler != nil { + p.handler(message.ChunkMessageEnd, nil) + } +} + +func (p *streamParser) closeTextMessage() { + if p.textActive { + p.endMessage() + p.textActive = false + p.textMsgID = "" + } +} + +func (p *streamParser) ensureTextMessage() (stopped bool) { + if !p.textActive { + id, stopped := p.beginMessage("text") + if stopped { + return true + } + p.textActive = true + p.textMsgID = id + } + return false +} + +func (p *streamParser) emitText(text string) (stopped bool) { + text = strings.ReplaceAll(text, "\r\n", "\n") + text = strings.ReplaceAll(text, "\r", "\n") + return p.handler != nil && p.handler(message.ChunkText, []byte(text)) != 0 +} + +func (p *streamParser) emitExecute(props map[string]any) (stopped bool) { + data, _ := json.Marshal(props) + return p.handler != nil && p.handler(message.ChunkExecute, data) != 0 +} + +func (p *streamParser) emitMetadata(data map[string]any) { + if p.handler == nil { + return + } + encoded, _ := json.Marshal(data) + p.handler(message.ChunkMetadata, encoded) +} + +// --- Event handlers --- + +func (p *streamParser) handleStepStart(msg map[string]any) (stopped bool) { + if p.handler == nil { + return false + } + + sessionID, _ := msg["sessionID"].(string) + meta := map[string]any{"opencode_session_id": sessionID} + + part, _ := msg["part"].(map[string]any) + if part != nil { + if id, ok := part["id"].(string); ok { + meta["step_id"] = id + } + } + + p.emitMetadata(meta) + + // Record the step ID so that a subsequent tool_use can be correlated. + // We do NOT emit a "running" execute here because we can't tell yet + // whether the step will involve tool calls or just text output. For + // pure text responses an empty execute widget would look wrong. The + // "running" indicator is instead emitted lazily in handleToolUse when + // the first tool event actually arrives. + if part != nil { + if stepID, ok := part["id"].(string); ok && stepID != "" { + p.pendingExec[stepID] = "" // placeholder — msgID assigned later + } + } + return false +} + +func (p *streamParser) handleText(msg map[string]any) (stopped bool) { + if p.handler == nil { + return false + } + + part, _ := msg["part"].(map[string]any) + if part == nil { + return false + } + content, _ := part["text"].(string) + if content == "" { + content, _ = part["content"].(string) + } + if content == "" { + return false + } + + // Close any pending execute message before text output. + p.closePendingExec(msg) + + if p.ensureTextMessage() { + return true + } + if p.emitText(content) { + p.closeTextMessage() + return true + } + p.closeTextMessage() + + return false +} + +func (p *streamParser) handleToolUse(msg map[string]any) (stopped bool) { + if p.handler == nil { + return false + } + + part, _ := msg["part"].(map[string]any) + if part == nil { + return false + } + + state, _ := part["state"].(map[string]any) + if state == nil { + return false + } + + toolName, _ := part["toolName"].(string) + if toolName == "" { + toolName, _ = part["tool"].(string) + } + toolID, _ := part["toolCallId"].(string) + if toolID == "" { + toolID = fmt.Sprintf("oc-tool_%d_%d", p.toolIndex, time.Now().UnixNano()) + } + p.toolIndex++ + + status, _ := state["status"].(string) + isError := status == "error" + + p.closeTextMessage() + // Clear any pending step placeholder (no msgID to reuse since we didn't + // emit anything at step_start). + p.closePendingExec(msg) + + // Build execute properties. + execProps := map[string]any{ + "tool": toolName, + "tool_id": toolID, + "status": status, + "runner": "opencode-cli", + } + if isError { + execProps["is_error"] = true + } + + var inputStr string + if input, ok := state["input"].(string); ok && input != "" { + inputStr = input + } else if inputObj, ok := state["input"].(map[string]any); ok { + inputJSON, _ := json.Marshal(inputObj) + inputStr = string(inputJSON) + } + if inputStr != "" { + execProps["input"] = json.RawMessage(inputStr) + summary := extractSummary(toolName, inputStr) + if summary != "" { + execProps["summary"] = summary + } + } + + if output, ok := state["output"].(string); ok && output != "" { + execProps["output"] = output + } else if outputObj := state["output"]; outputObj != nil { + execProps["output"] = outputObj + } + + // Single message group with the complete tool result. + if _, stopped := p.beginMessage("execute"); stopped { + return true + } + if p.emitExecute(execProps) { + p.endMessage() + return true + } + p.endMessage() + return false +} + +func (p *streamParser) handleStepFinish(msg map[string]any) error { + p.closeTextMessage() + p.closePendingExec(msg) + + part, _ := msg["part"].(map[string]any) + reason := "" + if part != nil { + reason, _ = part["reason"].(string) + if reason == "" { + reason, _ = part["finishReason"].(string) + } + } + + if reason == "stop" || reason == "end_turn" { + p.completed = true + + if p.handler != nil { + finishMeta := map[string]any{ + "result_summary": map[string]any{ + "finish_reason": reason, + }, + } + // Include token/cost info if available. + if part != nil { + if tokens, ok := part["tokens"].(map[string]any); ok { + finishMeta["result_summary"].(map[string]any)["tokens"] = tokens + } + if cost, ok := part["cost"]; ok { + finishMeta["result_summary"].(map[string]any)["cost"] = cost + } + } + p.emitMetadata(finishMeta) + } + return nil + } + + // reason == "tool-calls" or other intermediate reasons: not final. + // Emit metadata so the frontend knows a new round is starting. + if p.handler != nil && reason != "" { + p.emitMetadata(map[string]any{ + "step_transition": map[string]any{ + "reason": reason, + }, + }) + } + return nil +} + +func (p *streamParser) handleReasoning(msg map[string]any) { + if p.handler == nil { + return + } + + part, _ := msg["part"].(map[string]any) + if part == nil { + return + } + content, _ := part["text"].(string) + if content == "" { + content, _ = part["content"].(string) + } + if content == "" { + return + } + + p.emitMetadata(map[string]any{ + "reasoning": content, + }) +} + +func (p *streamParser) handleError(msg map[string]any) error { + p.closePendingExec(msg) + + var errMsg string + + if part, ok := msg["part"].(map[string]any); ok { + errMsg, _ = part["error"].(string) + if errMsg == "" { + errMsg, _ = part["message"].(string) + } + } + if errMsg == "" { + switch e := msg["error"].(type) { + case string: + errMsg = e + case map[string]any: + errMsg, _ = e["message"].(string) + if errMsg == "" { + if data, ok := e["data"].(map[string]any); ok { + errMsg, _ = data["message"].(string) + } + } + if errMsg == "" { + name, _ := e["name"].(string) + if name != "" { + errMsg = name + } + } + } + } + if errMsg == "" { + errMsg = "unknown OpenCode error" + } + + if p.handler != nil { + p.handler(message.ChunkError, []byte(errMsg)) + } + return fmt.Errorf("OpenCode CLI error: %s", errMsg) +} + +// --- Pending exec helpers --- + +// closePendingExec clears step placeholders recorded at step_start. +func (p *streamParser) closePendingExec(msg map[string]any) { + if len(p.pendingExec) == 0 { + return + } + // Best-effort: clear matching step or all if we can't match. + part, _ := msg["part"].(map[string]any) + if part != nil { + for _, key := range []string{"id", "messageID"} { + if id, _ := part[key].(string); id != "" { + if _, ok := p.pendingExec[id]; ok { + delete(p.pendingExec, id) + return + } + } + } + } + // Fallback: clear the single pending entry (most common). + if len(p.pendingExec) == 1 { + for k := range p.pendingExec { + delete(p.pendingExec, k) + } + } +} + +// --- Utility --- + +// extractSummary builds a short human-readable summary from the tool input. +func extractSummary(toolName string, inputJSON string) string { + if inputJSON == "" { + return "" + } + var obj map[string]any + if err := json.Unmarshal([]byte(inputJSON), &obj); err != nil { + return "" + } + + switch strings.ToLower(toolName) { + case "bash", "execute": + if cmd, ok := obj["command"].(string); ok { + return truncate(cmd, 80) + } + case "write", "create": + if fp, ok := obj["file_path"].(string); ok { + return fp + } + case "read": + if fp, ok := obj["file_path"].(string); ok { + return fp + } + case "edit": + if fp, ok := obj["file_path"].(string); ok { + return fp + } + } + + for _, key := range []string{"path", "file_path", "command", "url", "query"} { + if v, ok := obj[key].(string); ok { + return truncate(v, 80) + } + } + return "" +} + +func truncate(s string, max int) string { + s = strings.TrimSpace(s) + s = strings.ReplaceAll(s, "\n", " ") + if len(s) > max { + return s[:max] + "..." + } + return s +} diff --git a/agent/sandbox/v2/opencode/parse_test.go b/agent/sandbox/v2/opencode/parse_test.go new file mode 100644 index 00000000..e3e049b6 --- /dev/null +++ b/agent/sandbox/v2/opencode/parse_test.go @@ -0,0 +1,528 @@ +package opencode + +import ( + "context" + "encoding/json" + "io" + "strings" + "sync" + "testing" + + "github.com/yaoapp/yao/agent/output/message" +) + +type chunkRecord struct { + eventType message.StreamChunkType + data string +} + +func collectHandler(records *[]chunkRecord, mu *sync.Mutex) message.StreamFunc { + return func(chunkType message.StreamChunkType, data []byte) int { + mu.Lock() + defer mu.Unlock() + *records = append(*records, chunkRecord{eventType: chunkType, data: string(data)}) + return 0 + } +} + +func makeJSONL(events ...map[string]any) string { + var lines []string + for _, e := range events { + data, _ := json.Marshal(e) + lines = append(lines, string(data)) + } + return strings.Join(lines, "\n") + "\n" +} + +func TestParse_StepStartEmitsMetadata(t *testing.T) { + input := makeJSONL( + map[string]any{ + "type": "step_start", + "timestamp": 1000, + "sessionID": "ses_123", + "part": map[string]any{ + "id": "prt_abc", + "type": "step-start", + "messageID": "msg_xyz", + "sessionID": "ses_123", + }, + }, + map[string]any{ + "type": "text", + "timestamp": 2000, + "sessionID": "ses_123", + "part": map[string]any{"text": "Hi!"}, + }, + map[string]any{ + "type": "step_finish", + "timestamp": 3000, + "sessionID": "ses_123", + "part": map[string]any{"reason": "stop"}, + }, + ) + + var records []chunkRecord + var mu sync.Mutex + handler := collectHandler(&records, &mu) + + parser := newStreamParser(handler) + reader := io.NopCloser(strings.NewReader(input)) + err := parser.parse(context.Background(), reader) + + if err != nil { + t.Fatalf("parse error: %v", err) + } + + mu.Lock() + defer mu.Unlock() + + // step_start should only emit metadata (no execute widget for pure text). + hasMeta := false + hasRunningExec := false + for _, r := range records { + if r.eventType == message.ChunkMetadata { + var meta map[string]any + json.Unmarshal([]byte(r.data), &meta) + if _, ok := meta["opencode_session_id"]; ok { + hasMeta = true + } + } + if r.eventType == message.ChunkExecute { + var props map[string]any + json.Unmarshal([]byte(r.data), &props) + if props["status"] == "running" { + hasRunningExec = true + } + } + } + if !hasMeta { + t.Error("step_start should emit metadata with session ID") + } + if hasRunningExec { + t.Error("step_start should NOT emit a running execute for pure text steps") + } +} + +func TestParse_TextEvent(t *testing.T) { + input := makeJSONL( + map[string]any{ + "type": "text", + "timestamp": 1000, + "sessionID": "ses_123", + "part": map[string]any{ + "type": "text", + "content": "Hello, world!", + }, + }, + map[string]any{ + "type": "step_finish", + "timestamp": 2000, + "sessionID": "ses_123", + "part": map[string]any{ + "type": "step-finish", + "finishReason": "stop", + }, + }, + ) + + var records []chunkRecord + var mu sync.Mutex + handler := collectHandler(&records, &mu) + + parser := newStreamParser(handler) + reader := io.NopCloser(strings.NewReader(input)) + err := parser.parse(context.Background(), reader) + + if err != nil { + t.Fatalf("parse error: %v", err) + } + if !parser.completed { + t.Error("parser should be completed") + } + + mu.Lock() + defer mu.Unlock() + + hasText := false + for _, r := range records { + if r.eventType == message.ChunkText { + hasText = true + if r.data != "Hello, world!" { + t.Errorf("text data = %q, want 'Hello, world!'", r.data) + } + } + } + if !hasText { + t.Error("should have emitted a ChunkText event") + } +} + +func TestParse_ToolUseEvent(t *testing.T) { + input := makeJSONL( + map[string]any{ + "type": "tool_use", + "timestamp": 1000, + "sessionID": "ses_123", + "part": map[string]any{ + "type": "tool", + "toolName": "bash", + "toolCallId": "call_abc", + "state": map[string]any{ + "status": "completed", + "input": `{"command":"ls -la"}`, + "output": "file1.txt\nfile2.txt", + }, + }, + }, + map[string]any{ + "type": "step_finish", + "timestamp": 2000, + "sessionID": "ses_123", + "part": map[string]any{ + "type": "step-finish", + "finishReason": "stop", + }, + }, + ) + + var records []chunkRecord + var mu sync.Mutex + handler := collectHandler(&records, &mu) + + parser := newStreamParser(handler) + reader := io.NopCloser(strings.NewReader(input)) + err := parser.parse(context.Background(), reader) + + if err != nil { + t.Fatalf("parse error: %v", err) + } + + mu.Lock() + defer mu.Unlock() + + hasCompletedExec := false + for _, r := range records { + if r.eventType == message.ChunkExecute { + var props map[string]any + json.Unmarshal([]byte(r.data), &props) + if props["tool"] == "bash" && props["status"] == "completed" { + hasCompletedExec = true + if props["runner"] != "opencode-cli" { + t.Errorf("runner = %v, want opencode-cli", props["runner"]) + } + } + } + } + if !hasCompletedExec { + t.Error("should have emitted a ChunkExecute with tool=bash status=completed") + } +} + +func TestParse_ErrorEvent(t *testing.T) { + input := makeJSONL( + map[string]any{ + "type": "error", + "timestamp": 1000, + "sessionID": "ses_123", + "error": "something went wrong", + }, + ) + + var records []chunkRecord + var mu sync.Mutex + handler := collectHandler(&records, &mu) + + parser := newStreamParser(handler) + reader := io.NopCloser(strings.NewReader(input)) + err := parser.parse(context.Background(), reader) + + if err == nil { + t.Fatal("expected error from parse") + } + if !strings.Contains(err.Error(), "something went wrong") { + t.Errorf("error should contain message, got: %v", err) + } + + mu.Lock() + defer mu.Unlock() + + hasError := false + for _, r := range records { + if r.eventType == message.ChunkError { + hasError = true + } + } + if !hasError { + t.Error("should have emitted a ChunkError event") + } +} + +func TestParse_ErrorEvent_Nested(t *testing.T) { + input := makeJSONL( + map[string]any{ + "type": "error", + "timestamp": 1000, + "sessionID": "ses_123", + "error": map[string]any{ + "name": "APIError", + "data": map[string]any{ + "message": "Authentication Fails", + "statusCode": 401, + }, + }, + }, + ) + + var records []chunkRecord + var mu sync.Mutex + handler := collectHandler(&records, &mu) + + parser := newStreamParser(handler) + reader := io.NopCloser(strings.NewReader(input)) + err := parser.parse(context.Background(), reader) + + if err == nil { + t.Fatal("expected error from parse") + } + if !strings.Contains(err.Error(), "Authentication Fails") { + t.Errorf("error should contain nested message, got: %v", err) + } +} + +func TestParse_StepFinishToolCalls(t *testing.T) { + input := makeJSONL( + map[string]any{ + "type": "step_finish", + "timestamp": 1000, + "sessionID": "ses_123", + "part": map[string]any{ + "type": "step-finish", + "finishReason": "tool-calls", + }, + }, + ) + + var records []chunkRecord + var mu sync.Mutex + handler := collectHandler(&records, &mu) + + parser := newStreamParser(handler) + reader := io.NopCloser(strings.NewReader(input)) + err := parser.parse(context.Background(), reader) + + if err != nil { + t.Fatalf("parse error: %v", err) + } + if parser.completed { + t.Error("tool-calls finish reason should NOT mark as completed") + } + + mu.Lock() + defer mu.Unlock() + + hasTransition := false + for _, r := range records { + if r.eventType == message.ChunkMetadata { + var meta map[string]any + json.Unmarshal([]byte(r.data), &meta) + if _, ok := meta["step_transition"]; ok { + hasTransition = true + } + } + } + if !hasTransition { + t.Error("tool-calls step_finish should emit step_transition metadata") + } +} + +func TestParse_MultiStepToolThenText(t *testing.T) { + input := makeJSONL( + map[string]any{"type": "step_start", "sessionID": "ses_1", "part": map[string]any{"id": "step-1"}}, + map[string]any{ + "type": "tool_use", "sessionID": "ses_1", + "part": map[string]any{ + "toolName": "bash", "toolCallId": "call_1", + "state": map[string]any{"status": "completed", "input": `{"command":"echo hi"}`, "output": "hi"}, + }, + }, + map[string]any{ + "type": "step_finish", "sessionID": "ses_1", + "part": map[string]any{"finishReason": "tool-calls"}, + }, + map[string]any{"type": "step_start", "sessionID": "ses_1", "part": map[string]any{"id": "step-2"}}, + map[string]any{ + "type": "text", "sessionID": "ses_1", + "part": map[string]any{"text": "The output was: hi"}, + }, + map[string]any{ + "type": "step_finish", "sessionID": "ses_1", + "part": map[string]any{"reason": "stop"}, + }, + ) + + var records []chunkRecord + var mu sync.Mutex + handler := collectHandler(&records, &mu) + + parser := newStreamParser(handler) + reader := io.NopCloser(strings.NewReader(input)) + err := parser.parse(context.Background(), reader) + + if err != nil { + t.Fatalf("parse error: %v", err) + } + if !parser.completed { + t.Error("multi-step stream should complete on final stop") + } + + mu.Lock() + defer mu.Unlock() + + var ( + completedCount int + hasText bool + ) + for _, r := range records { + if r.eventType == message.ChunkExecute { + var props map[string]any + json.Unmarshal([]byte(r.data), &props) + if props["status"] == "completed" { + completedCount++ + } + } + if r.eventType == message.ChunkText && r.data == "The output was: hi" { + hasText = true + } + } + if completedCount < 1 { + t.Error("should have at least 1 completed exec from tool_use") + } + if !hasText { + t.Error("should have emitted ChunkText for final text reply") + } +} + +func TestParse_ReasoningEvent(t *testing.T) { + input := makeJSONL( + map[string]any{ + "type": "reasoning", + "timestamp": 1000, + "sessionID": "ses_123", + "part": map[string]any{ + "type": "reasoning", + "text": "Let me think about this...", + }, + }, + map[string]any{ + "type": "step_finish", + "timestamp": 2000, + "sessionID": "ses_123", + "part": map[string]any{ + "type": "step-finish", + "finishReason": "stop", + }, + }, + ) + + var records []chunkRecord + var mu sync.Mutex + handler := collectHandler(&records, &mu) + + parser := newStreamParser(handler) + reader := io.NopCloser(strings.NewReader(input)) + err := parser.parse(context.Background(), reader) + + if err != nil { + t.Fatalf("parse error: %v", err) + } + + mu.Lock() + defer mu.Unlock() + + hasReasoning := false + for _, r := range records { + if r.eventType == message.ChunkMetadata { + var meta map[string]any + json.Unmarshal([]byte(r.data), &meta) + if _, ok := meta["reasoning"]; ok { + hasReasoning = true + } + } + } + if !hasReasoning { + t.Error("should have emitted reasoning metadata") + } +} + +func TestParse_UnknownEventType(t *testing.T) { + input := makeJSONL( + map[string]any{ + "type": "future_event_type", + "timestamp": 1000, + "sessionID": "ses_123", + }, + map[string]any{ + "type": "step_finish", + "timestamp": 2000, + "sessionID": "ses_123", + "part": map[string]any{ + "type": "step-finish", + "finishReason": "stop", + }, + }, + ) + + var records []chunkRecord + var mu sync.Mutex + handler := collectHandler(&records, &mu) + + parser := newStreamParser(handler) + reader := io.NopCloser(strings.NewReader(input)) + err := parser.parse(context.Background(), reader) + + if err != nil { + t.Fatalf("unknown event type should not cause error: %v", err) + } + if !parser.completed { + t.Error("parser should still complete") + } +} + +func TestExtractSummary(t *testing.T) { + cases := []struct { + tool, input, want string + }{ + {"bash", `{"command":"ls -la /tmp"}`, "ls -la /tmp"}, + {"read", `{"file_path":"main.go"}`, "main.go"}, + {"write", `{"file_path":"output.txt"}`, "output.txt"}, + {"unknown", `{"query":"select * from users"}`, "select * from users"}, + {"bash", `{"no_command_key": true}`, ""}, + {"bash", `invalid json`, ""}, + {"bash", "", ""}, + } + for _, tc := range cases { + got := extractSummary(tc.tool, tc.input) + if got != tc.want { + t.Errorf("extractSummary(%q, %q) = %q, want %q", tc.tool, tc.input, got, tc.want) + } + } +} + +func TestTruncate(t *testing.T) { + short := "hello" + if truncate(short, 80) != "hello" { + t.Error("short string should not be truncated") + } + + long := strings.Repeat("a", 100) + result := truncate(long, 80) + if len(result) != 83 { // 80 + "..." + t.Errorf("truncated len = %d, want 83", len(result)) + } + + withNewlines := "line1\nline2\nline3" + result = truncate(withNewlines, 80) + if strings.Contains(result, "\n") { + t.Error("truncate should replace newlines with spaces") + } +} diff --git a/agent/sandbox/v2/opencode/platform.go b/agent/sandbox/v2/opencode/platform.go new file mode 100644 index 00000000..4dc3cc44 --- /dev/null +++ b/agent/sandbox/v2/opencode/platform.go @@ -0,0 +1,62 @@ +package opencode + +import ( + "fmt" + "path" + "strings" + + infra "github.com/yaoapp/yao/sandbox/v2" +) + +// platform encapsulates OS-dependent behaviors for the target environment. +type platform interface { + OS() string + Shell() string + HomeEnv(workDir string) map[string]string + PathJoin(parts ...string) string + ShellCmd(script string) []string + KillCmd(pattern string) []string + KillSessionCmd(sessionName string) []string +} + +type posixBase struct { + os string + workDir string + shell string +} + +func (b *posixBase) OS() string { return b.os } +func (b *posixBase) Shell() string { return b.shell } +func (b *posixBase) PathJoin(parts ...string) string { return path.Join(parts...) } + +func (b *posixBase) HomeEnv(workDir string) map[string]string { + return map[string]string{"HOME": workDir} +} + +func (b *posixBase) ShellCmd(script string) []string { + return []string{"bash", "-c", script} +} + +func (b *posixBase) KillCmd(pattern string) []string { + return []string{"sh", "-c", fmt.Sprintf("pkill -f '%s' || true", pattern)} +} + +func (b *posixBase) KillSessionCmd(sessionName string) []string { + return []string{"sh", "-c", fmt.Sprintf("pkill -9 -f '%s' || true", sessionName)} +} + +func resolvePlatform(computer infra.Computer) platform { + sys := computer.ComputerInfo().System + osName := strings.ToLower(sys.OS) + workDir := computer.GetWorkDir() + shell := sys.Shell + + base := posixBase{os: osName, workDir: workDir, shell: shell} + if base.shell == "" { + base.shell = "bash" + } + if base.os == "" { + base.os = "linux" + } + return &base +} diff --git a/agent/sandbox/v2/opencode/roles_test.go b/agent/sandbox/v2/opencode/roles_test.go new file mode 100644 index 00000000..08230363 --- /dev/null +++ b/agent/sandbox/v2/opencode/roles_test.go @@ -0,0 +1,493 @@ +package opencode + +import ( + "encoding/json" + "testing" + + "github.com/yaoapp/gou/connector" + gouTypes "github.com/yaoapp/gou/types" + "github.com/yaoapp/xun/dbal/query" + "github.com/yaoapp/xun/dbal/schema" + "github.com/yaoapp/yao/agent/sandbox/v2/types" +) + +type fakeConn struct { + id string + typ int + settings map[string]interface{} +} + +func (f *fakeConn) Register(string, string, []byte) error { return nil } +func (f *fakeConn) Query() (query.Query, error) { return nil, nil } +func (f *fakeConn) Schema() (schema.Schema, error) { return nil, nil } +func (f *fakeConn) Close() error { return nil } +func (f *fakeConn) ID() string { return f.id } +func (f *fakeConn) Is(t int) bool { return f.typ == t } +func (f *fakeConn) Setting() map[string]interface{} { return f.settings } +func (f *fakeConn) GetMetaInfo() gouTypes.MetaInfo { return gouTypes.MetaInfo{} } + +func newFakeOpenAI(id, host, model, key string) *fakeConn { + return &fakeConn{ + id: id, + typ: connector.OPENAI, + settings: map[string]interface{}{ + "host": host, + "model": model, + "key": key, + }, + } +} + +func newFakeAnthropic(id, host, model, key string) *fakeConn { + return &fakeConn{ + id: id, + typ: connector.ANTHROPIC, + settings: map[string]interface{}{ + "host": host, + "model": model, + "key": key, + }, + } +} + +func registerFakeConnectors(t *testing.T, conns map[string]connector.Connector) func() { + t.Helper() + for id, c := range conns { + connector.Connectors[id] = c + } + return func() { + for id := range conns { + delete(connector.Connectors, id) + } + } +} + +// --------------------------------------------------------------------------- +// injectRoleProviders tests +// --------------------------------------------------------------------------- + +func TestInjectRoleProviders_VisionCustomProvider(t *testing.T) { + visionConn := newFakeOpenAI("vis", "https://api.mymaas.com/v1", "gpt-4o-mini", "sk-vis") + cleanup := registerFakeConnectors(t, map[string]connector.Connector{"vision-conn": visionConn}) + defer cleanup() + + primaryConn := newFakeOpenAI("primary", "https://api.deepseek.com", "deepseek-v4-flash", "sk-ds") + cfg := map[string]any{ + "provider": map[string]any{"custom": map[string]any{"npm": "@ai-sdk/openai-compatible"}}, + "model": "custom/deepseek-v4-flash", + "enabled_providers": []string{"custom"}, + } + + req := &types.PrepareRequest{ + Connector: primaryConn, + Config: &types.SandboxConfig{ + Runner: types.RunnerConfig{ + Connectors: map[string]*types.RoleConnector{ + "vision": {Connector: "vision-conn", Override: "force"}, + }, + }, + }, + } + + injectRoleProviders(cfg, req) + + providers := cfg["provider"].(map[string]any) + visionBlock, ok := providers["vision"] + if !ok { + t.Fatal("should have injected 'vision' provider block") + } + + vBlock := visionBlock.(map[string]any) + models := vBlock["models"].(map[string]any) + modelCfg := models["gpt-4o-mini"].(map[string]any) + + mods, ok := modelCfg["modalities"].(map[string][]string) + if !ok { + t.Fatal("vision model should have modalities declared") + } + if len(mods["input"]) != 2 || mods["input"][0] != "text" || mods["input"][1] != "image" { + t.Errorf("modalities.input = %v, want [text, image]", mods["input"]) + } + + enabled := cfg["enabled_providers"].([]string) + hasVision := false + for _, e := range enabled { + if e == "vision" { + hasVision = true + } + } + if !hasVision { + t.Error("enabled_providers should contain 'vision'") + } +} + +func TestInjectRoleProviders_VisionNativeOpenAI(t *testing.T) { + visionConn := newFakeOpenAI("vis", "", "gpt-4o-mini", "sk-oai") + cleanup := registerFakeConnectors(t, map[string]connector.Connector{"oai-vision": visionConn}) + defer cleanup() + + primaryConn := newFakeOpenAI("primary", "https://api.deepseek.com", "deepseek-v4-flash", "sk-ds") + cfg := map[string]any{ + "provider": map[string]any{"custom": map[string]any{"npm": "@ai-sdk/openai-compatible"}}, + "model": "custom/deepseek-v4-flash", + "enabled_providers": []string{"custom"}, + } + + req := &types.PrepareRequest{ + Connector: primaryConn, + Config: &types.SandboxConfig{ + Runner: types.RunnerConfig{ + Connectors: map[string]*types.RoleConnector{ + "vision": {Connector: "oai-vision", Override: "force"}, + }, + }, + }, + } + + injectRoleProviders(cfg, req) + + providers := cfg["provider"].(map[string]any) + visionBlock, ok := providers["vision"] + if !ok { + t.Fatal("should have separate 'vision' provider (different host from primary)") + } + + vBlock := visionBlock.(map[string]any) + models := vBlock["models"].(map[string]any) + modelCfg := models["gpt-4o-mini"].(map[string]any) + + if _, ok := modelCfg["modalities"]; !ok { + t.Error("native OpenAI vision model should still declare modalities") + } +} + +func TestInjectRoleProviders_LightWithDifferentHost(t *testing.T) { + lightConn := newFakeOpenAI("moonshot", "https://api.moonshot.cn/v1", "moonshot-v1-8k", "sk-moon") + cleanup := registerFakeConnectors(t, map[string]connector.Connector{"moonshot-conn": lightConn}) + defer cleanup() + + primaryConn := newFakeOpenAI("primary", "https://api.deepseek.com", "deepseek-v4-flash", "sk-ds") + cfg := map[string]any{ + "provider": map[string]any{"custom": map[string]any{"npm": "@ai-sdk/openai-compatible"}}, + "model": "custom/deepseek-v4-flash", + "enabled_providers": []string{"custom"}, + } + + req := &types.PrepareRequest{ + Connector: primaryConn, + Config: &types.SandboxConfig{ + Runner: types.RunnerConfig{ + Connectors: map[string]*types.RoleConnector{ + "light": {Connector: "moonshot-conn", Override: "force"}, + }, + }, + }, + } + + injectRoleProviders(cfg, req) + + providers := cfg["provider"].(map[string]any) + if _, ok := providers["light"]; !ok { + t.Fatal("light role should have its own provider block when host differs from primary") + } + + smallModel, ok := cfg["small_model"].(string) + if !ok || smallModel == "" { + t.Fatal("small_model should be set for light role") + } + if smallModel != "light/moonshot-v1-8k" { + t.Errorf("small_model = %q, want 'light/moonshot-v1-8k'", smallModel) + } + + enabled := cfg["enabled_providers"].([]string) + hasLight := false + for _, e := range enabled { + if e == "light" { + hasLight = true + } + } + if !hasLight { + t.Error("enabled_providers should contain 'light'") + } +} + +func TestInjectRoleProviders_LightSameHostAsPrimary(t *testing.T) { + lightConn := newFakeOpenAI("ds-light", "https://api.deepseek.com", "deepseek-chat", "sk-ds") + cleanup := registerFakeConnectors(t, map[string]connector.Connector{"ds-light-conn": lightConn}) + defer cleanup() + + primaryConn := newFakeOpenAI("primary", "https://api.deepseek.com", "deepseek-v4-flash", "sk-ds") + primaryProviderID, primaryCfg, modelStr := buildProviderConfig(primaryConn) + + cfg := map[string]any{ + "provider": map[string]any{primaryProviderID: primaryCfg}, + "model": modelStr, + "enabled_providers": []string{primaryProviderID}, + } + + req := &types.PrepareRequest{ + Connector: primaryConn, + Config: &types.SandboxConfig{ + Runner: types.RunnerConfig{ + Connectors: map[string]*types.RoleConnector{ + "light": {Connector: "ds-light-conn", Override: "force"}, + }, + }, + }, + } + + injectRoleProviders(cfg, req) + + providers := cfg["provider"].(map[string]any) + if _, ok := providers["light"]; ok { + t.Error("light should merge into primary block when same host, not create separate block") + } + + customBlock := providers["custom"].(map[string]any) + models := customBlock["models"].(map[string]any) + if _, ok := models["deepseek-chat"]; !ok { + t.Error("light model should be merged into primary's 'custom' provider models") + } + + smallModel := cfg["small_model"].(string) + if smallModel != "custom/deepseek-chat" { + t.Errorf("small_model = %q, want 'custom/deepseek-chat'", smallModel) + } +} + +func TestInjectRoleProviders_NoConnectors(t *testing.T) { + cfg := map[string]any{ + "provider": map[string]any{"openai": map[string]any{}}, + "model": "openai/gpt-4o", + "enabled_providers": []string{"openai"}, + } + + req := &types.PrepareRequest{ + Config: &types.SandboxConfig{}, + } + + injectRoleProviders(cfg, req) + + enabled := cfg["enabled_providers"].([]string) + if len(enabled) != 1 || enabled[0] != "openai" { + t.Errorf("enabled_providers should be unchanged: %v", enabled) + } +} + +func TestInjectRoleProviders_AnthropicVision(t *testing.T) { + visionConn := newFakeAnthropic("claude-vis", "https://api.anthropic.com", "claude-sonnet-4-5-20250929", "sk-ant") + cleanup := registerFakeConnectors(t, map[string]connector.Connector{"anthropic-vision": visionConn}) + defer cleanup() + + primaryConn := newFakeOpenAI("primary", "https://api.deepseek.com", "deepseek-v4-flash", "sk-ds") + cfg := map[string]any{ + "provider": map[string]any{"custom": map[string]any{"npm": "@ai-sdk/openai-compatible"}}, + "model": "custom/deepseek-v4-flash", + "enabled_providers": []string{"custom"}, + } + + req := &types.PrepareRequest{ + Connector: primaryConn, + Config: &types.SandboxConfig{ + Runner: types.RunnerConfig{ + Connectors: map[string]*types.RoleConnector{ + "vision": {Connector: "anthropic-vision", Override: "force"}, + }, + }, + }, + } + + injectRoleProviders(cfg, req) + + providers := cfg["provider"].(map[string]any) + visionBlock, ok := providers["vision"] + if !ok { + t.Fatal("should inject 'vision' provider for Anthropic connector") + } + + vBlock := visionBlock.(map[string]any) + if vBlock["npm"] != nil { + t.Error("Anthropic provider should NOT have npm field") + } +} + +// --------------------------------------------------------------------------- +// buildEnv role injection tests +// --------------------------------------------------------------------------- + +func TestInjectRoleEnvVars_Vision(t *testing.T) { + visionConn := newFakeOpenAI("vis", "https://api.mymaas.com/v1", "gpt-4o-mini", "sk-vis-key") + cleanup := registerFakeConnectors(t, map[string]connector.Connector{"vision-conn": visionConn}) + defer cleanup() + + req := &types.StreamRequest{ + Config: &types.SandboxConfig{ + Runner: types.RunnerConfig{ + Connectors: map[string]*types.RoleConnector{ + "vision": {Connector: "vision-conn", Override: "force"}, + }, + }, + }, + } + + env := map[string]string{} + injectRoleEnvVars(env, req) + + if env["YAO_VISION_KEY"] != "sk-vis-key" { + t.Errorf("YAO_VISION_KEY = %q, want 'sk-vis-key'", env["YAO_VISION_KEY"]) + } + if env["YAO_VISION_BASE_URL"] != "https://api.mymaas.com/v1" { + t.Errorf("YAO_VISION_BASE_URL = %q, want 'https://api.mymaas.com/v1'", env["YAO_VISION_BASE_URL"]) + } + if env["YAO_VISION_MODEL"] != "gpt-4o-mini" { + t.Errorf("YAO_VISION_MODEL = %q, want 'gpt-4o-mini'", env["YAO_VISION_MODEL"]) + } +} + +func TestInjectRoleEnvVars_Light(t *testing.T) { + lightConn := newFakeOpenAI("moon", "https://api.moonshot.cn/v1", "moonshot-v1-8k", "sk-moon") + cleanup := registerFakeConnectors(t, map[string]connector.Connector{"moon-conn": lightConn}) + defer cleanup() + + req := &types.StreamRequest{ + Config: &types.SandboxConfig{ + Runner: types.RunnerConfig{ + Connectors: map[string]*types.RoleConnector{ + "light": {Connector: "moon-conn", Override: "force"}, + }, + }, + }, + } + + env := map[string]string{} + injectRoleEnvVars(env, req) + + if env["YAO_LIGHT_KEY"] != "sk-moon" { + t.Errorf("YAO_LIGHT_KEY = %q, want 'sk-moon'", env["YAO_LIGHT_KEY"]) + } + if env["YAO_LIGHT_BASE_URL"] != "https://api.moonshot.cn/v1" { + t.Errorf("YAO_LIGHT_BASE_URL = %q, want 'https://api.moonshot.cn/v1'", env["YAO_LIGHT_BASE_URL"]) + } + if env["YAO_LIGHT_MODEL"] != "moonshot-v1-8k" { + t.Errorf("YAO_LIGHT_MODEL = %q, want 'moonshot-v1-8k'", env["YAO_LIGHT_MODEL"]) + } +} + +func TestInjectRoleEnvVars_NoConnectors(t *testing.T) { + req := &types.StreamRequest{ + Config: &types.SandboxConfig{}, + } + + env := map[string]string{} + injectRoleEnvVars(env, req) + + for _, prefix := range []string{"YAO_VISION", "YAO_LIGHT", "YAO_HEAVY", "YAO_SUBAGENT"} { + for _, suffix := range []string{"_KEY", "_BASE_URL", "_MODEL"} { + if v, ok := env[prefix+suffix]; ok { + t.Errorf("unexpected env %s=%s with no connectors", prefix+suffix, v) + } + } + } +} + +func TestInjectRoleEnvVars_MultipleRoles(t *testing.T) { + visionConn := newFakeOpenAI("vis", "https://api.vision.com", "vis-model", "sk-vis") + lightConn := newFakeOpenAI("light-c", "https://api.light.com", "light-model", "sk-light") + heavyConn := newFakeOpenAI("heavy-c", "https://api.heavy.com", "heavy-model", "sk-heavy") + + cleanup := registerFakeConnectors(t, map[string]connector.Connector{ + "vis-c": visionConn, + "light-c": lightConn, + "heavy-c": heavyConn, + }) + defer cleanup() + + req := &types.StreamRequest{ + Config: &types.SandboxConfig{ + Runner: types.RunnerConfig{ + Connectors: map[string]*types.RoleConnector{ + "vision": {Connector: "vis-c", Override: "force"}, + "light": {Connector: "light-c", Override: "force"}, + "heavy": {Connector: "heavy-c", Override: "force"}, + }, + }, + }, + } + + env := map[string]string{} + injectRoleEnvVars(env, req) + + if env["YAO_VISION_KEY"] != "sk-vis" { + t.Errorf("YAO_VISION_KEY = %q", env["YAO_VISION_KEY"]) + } + if env["YAO_LIGHT_KEY"] != "sk-light" { + t.Errorf("YAO_LIGHT_KEY = %q", env["YAO_LIGHT_KEY"]) + } + if env["YAO_HEAVY_KEY"] != "sk-heavy" { + t.Errorf("YAO_HEAVY_KEY = %q", env["YAO_HEAVY_KEY"]) + } +} + +// --------------------------------------------------------------------------- +// Full integration: buildOpenCodeConfig with role connectors +// --------------------------------------------------------------------------- + +func TestBuildOpenCodeConfig_WithVisionAndLight(t *testing.T) { + visionConn := newFakeOpenAI("vis", "https://api.mymaas.com/v1", "gpt-4o-mini", "sk-vis") + lightConn := newFakeOpenAI("moon", "https://api.moonshot.cn/v1", "moonshot-v1-8k", "sk-moon") + cleanup := registerFakeConnectors(t, map[string]connector.Connector{ + "vision-conn": visionConn, + "light-conn": lightConn, + }) + defer cleanup() + + primaryConn := newFakeOpenAI("primary", "https://api.deepseek.com", "deepseek-v4-flash", "sk-ds") + req := &types.PrepareRequest{ + AssistantID: "test-assistant", + Connector: primaryConn, + Config: &types.SandboxConfig{ + Runner: types.RunnerConfig{ + Connectors: map[string]*types.RoleConnector{ + "vision": {Connector: "vision-conn", Override: "force"}, + "light": {Connector: "light-conn", Override: "force"}, + }, + }, + }, + } + + data := buildOpenCodeConfig(req, nil) + var cfg map[string]any + if err := json.Unmarshal(data, &cfg); err != nil { + t.Fatalf("invalid JSON: %v", err) + } + + providers := cfg["provider"].(map[string]any) + + if _, ok := providers["custom"]; !ok { + t.Error("should have 'custom' provider for primary DeepSeek") + } + if _, ok := providers["vision"]; !ok { + t.Error("should have 'vision' provider block") + } + if _, ok := providers["light"]; !ok { + t.Error("should have 'light' provider block (different host from primary)") + } + + if cfg["model"] != "custom/deepseek-v4-flash" { + t.Errorf("model = %v, want custom/deepseek-v4-flash", cfg["model"]) + } + if cfg["small_model"] != "light/moonshot-v1-8k" { + t.Errorf("small_model = %v, want light/moonshot-v1-8k", cfg["small_model"]) + } + + enabled := cfg["enabled_providers"].([]any) + enabledSet := map[string]bool{} + for _, e := range enabled { + enabledSet[e.(string)] = true + } + for _, want := range []string{"custom", "vision", "light"} { + if !enabledSet[want] { + t.Errorf("enabled_providers should contain %q", want) + } + } +} diff --git a/agent/sandbox/v2/opencode/runner.go b/agent/sandbox/v2/opencode/runner.go new file mode 100644 index 00000000..4cb69855 --- /dev/null +++ b/agent/sandbox/v2/opencode/runner.go @@ -0,0 +1,228 @@ +package opencode + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/yaoapp/kun/log" + agentContext "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/output/message" + "github.com/yaoapp/yao/agent/sandbox/v2/shared" + "github.com/yaoapp/yao/agent/sandbox/v2/types" + infra "github.com/yaoapp/yao/sandbox/v2" +) + +// Runner implements the sandbox Runner interface for OpenCode CLI. +type Runner struct { + mode string + hasMCP bool + mcpServers []types.MCPServer + lastCompleted bool + lastChatID string + logger *agentContext.RequestLogger +} + +// New creates a new OpenCode Runner. +func New() *Runner { + return &Runner{mode: "cli"} +} + +// Name returns the runner identifier. Must NOT be "yao" (see agent.go branching). +func (r *Runner) Name() string { return "opencode" } + +// Prepare executes user-defined and runner-specific prepare steps. +func (r *Runner) Prepare(ctx context.Context, req *types.PrepareRequest) error { + r.mode = req.Config.Runner.Mode + if r.mode == "" { + r.mode = "cli" + } + + assistantID := req.AssistantID + prefix := ".yao/assistants/" + assistantID + if assistantID == "" { + prefix = ".opencode" + } + + steps := append([]types.PrepareStep{}, req.Config.Prepare...) + + // 1. Skills copy (aligned with Claude Runner) + if req.SkillsDir != "" { + ws := req.Computer.Workplace() + if ws != nil { + src := "local:///" + req.SkillsDir + dst := prefix + "/skills" + if _, err := ws.Copy(src, dst); err != nil { + log.Warn("[opencode-runner] copy skills %s -> %s: %v", src, dst, err) + } + } + } + + // 2. MCP servers -> stored for opencode.json generation + if len(req.MCPServers) > 0 { + r.hasMCP = true + r.mcpServers = req.MCPServers + } + + // 3. Create OPENCODE_*_DIR directories (data/config/state/cache) via + // workspace.FS so directory ownership matches the workspace mount. + // OpenCode writes files (e.g. .gitignore, SQLite DB) into these dirs + // on first startup and crashes if they don't exist. + if ws := req.Computer.Workplace(); ws != nil { + for _, sub := range []string{"data", "config", "state", "cache"} { + ws.MkdirAll(prefix+"/opencode/"+sub, 0777) + } + } + + // 4. Copy custom tools into OpenCode global config dir ($HOME/.config/opencode/tools/). + // Bun resolves symlinks and breaks module resolution (opencode#11001), so use cp. + // HOME is set to workDir (/workspace) at runtime, so this lands in persistent storage. + steps = append(steps, types.PrepareStep{ + Action: "exec", + Cmd: "mkdir -p $HOME/.config/opencode/tools && for f in /opt/opencode-tools/*.ts; do [ -f \"$f\" ] && cp -f \"$f\" $HOME/.config/opencode/tools/; done", + Once: true, + IgnoreError: true, + }) + + // 5. Generate opencode.json (project config at workspace root) + configJSON := buildOpenCodeConfig(req, r.mcpServers) + steps = append(steps, types.PrepareStep{ + Action: "file", + Path: "opencode.json", + Content: configJSON, + }) + + // 5. System prompt file + // Written via a prepare step so configHash dedup applies. + // opencode.json instructions field references this path. + // (System prompt is injected at Stream time if not a continuation.) + + // 6. Execute all prepare steps via RunPrepareSteps (configHash dedup) + if req.RunSteps != nil && len(steps) > 0 { + if err := req.RunSteps(ctx, steps, req.Computer, req.AssistantID, req.ConfigHash, req.AssistantDir); err != nil { + return fmt.Errorf("opencode prepare steps: %w", err) + } + } + + return nil +} + +// Stream executes the OpenCode CLI and streams output to handler. +func (r *Runner) Stream(ctx context.Context, req *types.StreamRequest, handler message.StreamFunc) error { + computer := req.Computer + if computer == nil { + return fmt.Errorf("computer is nil") + } + + p := resolvePlatform(computer) + + // Resolve attachments (shared with Claude runner). + var attachmentPaths []string + if req.ChatID != "" { + if ws := computer.Workplace(); ws != nil { + _, resolved, err := shared.PrepareAttachments(ctx, req.Messages, req.ChatID, ws) + if err != nil { + return fmt.Errorf("prepareAttachments: %w", err) + } + workDir := computer.GetWorkDir() + for _, ar := range resolved { + attachmentPaths = append(attachmentPaths, p.PathJoin(workDir, ar.Path)) + } + } + } + + // Write system prompt if this is the first turn. + assistantID := req.AssistantID + chatID := req.ChatID + storeKey := "opencode-session:" + assistantID + ":" + chatID + isContinuation := chatID != "" && chatSessionExists(storeKey) + + if !isContinuation && req.SystemPrompt != "" { + if ws := computer.Workplace(); ws != nil { + prefix := ".yao/assistants/" + assistantID + if assistantID == "" { + prefix = ".opencode" + } + promptPath := prefix + "/system-prompt.md" + envPrompt := buildSandboxEnvPrompt(p, computer.GetWorkDir()) + fullPrompt := req.SystemPrompt + "\n\n" + envPrompt + ws.MkdirAll(prefix, 0755) + ws.WriteFile(promptPath, []byte(fullPrompt), 0644) + } + } + + cmd := r.buildCommand(req, p, attachmentPaths) + + r.logger = req.Logger + if r.logger == nil { + r.logger = agentContext.NoopLogger() + } + + r.lastChatID = chatID + + log.Trace("[opencode-runner] Stream started: assistantID=%s chatID=%s", assistantID, chatID) + r.logger.Debug("env vars passed to session (%d total):", len(cmd.env)) + for k, v := range cmd.env { + if strings.HasPrefix(k, "CTX_") || k == "OPENCODE_DATA_DIR" || k == "HOME" || k == "WORKDIR" { + r.logger.Debug(" %s=%s", k, v) + } else { + r.logger.Debug(" %s=(set, len=%d)", k, len(v)) + } + } + + 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 + elapsed := time.Since(streamStart).Round(time.Second) + log.Trace("[opencode-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) + + // Mark session in store after a clean run (err==nil) so future + // requests can use --continue --session to resume. This covers both + // completed=true (single-step stop) and completed=false (multi-step + // with tool-calls where the process exited normally). + if err == nil && chatID != "" { + sessionID := chatIDToSessionID(assistantID, chatID) + markChatSession(storeKey, sessionID, 90*24*time.Hour) + } + + if completed || err == nil { + sess.shutdown() + } + return err +} + +// Cleanup kills any remaining opencode processes. If the stream completed +// normally (received step_finish with stop), child processes are preserved. +func (r *Runner) Cleanup(ctx context.Context, computer infra.Computer) error { + if computer == nil { + return nil + } + + log.Trace("[opencode-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") + } + return nil + } + + if r.mode != "service" { + p := resolvePlatform(computer) + if r.lastChatID != "" { + computer.Exec(ctx, p.KillSessionCmd(sanitizeSessionName(r.lastChatID))) + } else { + computer.Exec(ctx, p.KillCmd("opencode")) + } + } + + return nil +} diff --git a/agent/sandbox/v2/opencode/runner_test.go b/agent/sandbox/v2/opencode/runner_test.go new file mode 100644 index 00000000..ec8a9a0b --- /dev/null +++ b/agent/sandbox/v2/opencode/runner_test.go @@ -0,0 +1,162 @@ +package opencode_test + +import ( + "context" + "fmt" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/yaoapp/yao/agent/caller" + agentcontext "github.com/yaoapp/yao/agent/context" + sandboxtestutils "github.com/yaoapp/yao/agent/sandbox/v2/testutils" + oauthtypes "github.com/yaoapp/yao/openapi/oauth/types" +) + +const defaultTimeout = 3 * time.Minute + +// --------------------------------------------------------------------------- +// Scenario 1: Oneshot — new container per request, no session persistence +// --------------------------------------------------------------------------- + +func TestOpenCode_Oneshot(t *testing.T) { + sandboxtestutils.Prepare(t) + defer sandboxtestutils.Clean(t) + require.NotNil(t, caller.AgentGetterFunc) + + const assistantID = "tests.sandbox-v2.opencode-oneshot-cli" + agent, err := caller.AgentGetterFunc(assistantID) + require.NoError(t, err) + + chatID := fmt.Sprintf("e2e-oneshot-%d", time.Now().UnixMilli()) + ctx := agentcontext.New( + context.Background(), + &oauthtypes.AuthorizedInfo{TeamID: "test-team-e2e", UserID: "test-user-e2e"}, + chatID, + ) + + resp := streamAndWait(t, agent, ctx, "Reply exactly with: hello opencode sandbox", defaultTimeout) + + require.NotNil(t, resp.Completion) + assert.Equal(t, "assistant", resp.Completion.Role) + content := contentString(t, resp) + t.Logf("Oneshot response: %s", content) + assert.Contains(t, strings.ToLower(content), "hello opencode sandbox") +} + +// --------------------------------------------------------------------------- +// Scenario 2 & 3: Session — first turn (new conversation) + continuation +// --------------------------------------------------------------------------- + +func TestOpenCode_Session(t *testing.T) { + sandboxtestutils.Prepare(t) + defer sandboxtestutils.Clean(t) + require.NotNil(t, caller.AgentGetterFunc) + + const assistantID = "tests.sandbox-v2.opencode-session-cli" + agent, err := caller.AgentGetterFunc(assistantID) + require.NoError(t, err) + + chatID := fmt.Sprintf("e2e-session-%d", time.Now().UnixMilli()) + + // ── Turn 1: first message — creates a new session ────────────────── + t.Run("turn1_new_session", func(t *testing.T) { + ctx := agentcontext.New( + context.Background(), + &oauthtypes.AuthorizedInfo{TeamID: "test-team-e2e", UserID: "test-user-e2e"}, + chatID, + ) + + resp := streamAndWait(t, agent, ctx, + "Remember this secret code: PINEAPPLE-42. Reply with: understood", + defaultTimeout, + ) + + require.NotNil(t, resp.Completion) + assert.Equal(t, "assistant", resp.Completion.Role) + content := contentString(t, resp) + t.Logf("Turn 1 response: %s", content) + assert.Contains(t, strings.ToLower(content), "understood") + }) + + // ── Turn 2: continuation — reuses the session ────────────────────── + t.Run("turn2_continue_session", func(t *testing.T) { + ctx := agentcontext.New( + context.Background(), + &oauthtypes.AuthorizedInfo{TeamID: "test-team-e2e", UserID: "test-user-e2e"}, + chatID, + ) + + resp := streamAndWait(t, agent, ctx, + "What was the secret code I told you? Reply with just the code.", + defaultTimeout, + ) + + require.NotNil(t, resp.Completion) + assert.Equal(t, "assistant", resp.Completion.Role) + content := contentString(t, resp) + t.Logf("Turn 2 response: %s", content) + assert.Contains(t, strings.ToLower(content), "pineapple-42", + "continuation should recall secret from turn 1") + }) +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +func streamAndWait( + t *testing.T, + agent caller.AgentCaller, + ctx *agentcontext.Context, + prompt string, + timeout time.Duration, +) *agentcontext.Response { + t.Helper() + + messages := []agentcontext.Message{{Role: "user", Content: prompt}} + + done := make(chan struct{}) + var resp *agentcontext.Response + var streamErr error + + go func() { + defer close(done) + resp, streamErr = agent.Stream(ctx, messages) + }() + + select { + case <-done: + case <-time.After(timeout): + t.Fatalf("timeout after %v", timeout) + } + + if streamErr != nil { + t.Logf("Stream error: %v", streamErr) + } + require.NoError(t, streamErr, "Stream should not return error") + require.NotNil(t, resp, "response should not be nil") + + if resp.Completion != nil { + t.Logf("Completion: role=%s content=%v", resp.Completion.Role, resp.Completion.Content) + } + + require.NotNil(t, ctx.Buffer, "ctx.Buffer should not be nil") + msgs := ctx.Buffer.GetMessages() + t.Logf("buffer message count: %d", len(msgs)) + for _, m := range msgs { + t.Logf(" seq=%d role=%s type=%s streaming=%v", + m.Sequence, m.Role, m.Type, m.IsStreaming) + } + + return resp +} + +func contentString(t *testing.T, resp *agentcontext.Response) string { + t.Helper() + s, ok := resp.Completion.Content.(string) + require.True(t, ok, "Content should be string, got %T", resp.Completion.Content) + return s +} diff --git a/agent/sandbox/v2/opencode/session.go b/agent/sandbox/v2/opencode/session.go new file mode 100644 index 00000000..7be5c9f7 --- /dev/null +++ b/agent/sandbox/v2/opencode/session.go @@ -0,0 +1,261 @@ +package opencode + +import ( + "context" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "sync" + "time" + + agentContext "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/output/message" + infra "github.com/yaoapp/yao/sandbox/v2" +) + +// session encapsulates a single OpenCode CLI execution lifecycle: +// process start, stderr collection, kill on cancel, Wait with timeout. +type session struct { + ctx context.Context + computer infra.Computer + plat platform + exec *infra.ExecStream + stderr strings.Builder + stderrMu sync.Mutex + logger *agentContext.RequestLogger + chatID string +} + +func startSession(ctx context.Context, computer infra.Computer, p platform, cmd command, chatID string, logger *agentContext.RequestLogger) (*session, error) { + opts := []infra.ExecOption{infra.WithWorkDir(cmd.workDir), infra.WithEnv(cmd.env)} + + logger.Info("opencode session starting: cmd=%v workDir=%s platform=%s chatID=%s", + cmd.shell, cmd.workDir, p.OS(), chatID) + + execStream, err := computer.Stream(ctx, cmd.shell, opts...) + if err != nil { + return nil, fmt.Errorf("computer.Stream: %w", err) + } + + // Write user message to stdin, then close. OpenCode reads the prompt + // from stdin when no positional message is given (same as Claude runner). + // Closing after write signals EOF so OpenCode begins processing. + if execStream.Stdin != nil { + if cmd.stdin != "" { + if _, err := io.WriteString(execStream.Stdin, cmd.stdin); err != nil { + logger.Warn("failed to write stdin: %v", err) + } + } + execStream.Stdin.Close() + } + + return &session{ + ctx: ctx, + computer: computer, + plat: p, + exec: execStream, + logger: logger, + chatID: chatID, + }, nil +} + +// runStream executes the main stream processing loop. +// Returns (completed, error) where completed=true means OpenCode CLI sent +// a step_finish with reason=stop and the stream finished normally. +func (s *session) runStream(handler message.StreamFunc) (completed bool, err error) { + s.collectStderr() + + cleanup := s.watchCancel() + defer cleanup() + + // Tee stdout to a debug log so we can inspect raw JSONL timing. + stdout := s.teeStdout() + + parser := newStreamParser(handler) + parseErr := parser.parse(s.ctx, stdout) + + s.logger.Debug("runStream: parse returned completed=%v parseErr=%v", parser.completed, parseErr) + + if parser.completed { + s.logger.Info("opencode stream completed normally") + return true, nil + } + + exitErr := s.waitForExit(parseErr) + if exitErr != nil { + if handler != nil { + handler(message.ChunkError, []byte(exitErr.Error())) + } + return false, exitErr + } + + s.stderrMu.Lock() + stderrStr := strings.TrimSpace(s.stderr.String()) + s.stderrMu.Unlock() + if stderrStr != "" { + s.logger.Warn("opencode exited with code 0 but stream incomplete and stderr present: %s", stderrStr) + errMsg := fmt.Errorf("opencode CLI setup failed: %s", stderrStr) + if handler != nil { + handler(message.ChunkError, []byte(errMsg.Error())) + } + return false, errMsg + } + + return false, nil +} + +// teeStdout wraps exec.Stdout with a TeeReader that writes a copy to a +// timestamped log file. Returns the original Stdout if tee setup fails. +func (s *session) teeStdout() io.ReadCloser { + logDir := os.Getenv("YAO_LOG_PATH") + if logDir == "" { + logDir = "/tmp" + } + logFile := filepath.Join(logDir, fmt.Sprintf("opencode-stream-%s-%d.jsonl", s.chatID, time.Now().Unix())) + f, err := os.Create(logFile) + if err != nil { + s.logger.Debug("teeStdout: cannot create %s: %v", logFile, err) + return s.exec.Stdout + } + s.logger.Info("teeStdout: raw JSONL -> %s", logFile) + + tee := io.TeeReader(s.exec.Stdout, f) + return &teeReadCloser{Reader: tee, closers: []io.Closer{s.exec.Stdout, f}} +} + +type teeReadCloser struct { + io.Reader + closers []io.Closer +} + +func (t *teeReadCloser) Close() error { + var firstErr error + for _, c := range t.closers { + if err := c.Close(); err != nil && firstErr == nil { + firstErr = err + } + } + return firstErr +} + +func (s *session) collectStderr() { + go func() { + buf := make([]byte, 4096) + for { + n, err := s.exec.Stderr.Read(buf) + if n > 0 { + chunk := string(buf[:n]) + s.stderrMu.Lock() + s.stderr.WriteString(chunk) + s.stderrMu.Unlock() + s.logger.Debug("opencode stderr: %s", chunk) + } + if err != nil { + return + } + } + }() +} + +// killProcess terminates the OpenCode CLI process (Node.js). +func (s *session) killProcess(ctx context.Context) { + if s.chatID != "" { + name := sanitizeSessionName(s.chatID) + result, err := s.computer.Exec(ctx, s.plat.KillSessionCmd(name)) + s.logger.Debug("killProcess: KillSessionCmd(%s) exitCode=%d err=%v", name, result.ExitCode, err) + return + } + // OpenCode is a Node.js process; match both "opencode" and "node.*opencode" + result, err := s.computer.Exec(ctx, s.plat.KillCmd("opencode")) + s.logger.Debug("killProcess: KillCmd(opencode) exitCode=%d err=%v", result.ExitCode, err) +} + +func (s *session) watchCancel() func() { + done := make(chan struct{}) + go func() { + select { + case <-s.ctx.Done(): + s.logger.Info("context cancelled, killing opencode: %v", s.ctx.Err()) + killCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + s.killProcess(killCtx) + s.exec.Cancel() + case <-done: + } + }() + return func() { close(done) } +} + +// shutdown cleans up after normal stream completion. +// +// OpenCode exits cleanly after step_finish(stop), but we still need to +// release the Docker exec connection. Like Claude runner, we first kill +// only the opencode process with SIGKILL (which cannot be caught, so +// OpenCode has no chance to propagate signals to child processes), then +// close the exec connection. Children (browsers, servers, etc.) that were +// launched via nohup/setsid survive because they are in separate sessions. +func (s *session) shutdown() { + s.logger.Info("shutting down completed opencode exec session: chatID=%s", s.chatID) + killCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + s.killProcess(killCtx) + s.exec.Cancel() +} + +func (s *session) waitForExit(parseErr error) error { + s.logger.Info("opencode stream did not complete normally, waiting for exit") + + type waitResult struct { + exitCode int + err error + } + ch := make(chan waitResult, 1) + go func() { + code, err := s.exec.Wait() + ch <- waitResult{code, err} + }() + + var exitCode int + var waitErr error + + select { + case wr := <-ch: + exitCode, waitErr = wr.exitCode, wr.err + case <-s.ctx.Done(): + select { + case wr := <-ch: + exitCode, waitErr = wr.exitCode, wr.err + case <-time.After(10 * time.Second): + s.exec.Cancel() + s.logger.Error("opencode did not exit after kill, timeout") + return fmt.Errorf("opencode did not exit after kill (timeout)") + } + } + + s.stderrMu.Lock() + stderrStr := strings.TrimSpace(s.stderr.String()) + s.stderrMu.Unlock() + + if parseErr != nil { + if stderrStr != "" { + return fmt.Errorf("%w (stderr: %s)", parseErr, stderrStr) + } + return parseErr + } + if waitErr != nil { + if stderrStr != "" { + return fmt.Errorf("%w (stderr: %s)", waitErr, stderrStr) + } + return waitErr + } + if exitCode != 0 { + s.logger.Warn("opencode exited with non-zero code: exitCode=%d stderr=%s", exitCode, stderrStr) + if stderrStr != "" { + return fmt.Errorf("opencode CLI exited with code %d: %s", exitCode, stderrStr) + } + return fmt.Errorf("opencode CLI exited with code %d", exitCode) + } + return nil +} diff --git a/agent/sandbox/v2/prepare.go b/agent/sandbox/v2/prepare.go index e7297694..83e2ca65 100644 --- a/agent/sandbox/v2/prepare.go +++ b/agent/sandbox/v2/prepare.go @@ -166,12 +166,18 @@ func runExecStep(ctx context.Context, computer infra.Computer, step types.Prepar } } - rootDir := "/" - if isWindowsComputer(computer) { - rootDir = `C:\` + workDir := computer.GetWorkDir() + if workDir == "" { + workDir = "/" + if isWindowsComputer(computer) { + workDir = `C:\` + } } - result, err := computer.Exec(ctx, shellWrap(kind, script), infra.WithWorkDir(rootDir)) + result, err := computer.Exec(ctx, shellWrap(kind, script), + infra.WithWorkDir(workDir), + infra.WithEnv(map[string]string{"HOME": workDir}), + ) if err != nil { return err } diff --git a/agent/sandbox/v2/shared/attachments.go b/agent/sandbox/v2/shared/attachments.go new file mode 100644 index 00000000..01ab7c80 --- /dev/null +++ b/agent/sandbox/v2/shared/attachments.go @@ -0,0 +1,251 @@ +package shared + +import ( + "context" + "fmt" + "path/filepath" + "strings" + + agentContext "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/attachment" + workspace "github.com/yaoapp/yao/tai/workspace" +) + +// AttachmentResult holds the resolved attachment info after copying to workspace. +type AttachmentResult struct { + Path string // workspace-relative path (e.g. ".attachments/{chatID}/image.png") + ContentType string + Filename string + Bytes int +} + +// PrepareAttachments resolves __yao.attachment:// URLs in user messages, +// copies actual files into the workspace .attachments/{chatID}/ directory, +// and returns processed messages plus a list of resolved file paths. +// +// The returned messages have multimodal content replaced with text references +// (for runners like Claude that need text-only). Callers that need the raw +// file paths (like OpenCode's --file) can use the returned []AttachmentResult. +func PrepareAttachments(ctx context.Context, messages []agentContext.Message, chatID string, ws workspace.FS) ([]agentContext.Message, []AttachmentResult, error) { + usedNames := make(map[string]int) + attachDir := ".attachments/" + chatID + var resolved []AttachmentResult + + result := make([]agentContext.Message, len(messages)) + copy(result, messages) + + for i, msg := range result { + if msg.Role != "user" { + continue + } + + parts, ok := msg.Content.([]interface{}) + if !ok { + if typedParts, ok := msg.Content.([]agentContext.ContentPart); ok { + iparts := make([]interface{}, len(typedParts)) + for j, p := range typedParts { + m := map[string]interface{}{"type": string(p.Type)} + if p.Text != "" { + m["text"] = p.Text + } + if p.ImageURL != nil { + m["image_url"] = map[string]interface{}{ + "url": p.ImageURL.URL, + "detail": string(p.ImageURL.Detail), + } + } + if p.File != nil { + m["file"] = map[string]interface{}{ + "url": p.File.URL, + "filename": p.File.Filename, + } + } + iparts[j] = m + } + parts = iparts + } else { + continue + } + } + + if len(parts) == 0 { + continue + } + + var textParts []string + + for _, item := range parts { + m, ok := item.(map[string]interface{}) + if !ok { + continue + } + partType, _ := m["type"].(string) + + switch partType { + case "text": + if text, ok := m["text"].(string); ok && text != "" { + textParts = append(textParts, text) + } + + case "image_url": + imgData, _ := m["image_url"].(map[string]interface{}) + if imgData == nil { + continue + } + url, _ := imgData["url"].(string) + if url == "" { + continue + } + uploaderName, fileID, isWrapper := attachment.Parse(url) + if !isWrapper { + textParts = append(textParts, fmt.Sprintf("[Image: %s]", url)) + continue + } + ar, ref, err := resolveAttachment(ctx, uploaderName, fileID, "", attachDir, usedNames, ws) + if err != nil { + textParts = append(textParts, "[Attached image: failed to load]") + continue + } + resolved = append(resolved, *ar) + textParts = append(textParts, ref) + + case "file": + fileData, _ := m["file"].(map[string]interface{}) + if fileData == nil { + continue + } + url, _ := fileData["url"].(string) + hintName, _ := fileData["filename"].(string) + if url == "" { + continue + } + uploaderName, fileID, isWrapper := attachment.Parse(url) + if !isWrapper { + textParts = append(textParts, fmt.Sprintf("[File: %s]", url)) + continue + } + ar, ref, err := resolveAttachment(ctx, uploaderName, fileID, hintName, attachDir, usedNames, ws) + if err != nil { + textParts = append(textParts, "[Attached file: failed to load]") + continue + } + resolved = append(resolved, *ar) + textParts = append(textParts, ref) + } + } + + if len(textParts) > 0 { + newMsg := result[i] + newMsg.Content = strings.Join(textParts, "\n\n") + result[i] = newMsg + } + } + + return result, resolved, nil +} + +// resolveAttachment gets the local path of an attachment and copies it into +// the workspace via ws.Copy("local:///abs/path", ".attachments/{chatID}/filename"). +func resolveAttachment( + ctx context.Context, + uploaderName, fileID, hintName, attachDir string, + usedNames map[string]int, + ws workspace.FS, +) (*AttachmentResult, string, error) { + manager, exists := attachment.Managers[uploaderName] + if !exists { + return nil, "", fmt.Errorf("attachment manager not found: %s", uploaderName) + } + + fileInfo, err := manager.Info(ctx, fileID) + if err != nil { + return nil, "", fmt.Errorf("failed to get file info: %w", err) + } + + absPath, _, err := manager.LocalPath(ctx, fileID) + if err != nil { + return nil, "", fmt.Errorf("failed to get local path: %w", err) + } + + filename := fileInfo.Filename + if filename == "" && hintName != "" { + filename = hintName + } + if filename == "" { + ext := ExtensionFromContentType(fileInfo.ContentType) + filename = fileID + ext + } + + baseName := filename + if count, exists := usedNames[baseName]; exists { + ext := filepath.Ext(filename) + name := strings.TrimSuffix(filename, ext) + filename = fmt.Sprintf("%s_%d%s", name, count+1, ext) + usedNames[baseName] = count + 1 + } else { + usedNames[baseName] = 0 + } + + dstPath := attachDir + "/" + filename + src := "local:///" + absPath + + if _, err := ws.Copy(src, dstPath); err != nil { + return nil, "", fmt.Errorf("failed to copy attachment to workspace: %w", err) + } + + sizeStr := FormatFileSize(fileInfo.Bytes) + ref := fmt.Sprintf("[Attached file: %s (%s, %s)]", dstPath, fileInfo.ContentType, sizeStr) + + ar := &AttachmentResult{ + Path: dstPath, + ContentType: fileInfo.ContentType, + Filename: filename, + Bytes: fileInfo.Bytes, + } + + return ar, ref, nil +} + +// ExtensionFromContentType maps common MIME types to file extensions. +func ExtensionFromContentType(contentType string) string { + switch contentType { + case "image/png": + return ".png" + case "image/jpeg": + return ".jpg" + case "image/gif": + return ".gif" + case "image/webp": + return ".webp" + case "image/svg+xml": + return ".svg" + case "application/pdf": + return ".pdf" + case "text/plain": + return ".txt" + case "text/html": + return ".html" + case "text/css": + return ".css" + case "text/javascript", "application/javascript": + return ".js" + case "application/json": + return ".json" + case "application/zip": + return ".zip" + default: + return "" + } +} + +// FormatFileSize returns a human-readable file size string. +func FormatFileSize(bytes int) string { + switch { + case bytes >= 1024*1024: + return fmt.Sprintf("%.1fMB", float64(bytes)/(1024*1024)) + case bytes >= 1024: + return fmt.Sprintf("%.1fKB", float64(bytes)/1024) + default: + return fmt.Sprintf("%dB", bytes) + } +} diff --git a/sui/test/runner_test.go b/sui/test/runner_test.go index 00470b96..38836361 100644 --- a/sui/test/runner_test.go +++ b/sui/test/runner_test.go @@ -166,7 +166,7 @@ func TestRunnerWithFailAndSkip(t *testing.T) { assert.Equal(t, "passed", r.Status) case "TestRuntimeError": assert.Equal(t, "error", r.Status) - assert.Contains(t, r.Error, "deliberate runtime error") + assert.Contains(t, r.Error, "deliberate execution error") } } }