opencode/yao
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #1498 from trheyi/main

Browse source 
refactor(sandbox): streamline sandbox configuration loading and enhan…
This commit is contained in:
Max 2026-03-18 11:27:53 +08:00 • committed by GitHub
commit 7255f613a9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 483 additions and 674 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
agent/store/types/sandbox_v2.go
View file

@ -7,9 +7,9 @@ import (
"os"
"path/filepath"
"sort"
"strings"
jsoniter "github.com/json-iterator/go"
"github.com/yaoapp/gou/application"
sandboxTypes "github.com/yaoapp/yao/agent/sandbox/v2/types"
)
@ -21,18 +21,9 @@ func LoadSandboxConfig(filePath string) (*sandboxTypes.SandboxConfig, error) {
return nil, fmt.Errorf("read sandbox config %s: %w", filePath, err)
}
ext := strings.ToLower(filepath.Ext(filePath))
var cfg sandboxTypes.SandboxConfig
switch ext {
case ".json", ".yao":
if err := jsoniter.Unmarshal(data, &cfg); err != nil {
return nil, fmt.Errorf("parse sandbox config (json): %w", err)
}
default:
if err := jsoniter.Unmarshal(data, &cfg); err != nil {
return nil, fmt.Errorf("parse sandbox config: %w", err)
}
if err := application.Parse(filepath.Base(filePath), data, &cfg); err != nil {
return nil, fmt.Errorf("parse sandbox config: %w", err)
}
if cfg.Version != sandboxTypes.SandboxVersionV2 {

60
config/types.go
View file

@ -14,31 +14,32 @@ func HostHasInternal(host string) bool {
// Config 象传应用引擎配置
type Config struct {
Mode string `json:"mode,omitempty" env:"YAO_ENV" envDefault:"production"` // The start mode production/development
AppSource string `json:"app,omitempty" env:"YAO_APP_SOURCE"` // The Application Source Root Path default same as Root
Root string `json:"root,omitempty" env:"YAO_ROOT" envDefault:"."` // The Application Root Path
Lang string `json:"lang,omitempty" env:"YAO_LANG" envDefault:"en-us"` // Default language setting
TimeZone string `json:"timezone,omitempty" env:"YAO_TIMEZONE"` // Default TimeZone
DataRoot string `json:"data_root,omitempty" env:"YAO_DATA_ROOT" envDefault:""` // The data root path
ExtensionRoot string `json:"extension_root,omitempty" env:"YAO_EXTENSION_ROOT" envDefault:""` // Plugin, Wasm root PATH, Default is <YAO_ROOT> (<YAO_ROOT>/plugins <YAO_ROOT>/wasms)
Host string `json:"host,omitempty" env:"YAO_HOST" envDefault:"0.0.0.0"` // The server host
Port int `json:"port,omitempty" env:"YAO_PORT" envDefault:"5099"` // The server port
Cert string `json:"cert,omitempty" env:"YAO_CERT"` // The HTTPS certificate path
Key string `json:"key,omitempty" env:"YAO_KEY"` // The HTTPS certificate key path
Log string `json:"log,omitempty" env:"YAO_LOG"` // The log file path
LogMode string `json:"log_mode,omitempty" env:"YAO_LOG_MODE" envDefault:"TEXT"` // The log mode TEXT|JSON
LogMaxSize int `json:"log_max_size,omitempty" env:"YAO_LOG_MAX_SIZE" envDefault:"100"` // The max log size in MB, the default is 100
LogMaxAage int `json:"log_max_age,omitempty" env:"YAO_LOG_MAX_AGE" envDefault:"7"` // The max log age in day, the default is 7
LogMaxBackups int `json:"log_max_backups" env:"YAO_LOG_MAX_BACKUPS" envDefault:"3"` // The max log backups, the default is 3
LogLocalTime bool `json:"log_local_time" env:"YAO_LOG_LOCAL_TIME" envDefault:"true"`
JWTSecret string `json:"jwt_secret,omitempty" env:"YAO_JWT_SECRET"` // The JWT Secret
DB Database `json:"db,omitempty"` // The database config
AllowFrom []string `json:"allowfrom,omitempty" envSeparator:"|" env:"YAO_ALLOW_FROM"` // Domain list the separator is |
Session Session `json:"session,omitempty"` // Session Config
Runtime Runtime `json:"runtime,omitempty"` // Runtime config
Trace Trace `json:"trace,omitempty"` // Trace config
Registry string `json:"registry,omitempty" env:"YAO_REGISTRY" envDefault:"https://registry.yaoagents.com"` // The package registry server URL
GRPC GRPCConfig `json:"grpc,omitempty"`
Mode string `json:"mode,omitempty" env:"YAO_ENV" envDefault:"production"` // The start mode production/development
AppSource string `json:"app,omitempty" env:"YAO_APP_SOURCE"` // The Application Source Root Path default same as Root
Root string `json:"root,omitempty" env:"YAO_ROOT" envDefault:"."` // The Application Root Path
Lang string `json:"lang,omitempty" env:"YAO_LANG" envDefault:"en-us"` // Default language setting
TimeZone string `json:"timezone,omitempty" env:"YAO_TIMEZONE"` // Default TimeZone
DataRoot string `json:"data_root,omitempty" env:"YAO_DATA_ROOT" envDefault:""` // The data root path
ExtensionRoot string `json:"extension_root,omitempty" env:"YAO_EXTENSION_ROOT" envDefault:""` // Plugin, Wasm root PATH, Default is <YAO_ROOT> (<YAO_ROOT>/plugins <YAO_ROOT>/wasms)
Host string `json:"host,omitempty" env:"YAO_HOST" envDefault:"0.0.0.0"` // The server host
Port int `json:"port,omitempty" env:"YAO_PORT" envDefault:"5099"` // The server port
Cert string `json:"cert,omitempty" env:"YAO_CERT"` // The HTTPS certificate path
Key string `json:"key,omitempty" env:"YAO_KEY"` // The HTTPS certificate key path
Log string `json:"log,omitempty" env:"YAO_LOG"` // The log file path
LogMode string `json:"log_mode,omitempty" env:"YAO_LOG_MODE" envDefault:"TEXT"` // The log mode TEXT|JSON
LogMaxSize int `json:"log_max_size,omitempty" env:"YAO_LOG_MAX_SIZE" envDefault:"100"` // The max log size in MB, the default is 100
LogMaxAage int `json:"log_max_age,omitempty" env:"YAO_LOG_MAX_AGE" envDefault:"7"` // The max log age in day, the default is 7
LogMaxBackups int `json:"log_max_backups" env:"YAO_LOG_MAX_BACKUPS" envDefault:"3"` // The max log backups, the default is 3
LogLocalTime bool `json:"log_local_time" env:"YAO_LOG_LOCAL_TIME" envDefault:"true"`
JWTSecret string `json:"jwt_secret,omitempty" env:"YAO_JWT_SECRET"` // The JWT Secret
DB Database `json:"db,omitempty"` // The database config
AllowFrom []string `json:"allowfrom,omitempty" envSeparator:"|" env:"YAO_ALLOW_FROM"` // Domain list the separator is |
Session Session `json:"session,omitempty"` // Session Config
Runtime Runtime `json:"runtime,omitempty"` // Runtime config
Trace Trace `json:"trace,omitempty"` // Trace config
Registry string `json:"registry,omitempty" env:"YAO_REGISTRY" envDefault:"https://registry.yaoagents.com"` // The package registry server URL
GRPC GRPCConfig `json:"grpc,omitempty"`
HostExec HostExecConfig `json:"host_exec,omitempty"`
}
// GRPCConfig gRPC server configuration
@ -88,6 +89,15 @@ type Runtime struct {
Import bool `json:"import,omitempty" env:"YAO_RUNTIME_IMPORT" envDefault:"true"` // If false the import statement will be disabled, the default value is true.
}
// HostExecConfig controls local host execution capability.
type HostExecConfig struct {
Enabled bool `json:"enabled,omitempty" env:"YAO_HOST_EXEC" envDefault:"false"` // Enable host execution on local node
FullAccess bool `json:"full_access,omitempty" env:"YAO_HOST_EXEC_FULL_ACCESS" envDefault:"false"` // Bypass command/dir checks
AllowedCommands []string `json:"allowed_commands,omitempty" env:"YAO_HOST_EXEC_ALLOWED_COMMANDS" envSeparator:","` // Allowed commands (comma-separated)
AllowedDirs []string `json:"allowed_dirs,omitempty" env:"YAO_HOST_EXEC_ALLOWED_DIRS" envSeparator:","` // Allowed working directories
DeniedDirs []string `json:"denied_dirs,omitempty" env:"YAO_HOST_EXEC_DENIED_DIRS" envSeparator:","` // Denied directories (higher priority)
}
// Trace config
type Trace struct {
Driver string `json:"driver,omitempty" env:"YAO_TRACE_DRIVER"` // The trace driver. local (development) | store (production)

66
engine/load.go
View file

@ -6,6 +6,7 @@ import (
"log"
"os"
"os/exec"
"path/filepath"
"regexp"
"strings"
"time"
@ -32,7 +33,6 @@ import (
"github.com/yaoapp/yao/kb"
"github.com/yaoapp/yao/mcp"
"github.com/yaoapp/yao/messenger"
"github.com/yaoapp/yao/moapi"
"github.com/yaoapp/yao/model"
"github.com/yaoapp/yao/monitor"
"github.com/yaoapp/yao/openapi"
@ -45,12 +45,10 @@ import (
"github.com/yaoapp/yao/schedule"
"github.com/yaoapp/yao/script"
"github.com/yaoapp/yao/share"
"github.com/yaoapp/yao/socket"
"github.com/yaoapp/yao/store"
sui "github.com/yaoapp/yao/sui/api"
tairegistry "github.com/yaoapp/yao/tai/registry"
"github.com/yaoapp/yao/tai"
"github.com/yaoapp/yao/task"
"github.com/yaoapp/yao/websocket"
"github.com/yaoapp/yao/widget"
"github.com/yaoapp/yao/widgets"
@ -133,20 +131,22 @@ func Load(cfg config.Config, options LoadOption, progressCallback ...func(string
warnings = append(warnings, Warning{Widget: "DB", Error: err})
}
// Initialize the Tai node registry (idempotent, safe to call early).
loadStep("Registry", func() error {
tairegistry.InitWithWriter(config.LogOutput, cfg.LogMode)
return nil
}, callback)
// Initialize the Sandbox manager and start it (auto-registers local Docker
// node if available, recovers existing containers, starts cleanup loop).
err = loadStep("Sandbox", func() error {
// Initialize the Tai registry, register local host node, then start the
// Sandbox manager (container recovery + cleanup loop).
err = loadStep("Registry", func() error {
dataDir := filepath.Join(cfg.DataRoot, "workspaces")
caps := tai.InitLocal(config.LogOutput, cfg.LogMode, dataDir)
if !caps.Docker {
log.Println("[Registry] Docker not available")
}
if caps.HostExec {
log.Println("[Registry] Host execution enabled (YAO_HOST_EXEC=true)")
}
sandbox.Init()
return sandbox.M().Start(context.Background())
}, callback)
if err != nil {
warnings = append(warnings, Warning{Widget: "Sandbox", Error: err})
warnings = append(warnings, Warning{Widget: "Registry", Error: err})
}
// Load Certs
@ -300,22 +300,6 @@ func Load(cfg config.Config, options LoadOption, progressCallback ...func(string
warnings = append(warnings, Warning{Widget: "API", Error: err})
}
// Load Sockets
err = loadStep("Socket", func() error {
return socket.Load(cfg)
}, callback)
if err != nil {
warnings = append(warnings, Warning{Widget: "Socket", Error: err})
}
// Load websockets (client mode)
err = loadStep("WebSocket", func() error {
return websocket.Load(cfg)
}, callback)
if err != nil {
warnings = append(warnings, Warning{Widget: "WebSocket", Error: err})
}
// Load tasks
err = loadStep("Task", func() error {
return task.Load(cfg)
@ -364,14 +348,6 @@ func Load(cfg config.Config, options LoadOption, progressCallback ...func(string
warnings = append(warnings, Warning{Widget: "SUI", Error: err})
}
// Load Moapi
err = loadStep("Moapi", func() error {
return moapi.Load(cfg)
}, callback)
if err != nil {
warnings = append(warnings, Warning{Widget: "Moapi", Error: err})
}
// Load Pipe
err = loadStep("Pipe", func() error {
return pipe.Load(cfg)
@ -496,8 +472,6 @@ func Unload() (err error) {
// importers
// tasks
// schedules
// sockets
// websockets
// widgets
// custom widget
@ -605,18 +579,6 @@ func Reload(cfg config.Config, options LoadOption) (err error) {
printErr(cfg.Mode, "API", err)
}
// Load Sockets
err = socket.Load(cfg) // Load sockets
if err != nil {
printErr(cfg.Mode, "Socket", err)
}
// Load websockets (client mode)
err = websocket.Load(cfg)
if err != nil {
printErr(cfg.Mode, "WebSocket", err)
}
// Load tasks
err = task.Load(cfg)
if err != nil {

36
moapi/api.go
View file

@ -1,36 +0,0 @@
package moapi
import "github.com/yaoapp/gou/api"
var dsl = []byte(`
{
"name": "Moapi API",
"description": "The API for Moapi",
"version": "1.0.0",
"guard": "bearer-jwt",
"group": "__moapi/v1",
"paths": [
{
"path": "/images/generations",
"method": "POST",
"process": "moapi.images.Generations",
"in": ["$payload.model", "$payload.prompt", ":payload"],
"out": { "status": 200, "type": "application/json" }
},
{
"path": "/chat/completions",
"guard": "query-jwt",
"method": "GET",
"process": "moapi.chat.Completions",
"processHandler": true,
"out": { "status": 200, "type": "text/event-stream" }
}
]
}
`)
func registerAPI() error {
_, err := api.LoadSource("<moapi.v1>.yao", dsl, "moapi.v1")
return err
}

169
moapi/moapi.go
View file

@ -1,169 +0,0 @@
package moapi
// *** WARNING ***
// Temporarily: change after the moapi is open source
//
import (
"fmt"
jsoniter "github.com/json-iterator/go"
"github.com/yaoapp/gou/http"
"github.com/yaoapp/yao/config"
"github.com/yaoapp/yao/share"
)
// Mirrors list all the mirrors
var cacheMirrors = []*Mirror{}
var cacheApps = []*App{}
var cacheMirrorsMap = map[string]*Mirror{}
// Models list all the models
var Models = []string{
"gpt-4-1106-preview",
"gpt-4-1106-vision-preview",
"gpt-4",
"gpt-4-32k",
"gpt-3.5-turbo",
"gpt-3.5-turbo-1106",
"gpt-3.5-turbo-instruct",
"dall-e-3",
"dall-e-2",
"tts-1",
"tts-1-hd",
"text-moderation-latest",
"text-moderation-stable",
"text-embedding-ada-002",
"whisper-1",
}
// Load load the moapi
func Load(cfg config.Config) error {
return registerAPI()
}
// Mirrors list all the mirrors
func Mirrors(cache bool) ([]*Mirror, error) {
if cache && len(cacheMirrors) > 0 {
return cacheMirrors, nil
}
bytes, err := httpGet("/api/moapi/mirrors")
if err != nil {
return nil, err
}
err = jsoniter.Unmarshal(bytes, &cacheMirrors)
if err != nil {
return nil, err
}
for _, mirror := range cacheMirrors {
cacheMirrorsMap[mirror.Host] = mirror
}
return cacheMirrors, nil
}
// Apps list all the apps
func Apps(cache bool) ([]*App, error) {
if cache && len(cacheApps) > 0 {
return cacheApps, nil
}
mirrors := SelectMirrors()
bytes, err := httpGet("/api/moapi/apps", mirrors...)
if err != nil {
return nil, err
}
err = jsoniter.Unmarshal(bytes, &cacheApps)
if err != nil {
return nil, err
}
channel := Channel()
if channel != "" {
for i := range cacheApps {
cacheApps[i].Homepage = cacheApps[i].Homepage + "?channel=" + channel
}
}
return cacheApps, nil
}
// Homepage get the home page url with the invite code
func Homepage() string {
channel := Channel()
if channel == "" {
return "https://store.moapi.ai"
}
return "https://store.moapi.ai" + "?channel=" + channel
}
// Channel get the channel
func Channel() string {
return share.App.Moapi.Channel
}
// SelectMirrors select the mirrors
func SelectMirrors() []*Mirror {
if share.App.Moapi.Mirrors == nil || len(share.App.Moapi.Mirrors) == 0 {
return []*Mirror{}
}
_, err := Mirrors(true)
if err != nil {
return []*Mirror{}
}
// pick the mirrors
var result []*Mirror
for _, host := range share.App.Moapi.Mirrors {
if mirror, ok := cacheMirrorsMap[host]; ok {
if mirror.Status == "on" {
result = append(result, mirror)
}
}
}
return result
}
// httpGet get the data from the api
func httpGet(api string, mirrors ...*Mirror) ([]byte, error) {
return httpGetRetry(api, mirrors, 0)
}
func httpGetRetry(api string, mirrors []*Mirror, retryTimes int) ([]byte, error) {
url := "https://" + share.MoapiHosts[retryTimes] + api
if len(mirrors) > retryTimes {
url = "https://" + mirrors[retryTimes].Host + api
}
secret := share.App.Moapi.Secret
organization := share.App.Moapi.Organization
http := http.New(url)
http.SetHeader("Authorization", "Bearer "+secret)
http.SetHeader("Content-Type", "application/json")
http.SetHeader("Moapi-Organization", organization)
resp := http.Get()
if resp.Code >= 500 {
if retryTimes > 3 {
return nil, fmt.Errorf("Moapi Server Error: %s", resp.Data)
}
return httpGetRetry(api, mirrors, retryTimes+1)
}
return jsoniter.Marshal(resp.Data)
}

146
moapi/process.go
View file

@ -1,146 +0,0 @@
package moapi
import (
"context"
"io"
"net/http"
"strings"
"github.com/gin-gonic/gin"
jsoniter "github.com/json-iterator/go"
"github.com/yaoapp/gou/process"
"github.com/yaoapp/kun/exception"
"github.com/yaoapp/yao/openai"
)
func init() {
process.RegisterGroup("moapi", map[string]process.Handler{
"images.generations": ImagesGenerations,
"chat.completions": ChatCompletions,
})
}
// ImagesGenerations Generate images
func ImagesGenerations(process *process.Process) interface{} {
process.ValidateArgNums(2)
model := process.ArgsString(0)
prompt := process.ArgsString(1)
option := process.ArgsMap(2, map[string]interface{}{})
if model == "" {
exception.New("ImagesGenerations error: model is required", 400).Throw()
}
if prompt == "" {
exception.New("ImagesGenerations error: prompt is required", 400).Throw()
}
ai, err := openai.NewMoapi(model)
if err != nil {
exception.New("ImagesGenerations error: %s", 400, err).Throw()
}
option["model"] = model
res, ex := ai.ImagesGenerations(prompt, option)
if ex != nil {
ex.Throw()
}
return res
}
// ChatCompletions chat completions
func ChatCompletions(process *process.Process) interface{} {
return func(c *gin.Context) {
option := map[string]interface{}{}
query := c.Query("payload")
err := jsoniter.UnmarshalFromString(query, &option)
if err != nil {
exception.New("ChatCompletions error: %s", 400, err).Throw()
}
// option := payload
// model := "gpt-3.5-turbo"
// messages := []map[string]interface{}{
// {
// "role": "system",
// "content": "You are a helpful assistant.",
// },
// {
// "role": "user",
// "content": "Hello!",
// },
// // }
// option["messages"] = messages
// option["model"] = model
delete(option, "context")
model, ok := option["model"].(string)
if !ok || model == "" {
exception.New("ChatCompletions error: model is required", 400).Throw()
}
ai, err := openai.NewMoapi(model)
if err != nil {
exception.New("ChatCompletions error: %s", 400, err).Throw()
}
if v, ok := option["stream"].(bool); ok && v {
chanStream := make(chan []byte, 1)
chanError := make(chan error, 1)
defer func() {
close(chanStream)
close(chanError)
}()
ctx, cancel := context.WithCancel(c.Request.Context())
defer cancel()
go ai.Stream(ctx, "/v1/chat/completions", option, func(data []byte) int {
if (string(data)) == "\n" || string(data) == "" {
return 1 // HandlerReturnOk
}
chanStream <- data
if strings.HasSuffix(string(data), "[DONE]") {
return 0 // HandlerReturnBreak0
}
return 1 // HandlerReturnOk
})
c.Header("Content-Type", "text/event-stream")
c.Stream(func(w io.Writer) bool {
select {
case err := <-chanError:
if err != nil {
c.JSON(http.StatusInternalServerError, err.Error())
}
return false
case msg := <-chanStream:
if string(msg) == "\n" {
return true
}
message := strings.TrimLeft(string(msg), "data: ")
c.SSEvent("message", message)
return true
case <-ctx.Done():
return false
}
})
return
}
return
}
}

34
moapi/types.go
View file

@ -1,34 +0,0 @@
package moapi
// Mirror is the mirror info
type Mirror struct {
Name string `json:"name"`
Host string `json:"host"`
Area string `json:"area"` // area code
Latency int `json:"latency"` // ms
Status string `json:"status"` // on, slow, off,
}
// App is the app info
type App struct {
Name string `json:"name"`
UpdatedAt int64 `json:"updated_at"`
CreatedAt int64 `json:"created_at"`
Country string `json:"country"`
Creator string `json:"creator"`
Description string `json:"description"`
Version string `json:"version"`
Short string `json:"short"`
Icon string `json:"icon"`
Homepage string `json:"homepage"`
Images []string `json:"images,omitempty"`
Videos []string `json:"videos,omitempty"`
Stat AppStat `json:"stat,omitempty"`
Languages []string `json:"languages"`
}
// AppStat is the app stat info
type AppStat struct {
Downloads int `json:"downloads"`
Stars int `json:"stars"`
}

4
openapi/computer/computer.go
View file

@ -89,7 +89,7 @@ func handleOptions(c *gin.Context) {
if kindFilter == "" || kindFilter == "host" {
for i := range snaps {
s := &snaps[i]
if !nodeOwnedBy(s, authInfo) {
if s.Mode != "local" && !nodeOwnedBy(s, authInfo) {
continue
}
if !s.Capabilities.HostExec {
@ -106,7 +106,7 @@ func handleOptions(c *gin.Context) {
if kindFilter == "" || kindFilter == "node" {
for i := range snaps {
s := &snaps[i]
if !nodeOwnedBy(s, authInfo) {
if s.Mode != "local" && !nodeOwnedBy(s, authInfo) {
continue
}
hasRuntime := s.Capabilities.Docker || s.Capabilities.K8s

20
openapi/nodes/nodes.go
View file

@ -98,19 +98,15 @@ func handleList(c *gin.Context) {
}
authInfo := authorized.GetInfo(c)
var snaps []taitypes.NodeMeta
if authInfo != nil && authInfo.TeamID != "" {
snaps = reg.ListByTeam(authInfo.TeamID)
} else if authInfo != nil && authInfo.UserID != "" {
snaps = reg.ListByUser(authInfo.UserID)
} else {
snaps = reg.List()
}
snaps := reg.List()
result := make([]nodeResponse, 0, len(snaps))
for _, s := range snaps {
result = append(result, snapToResponse(s))
for i := range snaps {
s := &snaps[i]
if s.Mode != "local" && !nodeOwnedBy(s, authInfo) {
continue
}
result = append(result, snapToResponse(*s))
}
response.RespondWithSuccess(c, http.StatusOK, result)
}
@ -130,7 +126,7 @@ func handleGet(c *gin.Context) {
}
authInfo := authorized.GetInfo(c)
if !nodeOwnedBy(snap, authInfo) {
if snap.Mode != "local" && !nodeOwnedBy(snap, authInfo) {
c.JSON(http.StatusForbidden, gin.H{"error": "no permission to access this node"})
return
}

6
openapi/sandbox/manage.go
View file

@ -251,15 +251,15 @@ func handleList(c *gin.Context) {
var result []sandboxResponse
// Host entries: list all nodes, filter by ownership + host_exec
// Host entries: list registered nodes that have any compute capability.
if reg := registry.Global(); reg != nil {
snaps := reg.List()
for i := range snaps {
s := &snaps[i]
if !nodeOwnedBy(s, authInfo) {
if s.Mode != "local" && !nodeOwnedBy(s, authInfo) {
continue
}
if !s.Capabilities.HostExec {
if !s.Capabilities.HostExec && !s.Capabilities.Docker {
continue
}
if nodeFilter != "" && s.TaiID != nodeFilter {

22
sandbox/v2/manager.go
View file

@ -4,13 +4,11 @@ import (
"context"
"fmt"
"log"
"path/filepath"
goruntime "runtime"
"strconv"
"sync"
"time"
"github.com/yaoapp/yao/config"
"github.com/yaoapp/yao/tai"
"github.com/yaoapp/yao/tai/registry"
tairuntime "github.com/yaoapp/yao/tai/runtime"
@ -29,16 +27,14 @@ func newManager() *Manager {
}
// Start discovers existing containers from all registered nodes and rebuilds
// the boxes map. If no "local" node is registered yet, it probes the local
// Docker environment and auto-registers one when available.
// the boxes map. The local node must already be registered by tai.InitLocal()
// before Start is called.
func (m *Manager) Start(ctx context.Context) error {
reg := registry.Global()
if reg == nil {
return nil
}
m.ensureLocalNode(reg)
for _, snap := range reg.List() {
res, err := m.getNode(snap.TaiID)
if err != nil {
@ -50,15 +46,6 @@ func (m *Manager) Start(ctx context.Context) error {
return nil
}
// ensureLocalNode delegates to tai.RegisterLocal() which probes the local
// Docker environment and registers a "local" node in the registry if available.
// The workspace data directory is derived from config.Conf.DataRoot so that
// workspace files persist across restarts.
func (m *Manager) ensureLocalNode(_ *registry.Registry) {
dataDir := filepath.Join(config.Conf.DataRoot, "workspaces")
tai.RegisterLocal(tai.WithDataDir(dataDir))
}
// Nodes returns the list of registered Tai nodes from the registry.
func (m *Manager) Nodes() []taitypes.NodeMeta {
reg := registry.Global()
@ -148,6 +135,11 @@ func (m *Manager) Create(ctx context.Context, opts CreateOptions) (*Box, error)
nodeID = targetNode
} else {
nodeID = node
if nodeID == "local" {
if ws, e := wsm.Get(ctx, opts.WorkspaceID); e == nil && ws.Owner != "" && ws.Owner != opts.Owner {
return nil, fmt.Errorf("sandbox: no permission to mount workspace %q", opts.WorkspaceID)
}
}
}
}
}

31
socket/socket.go
View file

@ -1,31 +0,0 @@
package socket
import (
"github.com/yaoapp/yao/config"
)
// Load 加载API
func Load(cfg config.Config) error {
// var root = filepath.Join(cfg.Root, "sockets")
// return LoadFrom(root, "")
return nil
}
// LoadFrom 从特定目录加载
// func LoadFrom(dir string, prefix string) error {
// if share.DirNotExists(dir) {
// return fmt.Errorf("%s does not exists", dir)
// }
// err := share.Walk(dir, ".sock.json", func(root, filename string) {
// name := prefix + share.SpecName(root, filename)
// content := share.ReadFile(filename)
// _, err := gou.LoadSocket(string(content), name)
// if err != nil {
// log.With(log.F{"root": root, "file": filename}).Error(err.Error())
// }
// })
// return err
// }

22
socket/socket_test.go
View file

@ -1,22 +0,0 @@
package socket
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/yaoapp/gou/socket"
"github.com/yaoapp/yao/config"
)
func TestLoad(t *testing.T) {
Load(config.Conf)
check(t)
}
func check(t *testing.T) {
keys := []string{}
for key := range socket.Sockets {
keys = append(keys, key)
}
assert.Equal(t, 0, len(keys))
}

25
tai/dial.go
View file

@ -7,6 +7,8 @@ import (
"net/http"
"time"
yaoconfig "github.com/yaoapp/yao/config"
"github.com/yaoapp/yao/tai/hostexec"
hepb "github.com/yaoapp/yao/tai/hostexec/pb"
"github.com/yaoapp/yao/tai/proxy"
"github.com/yaoapp/yao/tai/registry"
@ -88,15 +90,17 @@ func DialTunnel(taiID string, reg *registry.Registry, opts ...DialOption) (*Conn
return res, nil
}
// DialLocal establishes connections to the local Docker daemon.
// DialLocal establishes connections to the local host as a Tai node.
// Docker is probed but not required — when unavailable the node still
// provides Volume (and optionally HostExec) capabilities.
// Does NOT interact with the registry. Caller must call ConnResources.Close().
func DialLocal(addr string, dataDir string, vol volume.Volume) (*ConnResources, error) {
sb, err := runtime.NewLocal(addr)
if err != nil && vol == nil {
return nil, err
}
sb, _ := runtime.NewLocal(addr) // Docker failure is non-fatal
res := &ConnResources{DataDir: dataDir}
res := &ConnResources{
DataDir: dataDir,
System: CollectSystemInfo(),
}
if sb != nil {
res.Runtime = sb
@ -105,6 +109,15 @@ func DialLocal(addr string, dataDir string, vol volume.Volume) (*ConnResources,
res.VNC = vnc.NewLocal(sb)
}
if yaoconfig.Conf.HostExec.Enabled {
res.HostExec = hostexec.NewLocalClient(dataDir, hostexec.Policy{
FullAccess: yaoconfig.Conf.HostExec.FullAccess,
AllowedCommands: yaoconfig.Conf.HostExec.AllowedCommands,
AllowedDirs: yaoconfig.Conf.HostExec.AllowedDirs,
DeniedDirs: yaoconfig.Conf.HostExec.DeniedDirs,
})
}
if vol != nil {
res.Volume = vol
} else {

324
tai/hostexec/local.go Normal file
View file

@ -0,0 +1,324 @@
package hostexec
import (
"bytes"
"context"
"fmt"
"io"
"os"
"os/exec"
"path/filepath"
"strings"
"time"
pb "github.com/yaoapp/yao/tai/hostexec/pb"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
)
const defaultMaxOutputBytes = 10 * 1024 * 1024 // 10 MB
// Policy controls which commands and directories are allowed.
type Policy struct {
FullAccess bool // bypass command and path checks
AllowedCommands []string // empty = all denied (unless FullAccess)
AllowedDirs []string // working_dir must be under one of these
DeniedDirs []string // higher priority than AllowedDirs
}
// ---------------------------------------------------------------------------
// LocalClient — in-process HostExecClient (no gRPC network hop)
// ---------------------------------------------------------------------------
// LocalClient implements pb.HostExecClient by executing commands directly on
// the current host via os/exec.
type LocalClient struct {
defaultDir string
policy Policy
}
// Compile-time interface check.
var _ pb.HostExecClient = (*LocalClient)(nil)
// NewLocalClient creates a LocalClient.
func NewLocalClient(defaultDir string, policy Policy) *LocalClient {
return &LocalClient{defaultDir: defaultDir, policy: policy}
}
// Exec runs a command synchronously and returns the result.
func (c *LocalClient) Exec(ctx context.Context, req *pb.ExecRequest, _ ...grpc.CallOption) (*pb.ExecResponse, error) {
if err := c.checkCommand(req.Command); err != nil {
return &pb.ExecResponse{Error: err.Error()}, nil
}
if err := c.checkWorkingDir(req.WorkingDir); err != nil {
return &pb.ExecResponse{Error: err.Error()}, nil
}
timeout := time.Duration(req.TimeoutMs) * time.Millisecond
if timeout <= 0 {
timeout = 5 * time.Minute
}
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
cmd := exec.CommandContext(ctx, req.Command, req.Args...)
cmd.Dir = c.resolveDir(req.WorkingDir)
cmd.Env = c.buildEnv(req.Env)
if len(req.Stdin) > 0 {
cmd.Stdin = bytes.NewReader(req.Stdin)
}
maxBytes := req.MaxOutputBytes
if maxBytes <= 0 {
maxBytes = defaultMaxOutputBytes
}
var stdout, stderr bytes.Buffer
cmd.Stdout = &limitWriter{buf: &stdout, max: maxBytes}
cmd.Stderr = &limitWriter{buf: &stderr, max: maxBytes}
start := time.Now()
err := cmd.Run()
resp := &pb.ExecResponse{
Stdout: stdout.Bytes(),
Stderr: stderr.Bytes(),
DurationMs: time.Since(start).Milliseconds(),
}
if int64(len(resp.Stdout)+len(resp.Stderr)) >= maxBytes {
resp.Truncated = true
}
if err != nil {
if ctx.Err() != nil {
resp.Error = "command timed out"
resp.ExitCode = -1
} else if exitErr, ok := err.(*exec.ExitError); ok {
resp.ExitCode = int32(exitErr.ExitCode())
} else {
resp.Error = err.Error()
resp.ExitCode = -1
}
}
return resp, nil
}
// ExecStream runs a command and streams stdout/stderr via a channel-based
// adapter that satisfies grpc.ServerStreamingClient[pb.ExecOutput].
func (c *LocalClient) ExecStream(ctx context.Context, req *pb.ExecRequest, _ ...grpc.CallOption) (grpc.ServerStreamingClient[pb.ExecOutput], error) {
if err := c.checkCommand(req.Command); err != nil {
return newErrorStream(ctx, err.Error()), nil
}
if err := c.checkWorkingDir(req.WorkingDir); err != nil {
return newErrorStream(ctx, err.Error()), nil
}
timeout := time.Duration(req.TimeoutMs) * time.Millisecond
if timeout <= 0 {
timeout = 5 * time.Minute
}
ctx, cancel := context.WithTimeout(ctx, timeout)
cmd := exec.CommandContext(ctx, req.Command, req.Args...)
cmd.Dir = c.resolveDir(req.WorkingDir)
cmd.Env = c.buildEnv(req.Env)
if len(req.Stdin) > 0 {
cmd.Stdin = bytes.NewReader(req.Stdin)
}
stdoutPipe, err := cmd.StdoutPipe()
if err != nil {
cancel()
return newErrorStream(ctx, err.Error()), nil
}
stderrPipe, err := cmd.StderrPipe()
if err != nil {
cancel()
return newErrorStream(ctx, err.Error()), nil
}
if err := cmd.Start(); err != nil {
cancel()
return newErrorStream(ctx, err.Error()), nil
}
ch := make(chan *pb.ExecOutput, 64)
go func() {
defer cancel()
defer close(ch)
done := make(chan struct{})
go func() {
defer close(done)
streamPipe(ch, stdoutPipe, pb.ExecOutput_STDOUT)
}()
streamPipe(ch, stderrPipe, pb.ExecOutput_STDERR)
<-done
waitErr := cmd.Wait()
final := &pb.ExecOutput{Done: true}
if waitErr != nil {
if exitErr, ok := waitErr.(*exec.ExitError); ok {
final.ExitCode = int32(exitErr.ExitCode())
} else {
final.Error = waitErr.Error()
final.ExitCode = -1
}
}
ch <- final
}()
return &localStream{ctx: ctx, ch: ch}, nil
}
// ---------------------------------------------------------------------------
// Policy checks (identical to Tai hostexec/server.go)
// ---------------------------------------------------------------------------
func (c *LocalClient) checkCommand(command string) error {
if c.policy.FullAccess {
return nil
}
if len(c.policy.AllowedCommands) == 0 {
return fmt.Errorf("hostexec: no commands are allowed (allowed_commands is empty)")
}
base := filepath.Base(command)
for _, allowed := range c.policy.AllowedCommands {
if command == allowed || base == allowed {
return nil
}
}
return fmt.Errorf("hostexec: command %q is not in the allowed list", command)
}
func (c *LocalClient) checkWorkingDir(dir string) error {
if dir == "" || c.policy.FullAccess {
return nil
}
absDir, err := filepath.Abs(dir)
if err != nil {
return fmt.Errorf("hostexec: invalid working_dir %q: %w", dir, err)
}
resolved, err := filepath.EvalSymlinks(absDir)
if err != nil {
resolved = absDir
}
for _, denied := range c.policy.DeniedDirs {
if matchDir(resolved, denied) {
return fmt.Errorf("hostexec: working_dir %q is in a denied directory", dir)
}
}
if len(c.policy.AllowedDirs) == 0 {
return nil
}
for _, allowed := range c.policy.AllowedDirs {
if matchDir(resolved, allowed) {
return nil
}
}
return fmt.Errorf("hostexec: working_dir %q is not in any allowed directory", dir)
}
func matchDir(resolved, dir string) bool {
absDir, _ := filepath.Abs(dir)
resolvedDir, err := filepath.EvalSymlinks(absDir)
if err != nil {
resolvedDir = absDir
}
if resolved == resolvedDir {
return true
}
return strings.HasPrefix(resolved, resolvedDir+string(filepath.Separator))
}
func (c *LocalClient) resolveDir(dir string) string {
if dir != "" {
return dir
}
if c.defaultDir != "" {
return c.defaultDir
}
return ""
}
func (c *LocalClient) buildEnv(userEnv map[string]string) []string {
env := os.Environ()
for k, v := range userEnv {
env = append(env, k+"="+v)
}
return env
}
// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------
func streamPipe(ch chan<- *pb.ExecOutput, pipe io.ReadCloser, st pb.ExecOutput_Stream) {
buf := make([]byte, 32*1024)
for {
n, err := pipe.Read(buf)
if n > 0 {
data := make([]byte, n)
copy(data, buf[:n])
ch <- &pb.ExecOutput{Stream: st, Data: data}
}
if err != nil {
return
}
}
}
type limitWriter struct {
buf *bytes.Buffer
max int64
}
func (w *limitWriter) Write(p []byte) (int, error) {
remaining := w.max - int64(w.buf.Len())
if remaining <= 0 {
return len(p), nil
}
if int64(len(p)) > remaining {
p = p[:remaining]
}
return w.buf.Write(p)
}
// ---------------------------------------------------------------------------
// localStream — channel-based grpc.ServerStreamingClient adapter
// ---------------------------------------------------------------------------
type localStream struct {
ctx context.Context
ch <-chan *pb.ExecOutput
}
var _ grpc.ServerStreamingClient[pb.ExecOutput] = (*localStream)(nil)
func (s *localStream) Recv() (*pb.ExecOutput, error) {
select {
case <-s.ctx.Done():
return nil, s.ctx.Err()
case msg, ok := <-s.ch:
if !ok {
return nil, io.EOF
}
return msg, nil
}
}
func (s *localStream) Header() (metadata.MD, error) { return nil, nil }
func (s *localStream) Trailer() metadata.MD { return nil }
func (s *localStream) CloseSend() error { return nil }
func (s *localStream) Context() context.Context { return s.ctx }
func (s *localStream) SendMsg(any) error { return nil }
func (s *localStream) RecvMsg(any) error { return nil }
// newErrorStream returns a stream that yields a single Done message with the
// given error, then EOF. Used for early policy-check failures.
func newErrorStream(ctx context.Context, errMsg string) grpc.ServerStreamingClient[pb.ExecOutput] {
ch := make(chan *pb.ExecOutput, 1)
ch <- &pb.ExecOutput{Done: true, Error: errMsg, ExitCode: -1}
close(ch)
return &localStream{ctx: ctx, ch: ch}
}

37
tai/sysinfo.go Normal file
View file

@ -0,0 +1,37 @@
package tai
import (
"os"
"os/exec"
goruntime "runtime"
"github.com/yaoapp/yao/tai/types"
)
// CollectSystemInfo gathers system information for the local host.
// The result is identical in structure to what a remote Tai node reports
// via the ServerInfo gRPC service, keeping local and remote nodes symmetric.
func CollectSystemInfo() types.SystemInfo {
hostname, _ := os.Hostname()
return types.SystemInfo{
OS: goruntime.GOOS,
Arch: goruntime.GOARCH,
Hostname: hostname,
NumCPU: goruntime.NumCPU(),
Shell: detectShell(),
TempDir: os.TempDir(),
}
}
func detectShell() string {
if goruntime.GOOS != "windows" {
return "sh"
}
if _, err := exec.LookPath("pwsh"); err == nil {
return "pwsh"
}
if _, err := exec.LookPath("powershell"); err == nil {
return "powershell"
}
return "cmd.exe"
}

37
tai/tai.go
View file

@ -1,6 +1,8 @@
package tai
import (
"io"
"github.com/yaoapp/yao/tai/registry"
"github.com/yaoapp/yao/tai/types"
"github.com/yaoapp/yao/tai/volume"
@ -69,10 +71,11 @@ func intOr(v, fallback int) int {
return fallback
}
// RegisterLocal probes the local Docker environment and, if reachable,
// registers it as the "local" node in the registry with ConnResources.
// Returns true if a local node was successfully registered.
// Silently returns false if Docker is not available — this is not an error.
// RegisterLocal probes the local environment and registers the current host
// as the "local" node. Capabilities are set based on actual availability:
// Docker is probed, HostExec is controlled by YAO_HOST_EXEC env var.
// Always returns true — the local node is always registered (at minimum
// with Volume capability).
func RegisterLocal(opts ...Option) bool {
reg := registry.Global()
if reg == nil {
@ -93,13 +96,35 @@ func RegisterLocal(opts ...Option) bool {
}
reg.Register(&registry.TaiNode{
TaiID: "local",
Mode: "local",
TaiID: "local",
Mode: "local",
System: res.System,
Capabilities: types.Capabilities{
Docker: res.Runtime != nil,
HostExec: res.HostExec != nil,
},
})
reg.SetResources("local", res)
return true
}
// InitLocal initializes the Tai registry and registers the local host as a
// node in a single call. This is the preferred entry point for application
// startup.
//
// Capabilities are determined by probing the environment:
// - Docker reachable → Docker capability
// - YAO_HOST_EXEC=true → HostExec capability (with Policy from env)
// - Volume is always available
func InitLocal(w io.Writer, logMode string, dataDir string) types.Capabilities {
registry.InitWithWriter(w, logMode)
RegisterLocal(WithDataDir(dataDir))
if meta, ok := registry.Global().Get("local"); ok {
return meta.Capabilities
}
return types.Capabilities{}
}
// GetResources returns the ConnResources for a registered Tai node.
func GetResources(taiID string) (*ConnResources, bool) {
reg := registry.Global()

38
websocket/websocket.go
View file

@ -1,38 +0,0 @@
package websocket
import (
"github.com/yaoapp/yao/config"
)
// Load 加载API
func Load(cfg config.Config) error {
// exts := []string{"*.http.yao", "*.http.json", "*.http.jsonc"}
// return application.App.Walk("websockets", func(root, file string, isdir bool) error {
// _, err := websocket.Load(file, share.ID(root, file))
// return err
// }, exts...)
// var root = filepath.Join(cfg.Root, "websockets")
// return LoadFrom(root, "")
return nil
}
// // LoadFrom 从特定目录加载
// func LoadFrom(dir string, prefix string) error {
// if share.DirNotExists(dir) {
// return fmt.Errorf("%s does not exists", dir)
// }
// err := share.Walk(dir, ".ws.json", func(root, filename string) {
// name := prefix + share.SpecName(root, filename)
// content := share.ReadFile(filename)
// _, err := gou.LoadWebSocket(string(content), name)
// if err != nil {
// log.With(log.F{"root": root, "file": filename}).Error(err.Error())
// }
// })
// return err
// }

65
websocket/websocket_test.go
View file

@ -1,65 +0,0 @@
package websocket
import (
"fmt"
"net"
"net/http"
"testing"
"time"
"github.com/gin-gonic/gin"
"github.com/yaoapp/gou/websocket"
"github.com/yaoapp/yao/config"
)
func TestLoad(t *testing.T) {
Load(config.Conf)
check(t)
}
func TestWebSocketOpen(t *testing.T) {
// Load(config.Conf)
// script.Load(config.Conf)
// srv, url := serve(t)
// defer srv.Stop()
// ws := websocket.Se("message")
// err := ws.Open(url, "messageV2", "chatV3")
// if err != nil {
// t.Fatal(err)
// }
}
func serve(t *testing.T) (*websocket.Upgrader, string) {
ws, err := websocket.NewUpgrader("test")
if err != nil {
t.Fatalf("%s", err)
}
gin.SetMode(gin.ReleaseMode)
router := gin.Default()
ws.SetHandler(func(message []byte, id int) ([]byte, error) { return message, nil })
ws.SetRouter(router)
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
go ws.Start()
go func() {
http.Serve(listener, router)
}()
time.Sleep(200 * time.Millisecond)
return ws, fmt.Sprintf("ws://127.0.0.1:%d/websocket/test", listener.Addr().(*net.TCPAddr).Port)
}
func check(t *testing.T) {
// keys := []string{}
// for key := range gou.WebSockets {
// keys = append(keys, key)
// }
// assert.Equal(t, 1, len(keys))
}