diff --git a/agent/assistant/agent.go b/agent/assistant/agent.go index d834dfdc..4f2fc7df 100644 --- a/agent/assistant/agent.go +++ b/agent/assistant/agent.go @@ -190,6 +190,9 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa ctx.Logger.Trace("Computer: %s", ci.BoxID) } ctx.Logger.Trace("Workspace: %s", ast.SandboxV2.WorkspaceID) + if conn, _, err := ast.GetConnector(ctx, opts); err == nil && conn != nil { + ctx.Logger.Trace("Connector: %s", conn.ID()) + } } } else if ast.HasSandbox() { ctx.Logger.Phase("Sandbox") @@ -328,7 +331,15 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa // Choose between sandbox execution or direct LLM execution if ast.HasSandboxV2() && v2Runner != nil && v2Computer != nil && v2Runner.Name() != "yao" { // V2 Sandbox execution path (non-yao runners replace LLM.Stream) - completionResponse, err = ast.executeSandboxV2Stream(ctx, completionMessages, agentNode, streamHandler, v2Runner, v2Computer, v2LoadingMsgID) + completionResponse, err = ast.executeSandboxV2Stream(ctx, &sandboxV2StreamParams{ + Messages: completionMessages, + AgentNode: agentNode, + Handler: streamHandler, + Runner: v2Runner, + Computer: v2Computer, + LoadingMsgID: v2LoadingMsgID, + Options: opts, + }) } else if ast.HasSandboxV2() && v2Runner != nil && v2Runner.Name() == "yao" { // V2 yao runner: Prepare is done, close loading, fall through to LLM if v2LoadingMsgID != "" { diff --git a/agent/assistant/sandbox_v2.go b/agent/assistant/sandbox_v2.go index 2ad61b64..07454367 100644 --- a/agent/assistant/sandbox_v2.go +++ b/agent/assistant/sandbox_v2.go @@ -79,15 +79,20 @@ func (ast *Assistant) initSandboxV2(ctx *context.Context, opts *context.Options) } } - // 3. Obtain Computer (passes connector for OPENAI_PROXY_* env injection). + // 3. Obtain Computer. updateLoadingV2(ctx, loadingMsgID, "sandbox.starting") - computer, identifier, err := sandboxv2.GetComputer(ctx, cfg, manager, conn) + computer, identifier, err := sandboxv2.GetComputer(ctx, cfg, manager) if err != nil { closeLoadingV2(ctx, loadingMsgID, "sandbox.failed") return nil, nil, nil, "", fmt.Errorf("getComputer failed: %w", err) } _ = identifier + // 3.5. Inject connector configs into a2o proxy via HTTP API. + if regErr := sandboxv2.RegisterProxyConfigs(stdCtx, computer); regErr != nil { + log.Printf("[sandbox/v2] RegisterProxyConfigs: %v (non-fatal)", regErr) + } + // 4. Get Runner. runner, err := sandboxv2.Get(cfg.Runner.Name) if err != nil { @@ -151,18 +156,23 @@ func (ast *Assistant) initSandboxV2(ctx *context.Context, opts *context.Options) return runner, computer, cleanup, loadingMsgID, nil } +// sandboxV2StreamParams groups arguments for executeSandboxV2Stream. +type sandboxV2StreamParams struct { + Messages []context.Message + AgentNode traceTypes.Node + Handler message.StreamFunc + Runner sandboxTypes.Runner + Computer infraV2.Computer + LoadingMsgID string + Options *context.Options +} + // executeSandboxV2Stream calls the V2 Runner.Stream and wraps it in the // standard completion response. func (ast *Assistant) executeSandboxV2Stream( - ctx *context.Context, - completionMessages []context.Message, - agentNode traceTypes.Node, - streamHandler message.StreamFunc, - runner sandboxTypes.Runner, - computer infraV2.Computer, - loadingMsgID string, + ctx *context.Context, p *sandboxV2StreamParams, ) (*context.CompletionResponse, error) { - _ = agentNode + _ = p.AgentNode cfg := ast.SandboxV2 manager := infraV2.M() @@ -170,16 +180,16 @@ func (ast *Assistant) executeSandboxV2Stream( // Build system prompt. var systemPrompt string if len(ast.Prompts) > 0 { - for _, p := range ast.Prompts { - if p.Role == "system" && p.Content != "" { - systemPrompt = p.Content + for _, pr := range ast.Prompts { + if pr.Role == "system" && pr.Content != "" { + systemPrompt = pr.Content break } } } - // Resolve connector for Stream. - conn, _, _ := ast.GetConnector(ctx) + // Resolve connector for Stream (respects user-selected connector via opts). + conn, _, _ := ast.GetConnector(ctx, p.Options) var tok *sandboxTypes.SandboxToken if ctx.Authorized != nil { @@ -191,10 +201,10 @@ func (ast *Assistant) executeSandboxV2Stream( } streamReq := &sandboxTypes.StreamRequest{ - Computer: computer, + Computer: p.Computer, Config: cfg, Connector: conn, - Messages: completionMessages, + Messages: p.Messages, SystemPrompt: systemPrompt, ChatID: ctx.ChatID, Token: tok, @@ -202,15 +212,15 @@ func (ast *Assistant) executeSandboxV2Stream( } execReq := &sandboxv2.ExecuteRequest{ - Computer: computer, - Runner: runner, + Computer: p.Computer, + Runner: p.Runner, Config: cfg, StreamReq: streamReq, Manager: manager, - LoadingMsgID: loadingMsgID, + LoadingMsgID: p.LoadingMsgID, } - return sandboxv2.ExecuteSandboxStream(ctx, execReq, streamHandler) + return sandboxv2.ExecuteSandboxStream(ctx, execReq, p.Handler) } // initStandaloneWorkspace loads the workspace FS into context when no sandbox diff --git a/agent/sandbox/v2/claude/command.go b/agent/sandbox/v2/claude/command.go index 42639b28..58c01b76 100644 --- a/agent/sandbox/v2/claude/command.go +++ b/agent/sandbox/v2/claude/command.go @@ -12,7 +12,7 @@ import ( infra "github.com/yaoapp/yao/sandbox/v2" ) -const defaultProxyPort = 3456 +const defaultA2OPort = 3099 type command struct { shell []string @@ -89,19 +89,19 @@ func buildEnv(req *types.StreamRequest, p platform) map[string]string { if req.Connector.Is(connector.ANTHROPIC) { env["ANTHROPIC_BASE_URL"] = host env["ANTHROPIC_API_KEY"] = key + if model != "" { + env["ANTHROPIC_MODEL"] = model + env["ANTHROPIC_DEFAULT_OPUS_MODEL"] = model + env["ANTHROPIC_DEFAULT_SONNET_MODEL"] = model + env["ANTHROPIC_DEFAULT_HAIKU_MODEL"] = model + env["CLAUDE_CODE_SUBAGENT_MODEL"] = model + } } else { - env["ANTHROPIC_BASE_URL"] = fmt.Sprintf("http://127.0.0.1:%d", defaultProxyPort) + connectorID := req.Connector.ID() + env["ANTHROPIC_BASE_URL"] = fmt.Sprintf("http://127.0.0.1:%d/c/%s", defaultA2OPort, connectorID) env["ANTHROPIC_API_KEY"] = "dummy" } - if model != "" { - env["ANTHROPIC_MODEL"] = model - env["ANTHROPIC_DEFAULT_OPUS_MODEL"] = model - env["ANTHROPIC_DEFAULT_SONNET_MODEL"] = model - env["ANTHROPIC_DEFAULT_HAIKU_MODEL"] = model - env["CLAUDE_CODE_SUBAGENT_MODEL"] = model - } - if thinking, ok := setting["thinking"].(map[string]interface{}); ok { thinkType, _ := thinking["type"].(string) switch thinkType { diff --git a/agent/sandbox/v2/lifecycle.go b/agent/sandbox/v2/lifecycle.go index 559403c2..02aaee14 100644 --- a/agent/sandbox/v2/lifecycle.go +++ b/agent/sandbox/v2/lifecycle.go @@ -1,12 +1,16 @@ package sandboxv2 import ( + "bytes" "context" "crypto/rand" "encoding/hex" + "encoding/json" "fmt" mathrand "math/rand" + "net/http" "strings" + "time" "github.com/yaoapp/gou/connector" "github.com/yaoapp/kun/log" @@ -96,9 +100,10 @@ func ResolveNodeID(ctx *agentContext.Context, cfg *types.SandboxConfig, manager } // GetComputer obtains or creates a Computer for the current request. -// An optional connector may be passed to inject OPENAI_PROXY_* env vars. +// Connector config injection is handled via RegisterProxyConfigs after +// the Computer is obtained. // Returns the Computer, the resolved identifier, and any error. -func GetComputer(ctx *agentContext.Context, cfg *types.SandboxConfig, manager *infra.Manager, conn ...connector.Connector) (infra.Computer, string, error) { +func GetComputer(ctx *agentContext.Context, cfg *types.SandboxConfig, manager *infra.Manager) (infra.Computer, string, error) { ownerID := resolveOwnerID(ctx) workspaceID := "" @@ -139,11 +144,11 @@ func GetComputer(ctx *agentContext.Context, cfg *types.SandboxConfig, manager *i if computerID != "" { log.Trace("[sandbox/v2] GetComputer: -> resolveComputerByID(%s)", computerID) - return resolveComputerByID(cfg, manager, computerID, ownerID, identifier, workspaceID, conn...) + return resolveComputerByID(cfg, manager, computerID, ownerID, identifier, workspaceID) } log.Trace("[sandbox/v2] GetComputer: -> resolveComputerByDSL (no computerID)") - return resolveComputerByDSL(cfg, manager, ownerID, identifier, workspaceID, conn...) + return resolveComputerByDSL(cfg, manager, ownerID, identifier, workspaceID) } // resolveComputerByID dispatches based on the runtime computer_id from metadata. @@ -151,7 +156,6 @@ func GetComputer(ctx *agentContext.Context, cfg *types.SandboxConfig, manager *i func resolveComputerByID( cfg *types.SandboxConfig, manager *infra.Manager, computerID, ownerID, identifier, workspaceID string, - conn ...connector.Connector, ) (infra.Computer, string, error) { // 1) Check if computer_id is a known Tai node (host or node kind). @@ -188,7 +192,7 @@ func resolveComputerByID( // Node with container runtime and DSL has image: create/reuse a box. cfg.Kind = "box" - return resolveBox(cfg, manager, ownerID, identifier, workspaceID, conn...) + return resolveBox(cfg, manager, ownerID, identifier, workspaceID) } // 2) Check if computer_id is an existing box ID. @@ -208,7 +212,6 @@ func resolveComputerByID( func resolveComputerByDSL( cfg *types.SandboxConfig, manager *infra.Manager, ownerID, identifier, workspaceID string, - conn ...connector.Connector, ) (infra.Computer, string, error) { log.Trace("[sandbox/v2] resolveComputerByDSL: cfgNodeID=%q image=%q", cfg.NodeID, cfg.Computer.Image) @@ -223,14 +226,13 @@ func resolveComputerByDSL( } log.Trace("[sandbox/v2] resolveComputerByDSL: -> resolveComputerByID(%s)", cfg.NodeID) - return resolveComputerByID(cfg, manager, cfg.NodeID, ownerID, identifier, workspaceID, conn...) + return resolveComputerByID(cfg, manager, cfg.NodeID, ownerID, identifier, workspaceID) } // resolveBox reuses or creates a box container. func resolveBox( cfg *types.SandboxConfig, manager *infra.Manager, ownerID, identifier, workspaceID string, - conn ...connector.Connector, ) (infra.Computer, string, error) { if workspaceID == "" && cfg.NodeID != "" { @@ -261,11 +263,7 @@ func resolveBox( } // Create new box. - var c connector.Connector - if len(conn) > 0 { - c = conn[0] - } - createOpts, err := BuildCreateOptions(cfg, identifier, ownerID, workspaceID, c) + createOpts, err := BuildCreateOptions(cfg, identifier, ownerID, workspaceID) if err != nil { return nil, identifier, fmt.Errorf("build create options: %w", err) } @@ -390,6 +388,128 @@ func pickNodeByFilter(filter *types.ComputerFilter, image string) (string, error return candidates[mathrand.Intn(len(candidates))], nil } +const defaultA2OPort = 3099 + +// a2oConnectorConfig mirrors the tai/a2o ConnectorConfig struct for JSON serialization. +type a2oConnectorConfig struct { + Backend string `json:"backend"` + Model string `json:"model"` + APIKey string `json:"api_key"` + Options map[string]interface{} `json:"options,omitempty"` +} + +// RegisterProxyConfigs injects all OpenAI-compatible connector configs into +// the a2o proxy running inside the container via POST /config. +// For host mode it uses Go's net/http directly; for box mode it uses computer.Exec. +func RegisterProxyConfigs(ctx context.Context, computer infra.Computer) error { + configs := buildA2OConfigs() + if len(configs) == 0 { + log.Trace("[sandbox/v2] RegisterProxyConfigs: no OpenAI connectors to register") + return nil + } + + data, err := json.Marshal(configs) + if err != nil { + return fmt.Errorf("marshal connector configs: %w", err) + } + + info := computer.ComputerInfo() + a2oURL := fmt.Sprintf("http://127.0.0.1:%d", defaultA2OPort) + + if info.Kind == "host" { + return registerProxyConfigsHTTP(a2oURL, data) + } + return registerProxyConfigsExec(ctx, computer, data) +} + +func buildA2OConfigs() map[string]*a2oConnectorConfig { + configs := make(map[string]*a2oConnectorConfig) + for id, conn := range connector.Connectors { + if !conn.Is(connector.OPENAI) { + continue + } + settings := conn.Setting() + if settings == nil { + continue + } + + cfg := &a2oConnectorConfig{} + if host, ok := settings["host"].(string); ok && host != "" { + cfg.Backend = connector.BuildAPIURL(host, "/chat/completions") + } + if model, ok := settings["model"].(string); ok && model != "" { + cfg.Model = model + } + if key, ok := settings["key"].(string); ok && key != "" { + cfg.APIKey = key + } + + extra := make(map[string]interface{}) + for k, v := range settings { + switch k { + case "host", "model", "key", "proxy", "type": + continue + default: + extra[k] = v + } + } + if len(extra) > 0 { + cfg.Options = extra + } + + if cfg.Backend != "" { + configs[id] = cfg + } + } + return configs +} + +func registerProxyConfigsHTTP(a2oURL string, data []byte) error { + client := &http.Client{Timeout: 10 * time.Second} + configURL := a2oURL + "/config" + + for attempt := 0; attempt < 10; attempt++ { + resp, err := client.Get(a2oURL + "/health") + if err == nil { + resp.Body.Close() + if resp.StatusCode == http.StatusOK { + break + } + } + time.Sleep(500 * time.Millisecond) + } + + resp, err := client.Post(configURL, "application/json", bytes.NewReader(data)) + if err != nil { + return fmt.Errorf("POST %s: %w", configURL, err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("POST %s: status %d", configURL, resp.StatusCode) + } + + log.Trace("[sandbox/v2] RegisterProxyConfigs(host): injected %d bytes", len(data)) + return nil +} + +func registerProxyConfigsExec(ctx context.Context, computer infra.Computer, data []byte) error { + escaped := strings.ReplaceAll(string(data), "'", "'\\''") + healthCmd := fmt.Sprintf("for i in $(seq 1 20); do wget -qO- http://127.0.0.1:%d/health >/dev/null 2>&1 && break; sleep 0.5; done", defaultA2OPort) + postCmd := fmt.Sprintf("echo '%s' | wget -qO- --post-data=@- --header='Content-Type: application/json' http://127.0.0.1:%d/config", escaped, defaultA2OPort) + script := healthCmd + " && " + postCmd + + result, err := computer.Exec(ctx, []string{"sh", "-c", script}) + if err != nil { + return fmt.Errorf("exec register proxy configs: %w", err) + } + if result.ExitCode != 0 { + return fmt.Errorf("register proxy configs: exit %d, stderr=%s", result.ExitCode, result.Stderr) + } + + log.Trace("[sandbox/v2] RegisterProxyConfigs(box): injected %d bytes, stdout=%s", len(data), result.Stdout) + return nil +} + func randomID() string { b := make([]byte, 8) _, _ = rand.Read(b) diff --git a/agent/sandbox/v2/options.go b/agent/sandbox/v2/options.go index 5080c7ee..2e51f55f 100644 --- a/agent/sandbox/v2/options.go +++ b/agent/sandbox/v2/options.go @@ -1,13 +1,11 @@ package sandboxv2 import ( - "encoding/json" "fmt" "os" "strings" "time" - "github.com/yaoapp/gou/connector" "github.com/yaoapp/yao/agent/sandbox/v2/types" infra "github.com/yaoapp/yao/sandbox/v2" ) @@ -21,9 +19,9 @@ func resolveEnvRef(value string) string { } // BuildCreateOptions converts a SandboxConfig into the V2 infrastructure -// CreateOptions. An optional connector is used to inject OPENAI_PROXY_* -// environment variables when the connector is OpenAI-compatible (non-Anthropic). -func BuildCreateOptions(cfg *types.SandboxConfig, identifier, ownerID, workspaceID string, conn ...connector.Connector) (infra.CreateOptions, error) { +// CreateOptions. Connector config injection is handled separately via the +// a2o HTTP API (POST /config) after the container starts. +func BuildCreateOptions(cfg *types.SandboxConfig, identifier, ownerID, workspaceID string) (infra.CreateOptions, error) { opts := infra.CreateOptions{ ID: identifier, Owner: ownerID, @@ -133,12 +131,6 @@ func BuildCreateOptions(cfg *types.SandboxConfig, identifier, ownerID, workspace opts.Env = make(map[string]string) } - // Inject OPENAI_PROXY_* when connector is OpenAI-compatible (non-Anthropic). - // The a2o proxy inside the container translates Anthropic API → OpenAI API. - if len(conn) > 0 && conn[0] != nil && !conn[0].Is(connector.ANTHROPIC) { - injectProxyEnv(opts.Env, conn[0]) - } - // Inject VNC_* environment variables from config. if cfg.Computer.VNC.Enabled { opts.Env["VNC_ENABLED"] = "true" @@ -156,42 +148,6 @@ func BuildCreateOptions(cfg *types.SandboxConfig, identifier, ownerID, workspace return opts, nil } -// injectProxyEnv extracts backend URL, model, and API key from an -// OpenAI-compatible connector's settings and writes them as OPENAI_PROXY_* -// environment variables into env. -func injectProxyEnv(env map[string]string, conn connector.Connector) { - settings := conn.Setting() - if settings == nil { - return - } - - if host, ok := settings["host"].(string); ok && host != "" { - env["OPENAI_PROXY_BACKEND"] = host - } - if model, ok := settings["model"].(string); ok && model != "" { - env["OPENAI_PROXY_MODEL"] = model - } - if key, ok := settings["key"].(string); ok && key != "" { - env["OPENAI_PROXY_API_KEY"] = key - } - - // Forward extra options as JSON. - extra := make(map[string]interface{}) - for k, v := range settings { - switch k { - case "host", "model", "key", "proxy", "type": - continue - default: - extra[k] = v - } - } - if len(extra) > 0 { - if data, err := json.Marshal(extra); err == nil { - env["OPENAI_PROXY_OPTIONS"] = string(data) - } - } -} - // parseMemory converts a human-readable memory string to bytes. // Supported formats: "4GB", "4G", "4g", "512MB", "512M", "512m", "1024KB", "1024K", "1024". func parseMemory(s string) (int64, error) {