diff --git a/apps/daemon/internal/agent/claudesdk/declaration.go b/apps/daemon/internal/agent/claudesdk/declaration.go index 30ebc5151..a7d7d536a 100644 --- a/apps/daemon/internal/agent/claudesdk/declaration.go +++ b/apps/daemon/internal/agent/claudesdk/declaration.go @@ -62,7 +62,7 @@ func discoverWithCheck(parent context.Context, options agent.DiscoveryOptions, d } out := &agent.Runtime{Info: descriptor, Session: func(context.Context, proto.PromptRequestPayload, chan<- proto.Envelope) (agent.Session, error) { return nil, fmt.Errorf("claude_sdk: configured runtime is unavailable") - }, View: nil} + }} var config Config fail := func(err error) *agent.Runtime { diff --git a/apps/daemon/internal/agent/codex/declaration.go b/apps/daemon/internal/agent/codex/declaration.go index d38c53382..8283da6ba 100644 --- a/apps/daemon/internal/agent/codex/declaration.go +++ b/apps/daemon/internal/agent/codex/declaration.go @@ -51,7 +51,7 @@ func discover(ctx context.Context, options agent.DiscoveryOptions, info proto.Su return discoverWithCheck(ctx, options, info, CheckCLIAvailable) } func discoverWithCheck(parent context.Context, options agent.DiscoveryOptions, info proto.SupportedAgentKind, check func(context.Context, string) (string, error)) *agent.Runtime { - runtime := &agent.Runtime{Info: info, Session: Factory, SessionCapabilityContext: true, ExecutorCapabilityContext: true, View: nil} + runtime := &agent.Runtime{Info: info, Session: Factory, SessionCapabilityContext: true, ExecutorCapabilityContext: true} ctx, cancel := context.WithTimeout(parent, 15*time.Second) defer cancel() version, err := check(ctx, "") diff --git a/apps/daemon/internal/agent/contract_declarations_test.go b/apps/daemon/internal/agent/contract_declarations_test.go index 76d404b7d..fc642d3f4 100644 --- a/apps/daemon/internal/agent/contract_declarations_test.go +++ b/apps/daemon/internal/agent/contract_declarations_test.go @@ -2,7 +2,6 @@ package agent_test import ( "encoding/json" - "fmt" "go/ast" "go/parser" "go/token" @@ -124,33 +123,6 @@ func TestPublicHarnessContractDeclarations(t *testing.T) { t.Errorf("%s needs explicit compile assertions on %v; got %v", name, want, assertions[name]) } } - // Each discovered Runtime decides its agent-host view explicitly, even when it declares none. - declaration, err := parser.ParseFile(token.NewFileSet(), filepath.Join(entry.Configuration, "declaration.go"), nil, 0) - if err != nil { - t.Fatal(err) - } - runtimes := 0 - ast.Inspect(declaration, func(node ast.Node) bool { - literal, ok := node.(*ast.CompositeLit) - if !ok { - return true - } - selector, ok := literal.Type.(*ast.SelectorExpr) - if !ok || selector.Sel.Name != "Runtime" { - return true - } - runtimes++ - for _, element := range literal.Elts { - if field, ok := element.(*ast.KeyValueExpr); ok && fmt.Sprint(field.Key) == "View" { - return true - } - } - t.Error("agent.Runtime literal must set View explicitly") - return true - }) - if runtimes == 0 { - t.Error("declaration.go constructs no agent.Runtime") - } }) } } diff --git a/apps/daemon/internal/agent/mcode/declaration.go b/apps/daemon/internal/agent/mcode/declaration.go index 276a7b0d7..08d02a53e 100644 --- a/apps/daemon/internal/agent/mcode/declaration.go +++ b/apps/daemon/internal/agent/mcode/declaration.go @@ -48,7 +48,7 @@ func discover(ctx context.Context, options agent.DiscoveryOptions, info proto.Su return discoverWithCheck(ctx, options, info, CheckCLIAvailable) } func discoverWithCheck(parent context.Context, options agent.DiscoveryOptions, result proto.SupportedAgentKind, check func(context.Context, string) (string, error)) *agent.Runtime { - runtime := &agent.Runtime{Info: result, Session: Factory, SessionCapabilityContext: true, ExecutorCapabilityContext: true, View: nil} + runtime := &agent.Runtime{Info: result, Session: Factory, SessionCapabilityContext: true, ExecutorCapabilityContext: true} ctx, cancel := context.WithTimeout(parent, 15*time.Second) defer cancel() diff --git a/apps/daemon/internal/agent/mcode/options.go b/apps/daemon/internal/agent/mcode/options.go index 5da808d31..1e7045d20 100644 --- a/apps/daemon/internal/agent/mcode/options.go +++ b/apps/daemon/internal/agent/mcode/options.go @@ -42,10 +42,12 @@ type launchOptions struct { } func prepareOptions(ctx context.Context, req proto.PromptRequestPayload) (launchOptions, error) { - return prepareOptionsWithSkills(ctx, req, true) + return prepareOptionsWithTools(ctx, req, nil) } -func prepareOptionsWithSkills(ctx context.Context, req proto.PromptRequestPayload, managedSkills bool) (launchOptions, error) { +// prepareOptionsWithTools installs the managed Skills unless the workspace +// bridge's tools present the Environment's. +func prepareOptionsWithTools(ctx context.Context, req proto.PromptRequestPayload, tools *workspaceTools) (launchOptions, error) { var result launchOptions if err := validateOptions(req); err != nil { return result, err @@ -67,7 +69,7 @@ func prepareOptionsWithSkills(ctx context.Context, req proto.PromptRequestPayloa if err := os.MkdirAll(result.Dir, 0o700); err != nil { return result, err } - if managedSkills { + if tools == nil { installed, err := managedskills.InstallManagedSkills(ctx, log.With("component", "mcode"), root, req.AgentOptions["skills"]) if err != nil { return result, err @@ -81,7 +83,7 @@ func prepareOptionsWithSkills(ctx context.Context, req proto.PromptRequestPayloa return result, err } defer data.Close() - if result.Model, err = writeNativeConfig(req, data); err != nil { + if result.Model, err = writeNativeConfig(req, data, result.DataDir, tools); err != nil { return result, err } opts := req.AgentOptions @@ -123,8 +125,10 @@ func validateOptions(req proto.PromptRequestPayload) error { } // writeNativeConfig writes the instructions and native configuration into the -// data directory and returns the model. -func writeNativeConfig(req proto.PromptRequestPayload, data *os.Root) (string, error) { +// data directory, which the native process sees at dataDir, and returns the +// model. With tools, the workspace bridge replaces native permissions and +// sandbox, and Subagents use it too. +func writeNativeConfig(req proto.PromptRequestPayload, data *os.Root, dataDir string, tools *workspaceTools) (string, error) { opts := req.AgentOptions prompt := optionString(opts, "system_prompt") if override := optionString(opts, "override_system_prompt"); override != "" { @@ -164,6 +168,31 @@ func writeNativeConfig(req proto.PromptRequestPayload, data *os.Root) (string, e return "", fmt.Errorf("mcode: unsupported permission mode") } config["permissionMode"] = mode + servers := map[string]any{} + if tools != nil { + config["permissionMode"] = "bypassPermissions" + config["sandbox"] = map[string]bool{"enabled": false} + if len(tools.skills) > 0 { + selected := config["agents"].(map[string]any)["default"].(map[string]any) + selected["skills"] = tools.skills + for _, key := range []string{"tools", "builtinTools"} { + selected[key] = append(selected[key].([]string), "skill") + } + } + raw, err := json.Marshal(tools.profile) + if err != nil { + return "", err + } + if err := data.WriteFile("workspace-profile.json", raw, 0o600); err != nil { + return "", err + } + if !req.DisableSubagents { + // This native data directory belongs to one public Session and its + // descendants. ACP's ephemeral server map otherwise covers only root. + server := tools.server(dataDir) + servers["oac_workspace"] = map[string]any{"type": "stdio", "command": server["command"], "args": server["args"], "env": map[string]string{}, "enabled": true} + } + } raw, err := json.Marshal(config) if err != nil { return "", err @@ -172,7 +201,10 @@ func writeNativeConfig(req proto.PromptRequestPayload, data *os.Root) (string, e return "", err } if req.StrictResume { - if err := data.WriteFile("mcp.json", []byte(`{"mcpServers":{}}`), 0o600); err != nil { + if raw, err = json.Marshal(map[string]any{"mcpServers": servers}); err != nil { + return "", err + } + if err := data.WriteFile("mcp.json", raw, 0o600); err != nil { return "", err } } diff --git a/apps/daemon/internal/agent/mcode/view.go b/apps/daemon/internal/agent/mcode/view.go index 8e3c703ed..44357fd3f 100644 --- a/apps/daemon/internal/agent/mcode/view.go +++ b/apps/daemon/internal/agent/mcode/view.go @@ -190,10 +190,12 @@ func (i viewInstall) prepare(_ context.Context, req proto.PromptRequestPayload, defer data.Close() opts := launchOptions{Dir: workspace, DataDir: filepath.Join(session.Home.Host, viewDataName), bindings: session.MCP, start: session.Launch, script: i.cli, home: session.Home.Host} - if opts.Model, err = writeNativeConfig(private, data); err != nil { + dataDir, tempDir := path.Join(session.Home.View, viewDataName), path.Join(session.Home.View, viewTempName) + tools := workspaceTools{node: i.node, bridge: i.bridge, profile: map[string]any{"workspace": workspace, "scratch": tempDir, "network": "enabled"}} + if opts.Model, err = writeNativeConfig(private, data, dataDir, &tools); err != nil { return opts, err } - dataDir, tempDir := path.Join(session.Home.View, viewDataName), path.Join(session.Home.View, viewTempName) + opts.MCP = append([]map[string]any{tools.server(dataDir)}, servers...) opts.Env = []string{ "PATH=" + path.Join(agent.ViewPrivateRoot, agent.ViewShimName), "TMPDIR=" + tempDir, @@ -208,7 +210,5 @@ func (i viewInstall) prepare(_ context.Context, req proto.PromptRequestPayload, opts.Env = append(opts.Env, nativeEnvironment(private, dataDir)...) opts.spawn = session.Spawn opts.reader = clirunner.StartOptions{Binary: i.node, Args: []string{path.Join(path.Dir(i.bridge), "subagent-snapshot.mjs"), dataDir}, Dir: dataDir, Env: opts.Env, OwnProcessGroup: true} - profile := map[string]any{"workspace": workspace, "scratch": tempDir, "network": "enabled"} - tools := workspaceTools{node: i.node, bridge: i.bridge, profile: path.Join(dataDir, "workspace-profile.json")} - return opts, writeWorkspaceTools(&opts, data, req, tools, profile, nil, servers) + return opts, nil } diff --git a/apps/daemon/internal/agent/mcode/workspace.go b/apps/daemon/internal/agent/mcode/workspace.go index ef7c77faa..2acd84092 100644 --- a/apps/daemon/internal/agent/mcode/workspace.go +++ b/apps/daemon/internal/agent/mcode/workspace.go @@ -2,7 +2,6 @@ package mcode import ( "context" - "encoding/json" "fmt" "os" "path/filepath" @@ -61,18 +60,29 @@ func prepareWorkspaceOptions(ctx context.Context, c WorkspaceConfig, req proto.P if err != nil { return launchOptions{}, err } + tools := workspaceTools{node: c.Node, bridge: c.Bridge, profile: map[string]any{"capabilityRoot": req.LocalEnvironment.CapabilityRoot, "workspace": c.Directory, "scratch": c.Scratch, "network": c.Network, "allowedDomains": (agentnetwork.Policy{Access: c.Network, AllowedDomains: c.AllowedDomains}).Hosts(), "skills": len(req.LocalEnvironment.Skills) > 0}} + file, err := localworkspace.ToolEnvironmentFile() + if err != nil { + return launchOptions{}, err + } + if file != "" { + tools.profile["toolEnvFile"] = file + } + for _, skill := range req.LocalEnvironment.Skills { + tools.skills = append(tools.skills, skill.Metadata.Name) + } // Reuse public option validation and private Session state provisioning. // The native process, ACP Session and workspace tools share the declared cwd. private := req private.LocalEnvironment, private.DisableExecutionEnvironment = nil, true // Public declarations have already been resolved into the transient ACP map. private.MCPHTTPServers = nil - opts, err := prepareOptionsWithSkills(ctx, private, false) + opts, err := prepareOptionsWithTools(ctx, private, &tools) if err != nil { return opts, err } opts.Dir, opts.bindings = c.Directory, bindings - var skills []string + opts.MCP = append([]map[string]any{tools.server(opts.DataDir)}, servers...) if len(req.LocalEnvironment.Skills) > 0 { root := filepath.Join(opts.DataDir, "skills") if err := os.MkdirAll(root, 0700); err != nil { @@ -90,79 +100,21 @@ func prepareWorkspaceOptions(ctx context.Context, c WorkspaceConfig, req proto.P } else if err := os.Symlink(target, link); err != nil { return opts, err } - skills = append(skills, skill.Metadata.Name) } } - profile := map[string]any{"capabilityRoot": req.LocalEnvironment.CapabilityRoot, "workspace": c.Directory, "scratch": c.Scratch, "network": c.Network, "allowedDomains": (agentnetwork.Policy{Access: c.Network, AllowedDomains: c.AllowedDomains}).Hosts(), "skills": len(skills) > 0} - file, err := localworkspace.ToolEnvironmentFile() - if err != nil { - return opts, err - } - if file != "" { - profile["toolEnvFile"] = file - } - data, err := os.OpenRoot(opts.DataDir) - if err != nil { - return opts, err - } - defer data.Close() - return opts, writeWorkspaceTools(&opts, data, req, workspaceTools{node: c.Node, bridge: c.Bridge, profile: filepath.Join(opts.DataDir, "workspace-profile.json")}, profile, skills, servers) + return opts, nil } -// workspaceTools are the workspace bridge as the native process runs it, and -// the bridge's profile path. -type workspaceTools struct{ node, bridge, profile string } +// workspaceTools is the workspace bridge as the native process runs it, the +// bridge's profile and the Skills it presents. +type workspaceTools struct { + node, bridge string + profile map[string]any + skills []string +} -// writeWorkspaceTools turns the native configuration in data over to the -// workspace bridge: native permissions and sandbox are off, the bridge's -// profile is written, and oac_workspace precedes the Session's servers. -func writeWorkspaceTools(opts *launchOptions, data *os.Root, req proto.PromptRequestPayload, tools workspaceTools, profile map[string]any, skills []string, servers []map[string]any) error { - raw, err := data.ReadFile("config.yaml") - if err != nil { - return err - } - var config map[string]any - if err = json.Unmarshal(raw, &config); err != nil { - return err - } - config["permissionMode"] = "bypassPermissions" - config["sandbox"] = map[string]bool{"enabled": false} - if len(skills) > 0 { - selected := config["agents"].(map[string]any)["default"].(map[string]any) - selected["skills"] = skills - for _, key := range []string{"tools", "builtinTools"} { - selected[key] = append(selected[key].([]any), "skill") - } - } - if raw, err = json.Marshal(config); err != nil { - return err - } - if err = data.WriteFile("config.yaml", raw, 0600); err != nil { - return err - } - if raw, err = json.Marshal(profile); err != nil { - return err - } - if err = data.WriteFile("workspace-profile.json", raw, 0600); err != nil { - return err - } - opts.MCP = []map[string]any{{"name": "oac_workspace", "command": tools.node, "args": []string{tools.bridge, tools.profile}, "env": []map[string]string{}}} - opts.MCP = append(opts.MCP, servers...) - if !req.DisableSubagents { - // This native data directory belongs to one public Session and its - // descendants. ACP's ephemeral server map otherwise covers only root. - configured := map[string]any{} - for _, server := range opts.MCP[:1] { - name, _ := server["name"].(string) - configured[name] = map[string]any{"type": "stdio", "command": server["command"], "args": server["args"], "env": map[string]string{}, "enabled": true} - } - raw, err := json.Marshal(map[string]any{"mcpServers": configured}) - if err != nil { - return err - } - if err := data.WriteFile("mcp.json", raw, 0o600); err != nil { - return err - } - } - return nil +// server is the bridge's ACP MCP server, with its profile in dataDir as the +// native process sees it. +func (t workspaceTools) server(dataDir string) map[string]any { + return map[string]any{"name": "oac_workspace", "command": t.node, "args": []string{t.bridge, filepath.Join(dataDir, "workspace-profile.json")}, "env": []map[string]string{}} } diff --git a/apps/daemon/internal/agenthost/admit.go b/apps/daemon/internal/agenthost/admit.go index 7a0e485cf..81fa4165c 100644 --- a/apps/daemon/internal/agenthost/admit.go +++ b/apps/daemon/internal/agenthost/admit.go @@ -3,7 +3,6 @@ package agenthost import ( "context" "crypto/x509" - "encoding/json" "fmt" "maps" "math" @@ -23,9 +22,6 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" ) -// modelName is the gateway's name for the request's model provider. -const modelName = "model_provider" - // etcFiles are the files the agent host writes for each Session and presents // at /etc/. var etcFiles = []string{"passwd", "group", "hosts", "resolv.conf", "nsswitch.conf"} @@ -72,7 +68,7 @@ func checkConfig(cfg Config) (*x509.CertPool, error) { func loadRoots(dir string) (*x509.CertPool, error) { entries, err := os.ReadDir(dir) if err != nil { - return nil, &Error{Kind: ErrInvalidConfig, Op: "CA directory", Err: err} + return nil, fmt.Errorf("%w: CA directory: %w", ErrInvalidConfig, err) } if len(entries) == 0 { return nil, invalidConfig("CA directory %s is empty", dir) @@ -84,7 +80,7 @@ func loadRoots(dir string) (*x509.CertPool, error) { } data, err := os.ReadFile(filepath.Join(dir, e.Name())) if err != nil { - return nil, &Error{Kind: ErrInvalidConfig, Op: "CA directory", Err: err} + return nil, fmt.Errorf("%w: CA directory: %w", ErrInvalidConfig, err) } if !roots.AppendCertsFromPEM(data) { return nil, invalidConfig("CA entry %s holds no PEM certificate", e.Name()) @@ -102,7 +98,7 @@ func admit(cfg Config, roots *x509.CertPool, s Session, openNetwork func(context req := s.Request view, err := cfg.Harnesses.ResolveView(req.AgentKind) if err != nil { - return nil, &Error{Kind: ErrUnsupported, Op: "admit", Err: err} + return nil, fmt.Errorf("%w: admit: %w", ErrUnsupported, err) } local := req.LocalEnvironment switch { @@ -145,7 +141,7 @@ func admit(cfg Config, roots *x509.CertPool, s Session, openNetwork func(context return nil, err } gw := gateway.Config{ - Models: []gateway.Model{{Name: modelName, Provider: provider}}, + Model: provider, MCP: bindings, Prompt: req, OpenNetwork: openNetwork, @@ -154,13 +150,10 @@ func admit(cfg Config, roots *x509.CertPool, s Session, openNetwork func(context } endpoints, err := gateway.Plan(gw) if err != nil { - return nil, &Error{Kind: ErrInvalidSession, Op: "gateway", Err: err} + return nil, fmt.Errorf("%w: gateway: %w", ErrInvalidSession, err) } - p := &plan{view: view, gateway: gw, proxy: endpoints.Proxy, + p := &plan{view: view, gateway: gw, request: handoff(req, provider, endpoints), proxy: endpoints.Proxy, executables: processbroker.Executables{Names: identity(view.Shims), Paths: identity(view.ShimPaths)}} - if p.request, err = handoff(req, provider, endpoints); err != nil { - return nil, err - } for _, b := range bindings { b.ServerURL, b.BearerToken, b.HTTPHeaders = endpoints.MCP[b.ServerLabel], nil, nil if b.AllowedTools != nil { @@ -175,23 +168,15 @@ func admit(cfg Config, roots *x509.CertPool, s Session, openNetwork func(context // handoff rewrites the request as a view Executor receives it: the model // provider is the gateway's listener with the placeholder key, and MCP is // only in ViewSession.MCP. -func handoff(req proto.PromptRequestPayload, provider modelprovider.Provider, endpoints gateway.Endpoints) (proto.PromptRequestPayload, error) { - provider.BaseURL, provider.APIKey = endpoints.Models[modelName], modelprovider.Placeholder - encoded, err := json.Marshal(provider) - if err != nil { - return req, invalidSession("model provider: %v", err) - } - var option map[string]any - if err := json.Unmarshal(encoded, &option); err != nil { - return req, invalidSession("model provider: %v", err) - } +func handoff(req proto.PromptRequestPayload, provider modelprovider.Provider, endpoints gateway.Endpoints) proto.PromptRequestPayload { + provider.BaseURL, provider.APIKey = endpoints.Model, modelprovider.Placeholder req.AgentOptions = maps.Clone(req.AgentOptions) - req.AgentOptions["model_provider"] = option + req.AgentOptions["model_provider"] = provider req.MCPHTTPServers = nil local := *req.LocalEnvironment local.MCP = nil req.LocalEnvironment = &local - return req, nil + return req } // checkLayout rejects a view whose overlays, masks or shim paths meet the @@ -211,7 +196,7 @@ func checkLayout(cfg Config, view agent.View) error { for _, p := range claimed { for _, q := range own { if p == q || strings.HasPrefix(p, q+"/") || strings.HasPrefix(q, p+"/") { - return &Error{Kind: ErrUnsupported, Op: "admit", Err: fmt.Errorf("%w: view path %s meets the agent host's %s", agent.ErrInvalidView, p, q)} + return fmt.Errorf("%w: admit: %w: view path %s meets the agent host's %s", ErrUnsupported, agent.ErrInvalidView, p, q) } } } @@ -267,13 +252,13 @@ func isViewPath(p string) bool { } func unsupported(format string, args ...any) error { - return &Error{Kind: ErrUnsupported, Op: "admit", Err: fmt.Errorf("%w: %s", agent.ErrUnsupportedOperation, fmt.Sprintf(format, args...))} + return fmt.Errorf("%w: admit: %w: %s", ErrUnsupported, agent.ErrUnsupportedOperation, fmt.Sprintf(format, args...)) } func invalidSession(format string, args ...any) error { - return &Error{Kind: ErrInvalidSession, Op: "admit", Err: fmt.Errorf(format, args...)} + return fmt.Errorf("%w: admit: %s", ErrInvalidSession, fmt.Sprintf(format, args...)) } func invalidConfig(format string, args ...any) error { - return &Error{Kind: ErrInvalidConfig, Err: fmt.Errorf(format, args...)} + return fmt.Errorf("%w: %s", ErrInvalidConfig, fmt.Sprintf(format, args...)) } diff --git a/apps/daemon/internal/agenthost/admit_linux_test.go b/apps/daemon/internal/agenthost/admit_linux_test.go index bd044f20d..d3572d6e5 100644 --- a/apps/daemon/internal/agenthost/admit_linux_test.go +++ b/apps/daemon/internal/agenthost/admit_linux_test.go @@ -90,7 +90,7 @@ func TestAdmissionRejectsBeforeAnyEffect(t *testing.T) { c.change(&req) s, _, _ := newSession(newResource(), req) var dials atomic.Int32 - err := run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), procs: &fakeProcesses{}}) + err := run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), tasks: noTasks}) for _, want := range c.want { if !errors.Is(err, want) { t.Errorf("%s: Run = %v, want %v", name, err, want) @@ -112,7 +112,7 @@ func TestAdmissionRejectsBeforeAnyEffect(t *testing.T) { s, _, _ := newSession(newResource(), request("viewed", "/workspace", "https://model.test", "sk-test")) change(&s.Binding) var dials atomic.Int32 - err := run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), procs: &fakeProcesses{}}) + err := run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), tasks: noTasks}) if !errors.Is(err, ErrInvalidSession) || dials.Load() != 0 { t.Errorf("%s: Run = %v after %d dials, want ErrInvalidSession", name, err, dials.Load()) } @@ -130,7 +130,7 @@ func TestViewExecutorReceivesTheGatewayRequest(t *testing.T) { original := maps.Clone(req.AgentOptions) s, _, _ := newSession(newResource(), req) var dials atomic.Int32 - err := run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), procs: &fakeProcesses{}}) + err := run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), tasks: noTasks}) if !errors.Is(err, ErrExecutor) || !errors.Is(err, errFactory) { t.Fatalf("Run = %v, want the factory's error as ErrExecutor", err) } @@ -164,7 +164,7 @@ func TestViewExecutorReceivesTheGatewayRequest(t *testing.T) { req.AgentOptions["mcp_servers"] = map[string]any{} s, _, _ = newSession(newResource(), req) f.req = proto.PromptRequestPayload{} - err = run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), procs: &fakeProcesses{}}) + err = run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), tasks: noTasks}) if !errors.Is(err, ErrUnsupported) || !errors.Is(err, agent.ErrViewHandoff) || f.req.AgentKind != "" { t.Errorf("Run with a connection option = %v, want ErrUnsupported and ErrViewHandoff before the adapter", err) } diff --git a/apps/daemon/internal/agenthost/agenthost.go b/apps/daemon/internal/agenthost/agenthost.go index 90131ce69..f352b8927 100644 --- a/apps/daemon/internal/agenthost/agenthost.go +++ b/apps/daemon/internal/agenthost/agenthost.go @@ -5,7 +5,6 @@ import ( "errors" "log/slog" "os" - "strings" "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" @@ -157,30 +156,3 @@ func (h *Host) Close() error { } return errors.Join(errs...) } - -// Error is a typed agent host failure. It matches Kind and, when present, -// Err. Its message never includes a credential. -type Error struct { - Kind error - Op string - Err error -} - -func (e *Error) Error() string { - var b strings.Builder - b.WriteString(e.Kind.Error()) - if e.Op != "" { - b.WriteString(": " + e.Op) - } - if e.Err != nil { - b.WriteString(": " + e.Err.Error()) - } - return b.String() -} - -func (e *Error) Unwrap() []error { - if e.Err == nil { - return []error{e.Kind} - } - return []error{e.Kind, e.Err} -} diff --git a/apps/daemon/internal/agenthost/agenthost_linux_test.go b/apps/daemon/internal/agenthost/agenthost_linux_test.go index 48d825887..e58e44325 100644 --- a/apps/daemon/internal/agenthost/agenthost_linux_test.go +++ b/apps/daemon/internal/agenthost/agenthost_linux_test.go @@ -100,7 +100,7 @@ func newResource() sandboxlink.ResourceRef { // countingDial counts dials and connects nothing. func countingDial(n *atomic.Int32) dialFunc { - return func(context.Context, func(sandboxlink.AttachmentClosed)) (attachLink, error) { + return func(context.Context, func(sandboxlink.AttachmentClosed)) (*sandboxlink.AttachLink, error) { n.Add(1) return nil, errors.New("no relay in this test") } @@ -116,9 +116,5 @@ func leftSessions(t *testing.T, cfg Config) []os.DirEntry { return entries } -// fakeProcesses is a process table that lists fixed tasks. -type fakeProcesses struct { - list []task -} - -func (f *fakeProcesses) tasks() ([]task, error) { return f.list, nil } +// noTasks lists no running task. +func noTasks() ([][4]uint32, error) { return nil, nil } diff --git a/apps/daemon/internal/agenthost/host_linux.go b/apps/daemon/internal/agenthost/host_linux.go index b4e35f2b8..6864c3eea 100644 --- a/apps/daemon/internal/agenthost/host_linux.go +++ b/apps/daemon/internal/agenthost/host_linux.go @@ -4,6 +4,7 @@ package agenthost import ( "errors" + "fmt" "os" "path/filepath" @@ -24,7 +25,7 @@ func Open(cfg Config) (_ *Host, err error) { return nil, invalidConfig("view cgroups %q is not the canonical path of a directory", cfg.ViewCgroups) } if err := os.MkdirAll(cfg.StateDir, 0o700); err != nil { - return nil, &Error{Kind: ErrInvalidConfig, Op: "state directory", Err: err} + return nil, fmt.Errorf("%w: state directory: %w", ErrInvalidConfig, err) } h := &Host{cfg: cfg} defer func() { @@ -43,13 +44,13 @@ func Open(cfg Config) (_ *Host, err error) { if errors.Is(err, sessionview.ErrCleanup) { kind = ErrTeardown } - return nil, &Error{Kind: kind, Op: "view cgroups", Err: err} + return nil, fmt.Errorf("%w: view cgroups: %w", kind, err) } if err := sessionview.Probe(); err != nil { - return nil, &Error{Kind: ErrUnsupported, Op: "views", Err: err} + return nil, fmt.Errorf("%w: views: %w", ErrUnsupported, err) } if err := os.RemoveAll(sessionsDir(cfg.StateDir)); err != nil { - return nil, &Error{Kind: ErrTeardown, Op: "remove session directories", Err: err} + return nil, fmt.Errorf("%w: remove session directories: %w", ErrTeardown, err) } return h, nil } @@ -59,7 +60,7 @@ func Open(cfg Config) (_ *Host, err error) { func lock(path string, flag int) (*os.File, error) { f, err := os.OpenFile(path, flag, 0o600) if err != nil { - return nil, &Error{Kind: ErrInvalidConfig, Op: "lock", Err: err} + return nil, fmt.Errorf("%w: lock: %w", ErrInvalidConfig, err) } if err := unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil { f.Close() @@ -67,7 +68,7 @@ func lock(path string, flag int) (*os.File, error) { if errors.Is(err, unix.EWOULDBLOCK) { kind = ErrStateLocked } - return nil, &Error{Kind: kind, Op: "lock " + path, Err: err} + return nil, fmt.Errorf("%w: lock %s: %w", kind, path, err) } return f, nil } diff --git a/apps/daemon/internal/agenthost/host_linux_test.go b/apps/daemon/internal/agenthost/host_linux_test.go index fe0fddcb4..4b63c31cb 100644 --- a/apps/daemon/internal/agenthost/host_linux_test.go +++ b/apps/daemon/internal/agenthost/host_linux_test.go @@ -128,7 +128,7 @@ func TestOpenRecoversWhatAnEarlierAgentHostLeft(t *testing.T) { t.Errorf("the %s process ended with %v, want SIGKILL", name, err) } } - if held, err := heldUIDs(procfs{}, UIDRange{First: cfg.UIDs.First, Count: 2}); err != nil || len(held) != 0 { + if held, err := heldUIDs(taskUIDs, UIDRange{First: cfg.UIDs.First, Count: 2}); err != nil || len(held) != 0 { t.Errorf("Session uids held after Open: %v, %v", held, err) } if left := sessionviewtest.Cgroups(t, cfg.ViewCgroups); len(left) != 0 { diff --git a/apps/daemon/internal/agenthost/launch_linux.go b/apps/daemon/internal/agenthost/launch_linux.go index 6df910e06..c48beb4a5 100644 --- a/apps/daemon/internal/agenthost/launch_linux.go +++ b/apps/daemon/internal/agenthost/launch_linux.go @@ -66,11 +66,11 @@ func (s *session) closeLive() { func (s *session) checkStart(opts clirunner.StartOptions) error { switch { case !slices.Contains(s.plan.view.LocalExec, opts.Binary): - return &Error{Kind: ErrLaunch, Err: fmt.Errorf("%w: %q", agent.ErrNotLocalExec, opts.Binary)} + return fmt.Errorf("%w: %w: %q", ErrLaunch, agent.ErrNotLocalExec, opts.Binary) case !isViewPath(opts.Dir): - return &Error{Kind: ErrLaunch, Err: fmt.Errorf("directory %q is not absolute and clean", opts.Dir)} + return fmt.Errorf("%w: directory %q is not absolute and clean", ErrLaunch, opts.Dir) case !opts.OwnProcessGroup: - return &Error{Kind: ErrLaunch, Err: errors.New("a view process runs in its own process group")} + return fmt.Errorf("%w: a view process runs in its own process group", ErrLaunch) } return nil } @@ -90,10 +90,10 @@ func (s *session) launch(opts clirunner.StartOptions) (*clirunner.Process, error switch { case s.ctx.Err() != nil: s.mu.Unlock() - return nil, &Error{Kind: ErrLaunch, Err: errors.New("the Session is ending")} + return nil, fmt.Errorf("%w: the Session is ending", ErrLaunch) case s.live != nil: s.mu.Unlock() - return nil, &Error{Kind: ErrLaunch, Err: errors.New("the Session already has a live view")} + return nil, fmt.Errorf("%w: the Session already has a live view", ErrLaunch) } lv := &liveView{} s.live = lv @@ -114,7 +114,7 @@ func (s *session) spawn(opts clirunner.StartOptions) (*clirunner.Process, error) } s.mu.Unlock() if v == nil { - return nil, &Error{Kind: ErrLaunch, Op: "spawn", Err: agent.ErrNoLiveView} + return nil, fmt.Errorf("%w: spawn: %w", ErrLaunch, agent.ErrNoLiveView) } if opts.Parent == nil { opts.Parent = context.Background() @@ -125,7 +125,7 @@ func (s *session) spawn(opts clirunner.StartOptions) (*clirunner.Process, error) if errors.Is(err, sessionview.ErrExited) || errors.Is(err, sessionview.ErrClosed) { err = fmt.Errorf("%w: %w", agent.ErrNoLiveView, err) } - return nil, &Error{Kind: ErrLaunch, Op: "spawn", Err: err} + return nil, fmt.Errorf("%w: spawn: %w", ErrLaunch, err) } var stdin io.WriteCloser if p.Stdin != nil { @@ -160,28 +160,34 @@ func (s *session) start(lv *liveView, opts clirunner.StartOptions) (*clirunner.P if scope, err = s.processScope(startCtx); err != nil { s.release(lv) if startCtx.Err() != nil { - return nil, &Error{Kind: ErrLaunch, Op: "describe", Err: err} + return nil, fmt.Errorf("%w: describe: %w", ErrLaunch, err) } return nil, s.brokerFailed("describe", err) } } if err := s.dir.chownHome(s.uid); err != nil { s.release(lv) - return nil, &Error{Kind: ErrLaunch, Op: "home", Err: err} + return nil, fmt.Errorf("%w: home: %w", ErrLaunch, err) } ends, err := newStdio(opts.NeedStdin) if err != nil { s.release(lv) - return nil, &Error{Kind: ErrLaunch, Op: "stdio", Err: err} + return nil, fmt.Errorf("%w: stdio: %w", ErrLaunch, err) } - // The gateway serves until the view has ended. - viewCtx, stopGateway := context.WithCancel(context.Background()) world := worldfs.New(worldExport, s.openFile) - spec := s.spec(viewCtx, world, opts, ends) + spec := s.spec(world, opts, ends) + // The gateway serves from the view's network hook until the view has ended. + var stopGateway func() + spec.Network.Setup = func(netns *os.File) (err error) { + stopGateway, err = gateway.Start(netns, s.plan.gateway) + return err + } v, err := sessionview.Start(startCtx, spec) ends.closeChild() if err != nil { - stopGateway() + if stopGateway != nil { + stopGateway() + } ends.closeParent() defer s.release(lv) // sessionview stops a world that served; Stop reports how that went. @@ -196,7 +202,7 @@ func (s *session) start(lv *liveView, opts clirunner.StartOptions) (*clirunner.P case left != nil: return nil, left } - return nil, &Error{Kind: ErrLaunch, Err: err} + return nil, fmt.Errorf("%w: %w", ErrLaunch, err) } p := v.Presentation() s.log.Info("agent host view started", "binary", opts.Binary, "targets", p.Targets, "links", p.Links, "synthesized", p.Synthesized) @@ -227,7 +233,7 @@ func (s *session) processScope(ctx context.Context) (sandboxprocess.Scope, error // brokerFailed fails the Session with a process broker failure. func (s *session) brokerFailed(op string, err error) error { - e := &Error{Kind: ErrProcessBroker, Op: op, Err: err} + e := fmt.Errorf("%w: %s: %w", ErrProcessBroker, op, err) s.fail(e) return e } @@ -259,14 +265,14 @@ func (s *session) own(lv *liveView, v runningView, world viewWorld, stopGateway case brokerErr != nil: err = s.brokerFailed("start", brokerErr) case closed: - err = &Error{Kind: ErrLaunch, Err: errors.New("the Session is ending")} + err = fmt.Errorf("%w: the Session is ending", ErrLaunch) } var process *clirunner.Process if err == nil { process, err = clirunner.FromHandle(h, clirunner.HandleOptions{Parent: opts.Parent, Stdin: ends.stdin(), Stdout: ends.parent[1], Stderr: ends.parent[2], KillTimeout: opts.KillTimeout}) if err != nil { - err = &Error{Kind: ErrLaunch, Err: err} + err = fmt.Errorf("%w: %w", ErrLaunch, err) } } if err != nil { @@ -307,7 +313,7 @@ func (h *ownedView) watch() { for { select { case <-h.world.Lost(): - h.s.fail(&Error{Kind: ErrWorld, Op: "world", Err: h.world.Err()}) + h.s.fail(fmt.Errorf("%w: world: %w", ErrWorld, h.world.Err())) return case <-relayEnded: // The broker stops serving on Close, which end calls only after @@ -353,14 +359,14 @@ func (h *ownedView) end(waitErr error) { <-h.watched if h.broker != nil { if err := h.broker.Close(); err != nil { - h.s.ended(&Error{Kind: ErrTeardown, Op: "close process broker", Err: err}, false) + h.s.ended(fmt.Errorf("%w: close process broker: %w", ErrTeardown, err), false) } } if errors.Is(waitErr, sessionview.ErrCleanup) { h.s.viewLeft(waitErr) } if lost := h.world.Err(); lost != nil { - h.s.fail(&Error{Kind: ErrWorld, Op: "world", Err: lost}) + h.s.fail(fmt.Errorf("%w: world: %w", ErrWorld, lost)) } // The view has stopped its world; Stop reports how that went. if err := h.world.Stop(); err != nil { @@ -370,9 +376,9 @@ func (h *ownedView) end(waitErr error) { } // spec builds the view: the closure and home directories, the agent -// host's /etc files and CA directory, the adapter's overlays and masks, the -// shim and the gateway in the view's network namespace. -func (s *session) spec(viewCtx context.Context, world *worldfs.World, opts clirunner.StartOptions, ends *stdio) sessionview.Spec { +// host's /etc files and CA directory, the adapter's overlays and masks, and +// the shim. +func (s *session) spec(world *worldfs.World, opts clirunner.StartOptions, ends *stdio) sessionview.Spec { view := s.plan.view var private []sessionview.PrivateDir for _, m := range view.Closure { @@ -402,10 +408,6 @@ func (s *session) spec(viewCtx context.Context, world *worldfs.World, opts cliru Process: sessionview.Process{Path: opts.Binary, Args: append([]string{opts.Binary}, opts.Args...), Env: opts.Env, Dir: opts.Dir, UID: s.uid, GID: s.uid, Stdin: ends.child[0], Stdout: ends.child[1], Stderr: ends.child[2], Grace: opts.KillTimeout}, - Network: sessionview.Network{Setup: func(netns *os.File) error { - _, err := gateway.Start(viewCtx, gateway.SessionNetwork{Namespace: netns}, s.plan.gateway) - return err - }}, StagingParent: s.dir.entry(stagingEntry), CgroupParent: s.cfg.ViewCgroups, } diff --git a/apps/daemon/internal/agenthost/link.go b/apps/daemon/internal/agenthost/link.go index 51ca9396a..51cd0fa37 100644 --- a/apps/daemon/internal/agenthost/link.go +++ b/apps/daemon/internal/agenthost/link.go @@ -11,20 +11,11 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" ) -// attachLink is the part of *sandboxlink.AttachLink the Session uses. -type attachLink interface { - OpenService(context.Context, sandboxlink.Open) (sandboxlink.Stream, sandboxlink.Opened, error) - Renew(context.Context, sandboxlink.RenewAttachment) (sandboxlink.AttachmentRenewed, error) - CloseAttachment(context.Context, sandboxwire.ID) error - Done() <-chan struct{} - Close() error -} - // dialFunc connects an attach link; onClosed is its OnAttachmentClosed. -type dialFunc func(ctx context.Context, onClosed func(sandboxlink.AttachmentClosed)) (attachLink, error) +type dialFunc func(ctx context.Context, onClosed func(sandboxlink.AttachmentClosed)) (*sandboxlink.AttachLink, error) func relayDial(cfg Config) dialFunc { - return func(ctx context.Context, onClosed func(sandboxlink.AttachmentClosed)) (attachLink, error) { + return func(ctx context.Context, onClosed func(sandboxlink.AttachmentClosed)) (*sandboxlink.AttachLink, error) { return sandboxlink.DialAttach(ctx, sandboxlink.AttachConfig{URL: cfg.RelayURL, TLS: cfg.TLS, RuntimeID: cfg.RuntimeID, Credential: cfg.Credential, OnAttachmentClosed: onClosed}) } @@ -50,7 +41,7 @@ type linkOwner struct { dialMu sync.Mutex // serializes dials mu sync.Mutex - link attachLink + link *sandboxlink.AttachLink // instance is the service instance of the first Opened; zero before. instance sandboxwire.ID lease time.Time @@ -73,7 +64,7 @@ func (b Binding) open(service sandboxlink.Service, version uint16, expected sand } // current returns the live link, dialing a new one when there is none. -func (l *linkOwner) current(ctx context.Context) (attachLink, error) { +func (l *linkOwner) current(ctx context.Context) (*sandboxlink.AttachLink, error) { l.dialMu.Lock() defer l.dialMu.Unlock() l.mu.Lock() @@ -103,7 +94,7 @@ func (l *linkOwner) open(ctx context.Context, service sandboxlink.Service, versi closing := l.closing l.mu.Unlock() if closing { - return nil, &Error{Kind: ErrLink, Op: "open " + service.String(), Err: errors.New("the Session is ending")} + return nil, fmt.Errorf("%w: open %s: the Session is ending", ErrLink, service) } link, err := l.current(ctx) if err != nil { @@ -112,7 +103,7 @@ func (l *linkOwner) open(ctx context.Context, service sandboxlink.Service, versi l.mu.Lock() if l.closing { l.mu.Unlock() - return nil, &Error{Kind: ErrLink, Op: "open " + service.String(), Err: errors.New("the Session is ending")} + return nil, fmt.Errorf("%w: open %s: the Session is ending", ErrLink, service) } l.opened = true expected := l.instance @@ -140,7 +131,7 @@ func (l *linkOwner) open(ctx context.Context, service sandboxlink.Service, versi // observe fails the Session on a Link failure that is not retryable and // returns err as a typed error. func (l *linkOwner) observe(op string, err error) error { - err = &Error{Kind: ErrLink, Op: op, Err: err} + err = fmt.Errorf("%w: %s: %w", ErrLink, op, err) if !retryable(err) { l.report(err) } @@ -167,7 +158,7 @@ func retryable(err error) bool { // closed is the link's OnAttachmentClosed. It never blocks. func (l *linkOwner) closed(c sandboxlink.AttachmentClosed) { if c.AttachmentID == l.binding.AttachmentID { - l.report(&Error{Kind: ErrLink, Op: "attachment", Err: fmt.Errorf("the relay closed the attachment (reason %d)", c.Reason)}) + l.report(fmt.Errorf("%w: attachment: the relay closed the attachment (reason %d)", ErrLink, c.Reason)) } } @@ -188,7 +179,7 @@ func (l *linkOwner) renew(ctx context.Context) { } for { if !time.Now().Before(lease) { - l.report(&Error{Kind: ErrLink, Op: "renew", Err: sandboxlink.LeaseExpired}) + l.report(fmt.Errorf("%w: renew: %w", ErrLink, sandboxlink.LeaseExpired)) return } attempt, cancel := context.WithDeadline(ctx, lease) @@ -238,7 +229,7 @@ func (l *linkOwner) close() error { if opened { ctx, cancel := context.WithTimeout(context.Background(), closeBound) for { - var link attachLink + var link *sandboxlink.AttachLink if link, err = l.current(ctx); err == nil { err = link.CloseAttachment(ctx, l.binding.AttachmentID) } @@ -256,7 +247,7 @@ func (l *linkOwner) close() error { link.Close() } if err != nil { - return &Error{Kind: ErrTeardown, Op: "close attachment", Err: err} + return fmt.Errorf("%w: close attachment: %w", ErrTeardown, err) } return nil } diff --git a/apps/daemon/internal/agenthost/procs_linux.go b/apps/daemon/internal/agenthost/procs_linux.go index fe7ba3504..f6cef95c5 100644 --- a/apps/daemon/internal/agenthost/procs_linux.go +++ b/apps/daemon/internal/agenthost/procs_linux.go @@ -15,30 +15,22 @@ import ( "golang.org/x/sys/unix" ) -// task is one running task, a thread of a process, as /proc lists it. -type task struct { - tgid, tid int - uids [4]uint32 // real, effective, saved and file-system -} - func (r UIDRange) has(id uint32) bool { return id >= r.First && id-r.First < r.Count } -// processTable lists the host's tasks. Tests replace it. -type processTable interface { - // tasks returns every task that runs, each thread of each process; a - // zombie runs nothing and is left out. - tasks() ([]task, error) -} +// listTasks returns the real, effective, saved and file-system uids of every +// task that runs, each thread of each process; a zombie runs nothing and is +// left out. taskUIDs lists the host's; tests list fixed tasks. +type listTasks func() ([][4]uint32, error) // heldUIDs returns the uids in r that a running task holds. -func heldUIDs(procs processTable, r UIDRange) (map[uint32]bool, error) { - tasks, err := procs.tasks() +func heldUIDs(tasks listTasks, r UIDRange) (map[uint32]bool, error) { + list, err := tasks() if err != nil { return nil, err } held := map[uint32]bool{} - for _, t := range tasks { - for _, id := range t.uids { + for _, uids := range list { + for _, id := range uids { if r.has(id) { held[id] = true } @@ -47,18 +39,16 @@ func heldUIDs(procs processTable, r UIDRange) (map[uint32]bool, error) { return held, nil } -// procfs is the host's /proc. -type procfs struct{} - // errGone is a process or task that has ended. var errGone = errors.New("ended") -func (procfs) tasks() ([]task, error) { +// taskUIDs is the listTasks of the host's /proc. +func taskUIDs() ([][4]uint32, error) { pids, err := os.ReadDir("/proc") if err != nil { return nil, err } - var list []task + var list [][4]uint32 for _, p := range pids { pid, err := strconv.Atoi(p.Name()) if err != nil || pid <= 0 { @@ -72,10 +62,6 @@ func (procfs) tasks() ([]task, error) { return nil, err } for _, e := range tids { - tid, err := strconv.Atoi(e.Name()) - if err != nil { - continue - } s, err := readStatus(procPath(pid, "task", e.Name(), "status")) if errors.Is(err, errGone) { continue @@ -84,7 +70,7 @@ func (procfs) tasks() ([]task, error) { return nil, err } if s.running() { - list = append(list, task{tgid: pid, tid: tid, uids: s.uids}) + list = append(list, s.uids) } } } diff --git a/apps/daemon/internal/agenthost/run_linux.go b/apps/daemon/internal/agenthost/run_linux.go index 7f8bd4041..e00a7b2f9 100644 --- a/apps/daemon/internal/agenthost/run_linux.go +++ b/apps/daemon/internal/agenthost/run_linux.go @@ -31,7 +31,7 @@ const ( // deps are the parts tests replace. type deps struct { dial dialFunc - procs processTable + tasks listTasks } // Run runs one Session until Input is closed, ctx ends or the Session fails, @@ -40,7 +40,7 @@ type deps struct { // that ended it, joined with any view cleanup and teardown failure. A failure // recorded during teardown counts. Sessions may run concurrently. func (h *Host) Run(ctx context.Context, s Session) error { - return run(ctx, h.cfg, s, deps{dial: relayDial(h.cfg), procs: procfs{}}) + return run(ctx, h.cfg, s, deps{dial: relayDial(h.cfg), tasks: taskUIDs}) } // session is one running Session. @@ -90,7 +90,7 @@ func run(ctx context.Context, cfg Config, in Session, d deps) error { if s.plan, err = admit(cfg, roots, in, s.openNetwork); err != nil { return err } - if s.uid, err = allocUID(cfg.UIDs, d.procs); err != nil { + if s.uid, err = allocUID(cfg.UIDs, d.tasks); err != nil { return err } if s.dir, err = createSessionDir(cfg.StateDir, in.Binding.SessionID, s.uid); err != nil { @@ -140,9 +140,9 @@ func (s *session) finish(exec agent.Executor, err, ended error) error { func executorError(err error) error { if errors.Is(err, agent.ErrUnsupportedOperation) || errors.Is(err, agent.ErrViewHandoff) { - return &Error{Kind: ErrUnsupported, Op: "executor", Err: err} + return fmt.Errorf("%w: executor: %w", ErrUnsupported, err) } - return &Error{Kind: ErrExecutor, Err: err} + return fmt.Errorf("%w: %w", ErrExecutor, err) } // fail records the Session's first failure and ends the Session: its live @@ -161,7 +161,7 @@ func (s *session) fail(err error) { // state, so the Session fails, and Run reports the error even after another // failure. func (s *session) worldEnded(op string, err error) error { - return s.ended(&Error{Kind: ErrWorld, Op: op, Err: err}, false) + return s.ended(fmt.Errorf("%w: %s: %w", ErrWorld, op, err), false) } // viewLeft records a view whose teardown did not finish within sessionview's @@ -169,12 +169,12 @@ func (s *session) worldEnded(op string, err error) error { // Session directory. The Session fails, Run reports the error even after // another failure, and teardown keeps the directory and the uid. func (s *session) viewLeft(err error) error { - return s.ended(&Error{Kind: ErrTeardown, Op: "view", Err: err}, true) + return s.ended(fmt.Errorf("%w: view: %w", ErrTeardown, err), true) } // ended records e, a resource that did not stop cleanly, and fails the // Session with it. left says that the Session directory may still be in use. -func (s *session) ended(e *Error, left bool) error { +func (s *session) ended(e error, left bool) error { s.failMu.Lock() s.cleanup = append(s.cleanup, e) s.left = s.left || left @@ -310,7 +310,7 @@ func (s *session) turn(exec agent.Executor, in Input) error { if startErr == nil { startErr = errors.New("no Turn") } - return &Error{Kind: ErrTurn, Op: "start", Err: startErr} + return fmt.Errorf("%w: start: %w", ErrTurn, startErr) } if startErr != nil { f.cancel() @@ -335,7 +335,7 @@ func (s *session) turn(exec agent.Executor, in Input) error { f.mu.Unlock() if nativeErr != nil || !settlement.Reusable || startErr != nil || protocolErr != nil || s.ctx.Err() != nil { if err := s.closeExecutor(exec); err != nil { - return &Error{Kind: ErrTurn, Op: "close executor", Err: errors.Join(nativeErr, err)} + return fmt.Errorf("%w: close executor: %w", ErrTurn, errors.Join(nativeErr, err)) } } // A confirmed Close confirms that out is closed too. What arrived during @@ -351,13 +351,13 @@ func (s *session) turn(exec agent.Executor, in Input) error { var result error switch { case nativeErr != nil: - failure, result = "executor Turn settlement failed", &Error{Kind: ErrTurn, Op: "settle", Err: nativeErr} + failure, result = "executor Turn settlement failed", fmt.Errorf("%w: settle: %w", ErrTurn, nativeErr) case protocolErr != nil: - failure, result = protocolErr.Error(), &Error{Kind: ErrTurn, Op: "output", Err: protocolErr} + failure, result = protocolErr.Error(), fmt.Errorf("%w: output: %w", ErrTurn, protocolErr) case startErr != nil: - failure, result = "executor Turn could not start", &Error{Kind: ErrTurn, Op: "start", Err: startErr} + failure, result = "executor Turn could not start", fmt.Errorf("%w: start: %w", ErrTurn, startErr) case !settlement.Reusable: - result = &Error{Kind: ErrTurn, Op: "settle", Err: fmt.Errorf("the Executor is not reusable: %s", settlement.Reason)} + result = fmt.Errorf("%w: settle: the Executor is not reusable: %s", ErrTurn, settlement.Reason) } if failure != "" { e, err := proto.NewEnvelope(proto.TypeError, in.RunID, proto.ErrorPayload{Error: failure}) @@ -459,7 +459,7 @@ func (s *session) teardown(exec agent.Executor) error { } errs = append(errs, s.link.close()) if closeErr != nil { - errs = append(errs, &Error{Kind: ErrTeardown, Op: "close executor", Err: closeErr}) + errs = append(errs, fmt.Errorf("%w: close executor: %w", ErrTeardown, closeErr)) } s.failMu.Lock() left := s.left @@ -468,7 +468,7 @@ func (s *session) teardown(exec agent.Executor) error { return errors.Join(errs...) } if err := os.RemoveAll(string(s.dir)); err != nil { - errs = append(errs, &Error{Kind: ErrTeardown, Op: "remove session directory", Err: err}) + errs = append(errs, fmt.Errorf("%w: remove session directory: %w", ErrTeardown, err)) } freeUID(s.uid) return errors.Join(errs...) diff --git a/apps/daemon/internal/agenthost/run_other.go b/apps/daemon/internal/agenthost/run_other.go index 8486ceabd..5c5a47eef 100644 --- a/apps/daemon/internal/agenthost/run_other.go +++ b/apps/daemon/internal/agenthost/run_other.go @@ -2,14 +2,17 @@ package agenthost -import "context" +import ( + "context" + "fmt" +) // Open reports that the agent host needs Linux. func Open(Config) (*Host, error) { - return nil, &Error{Kind: ErrUnsupported, Op: "open"} + return nil, fmt.Errorf("%w: open", ErrUnsupported) } // Run reports that the agent host needs Linux. func (*Host) Run(context.Context, Session) error { - return &Error{Kind: ErrUnsupported, Op: "run"} + return fmt.Errorf("%w: run", ErrUnsupported) } diff --git a/apps/daemon/internal/agenthost/session_linux_test.go b/apps/daemon/internal/agenthost/session_linux_test.go index b77848bb0..06bcb99dc 100644 --- a/apps/daemon/internal/agenthost/session_linux_test.go +++ b/apps/daemon/internal/agenthost/session_linux_test.go @@ -6,8 +6,10 @@ import ( "context" "errors" "log/slog" + "net" "os" "reflect" + "strings" "sync" "sync/atomic" "syscall" @@ -16,8 +18,10 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent/clirunner" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/gateway" "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/sessionview" "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "github.com/MiniMax-AI/OpenAgentCore/internal/modelprovider" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" ) @@ -25,18 +29,20 @@ import ( func TestAllocationSkipsUIDsThatProcessesHold(t *testing.T) { r := UIDRange{First: 71000, Count: 2} // A thread holds the uid; its process's leader does not. - procs := &fakeProcesses{list: []task{{tgid: 10, tid: 10, uids: [4]uint32{1000, 1000, 1000, 1000}}, {tgid: 10, tid: 11, uids: [4]uint32{1000, 71000, 1000, 1000}}}} - id, err := allocUID(r, procs) + tasks := func() ([][4]uint32, error) { + return [][4]uint32{{1000, 1000, 1000, 1000}, {1000, 71000, 1000, 1000}}, nil + } + id, err := allocUID(r, tasks) if err != nil || id != 71001 { t.Fatalf("allocUID = %d, %v; want 71001", id, err) } defer freeUID(id) - if _, err := allocUID(r, procs); !errors.Is(err, ErrCapacity) { + if _, err := allocUID(r, tasks); !errors.Is(err, ErrCapacity) { t.Fatalf("allocUID with every uid taken = %v", err) } // The host's table shows this process with its own uids. self := uint32(os.Getuid()) - if held, err := heldUIDs(procfs{}, UIDRange{First: self, Count: 1}); err != nil || !held[self] { + if held, err := heldUIDs(taskUIDs, UIDRange{First: self, Count: 1}); err != nil || !held[self] { t.Fatalf("/proc shows uid %d held: %v, %v", self, held[self], err) } } @@ -77,6 +83,55 @@ func TestViewEndReleasesTheSlotBeforeTheProcessEnds(t *testing.T) { } } +// TestViewEndStopsTheGateway checks that the gateway's listeners are closed +// once Process.Wait returns. It serves the gateway in the test's own network +// namespace, so it runs with the view suite. +func TestViewEndStopsTheGateway(t *testing.T) { + if os.Getenv(gateEnv) != "1" { + t.Skipf("set %s=1 and run the test binary as root in a privileged container; see view_linux_test.go", gateEnv) + } + ns, err := os.Open("/proc/self/ns/net") + if err != nil { + t.Fatal(err) + } + defer ns.Close() + cfg := gateway.Config{Model: modelprovider.Provider{Protocol: modelprovider.Anthropic, BaseURL: "https://model.test", APIKey: upstreamKey}} + eps, err := gateway.Plan(cfg) + if err != nil { + t.Fatal(err) + } + stop, err := gateway.Start(ns, cfg) + if err != nil { + t.Fatal(err) + } + s := newOwnerSession(t) + lv := &liveView{} + s.live = lv + s.views.Add(1) + ends, err := newStdio(false) + if err != nil { + t.Fatal(err) + } + ends.closeChild() + v := &fakeView{exit: make(chan struct{})} + p, err := s.own(lv, v, fakeWorld{}, stop, clirunner.StartOptions{Parent: context.Background(), KillTimeout: time.Second}, ends, 0) + if err != nil { + t.Fatal(err) + } + model := strings.TrimPrefix(eps.Model, "http://") + c, err := net.Dial("tcp", model) + if err != nil { + t.Fatalf("the gateway does not serve: %v", err) + } + c.Close() + v.Close() + _ = p.Wait() + if c, err := net.Dial("tcp", model); err == nil { + c.Close() + t.Fatal("the model listener accepts after Process.Wait") + } +} + // TestSpawnKeepsItsErrors checks that only a view that has ended makes a Spawn fail with ErrNoLiveView, and that every other failure keeps its own error. func TestSpawnKeepsItsErrors(t *testing.T) { emfile := &sessionview.Error{Kind: sessionview.ErrLauncher, Op: "pipe", Err: syscall.EMFILE} @@ -201,7 +256,7 @@ func TestFailedCloseKeepsTheSessionDirectoryAndUID(t *testing.T) { output := make(chan proto.Envelope, 4) s.in.Output = output // Each case takes its own uid. - uid, err := allocUID(UIDRange{First: 72000, Count: 2}, &fakeProcesses{}) + uid, err := allocUID(UIDRange{First: 72000, Count: 2}, noTasks) if err != nil { t.Fatal(err) } diff --git a/apps/daemon/internal/agenthost/sessiondir_linux.go b/apps/daemon/internal/agenthost/sessiondir_linux.go index 1602e63f1..6ecb6b601 100644 --- a/apps/daemon/internal/agenthost/sessiondir_linux.go +++ b/apps/daemon/internal/agenthost/sessiondir_linux.go @@ -22,10 +22,10 @@ var uids = struct { }{used: map[uint32]bool{}} // allocUID returns a uid in r that no Session uses and no process holds. -func allocUID(r UIDRange, procs processTable) (uint32, error) { - held, err := heldUIDs(procs, r) +func allocUID(r UIDRange, tasks listTasks) (uint32, error) { + held, err := heldUIDs(tasks, r) if err != nil { - return 0, &Error{Kind: ErrInvalidConfig, Op: "processes", Err: err} + return 0, fmt.Errorf("%w: processes: %w", ErrInvalidConfig, err) } uids.Lock() defer uids.Unlock() @@ -35,7 +35,7 @@ func allocUID(r UIDRange, procs processTable) (uint32, error) { return id, nil } } - return 0, &Error{Kind: ErrCapacity} + return 0, ErrCapacity } func freeUID(id uint32) { @@ -66,17 +66,17 @@ func (d sessionDir) entry(name ...string) string { func createSessionDir(stateDir string, id sandboxwire.ID, uid uint32) (sessionDir, error) { parent := sessionsDir(stateDir) if err := os.MkdirAll(parent, 0o700); err != nil { - return "", &Error{Kind: ErrInvalidConfig, Op: "state directory", Err: err} + return "", fmt.Errorf("%w: state directory: %w", ErrInvalidConfig, err) } d := sessionDir(filepath.Join(parent, id.String())) if err := os.Mkdir(string(d), 0o700); errors.Is(err, fs.ErrExist) { - return "", &Error{Kind: ErrSessionExists, Err: err} + return "", fmt.Errorf("%w: %w", ErrSessionExists, err) } else if err != nil { - return "", &Error{Kind: ErrInvalidConfig, Op: "session directory", Err: err} + return "", fmt.Errorf("%w: session directory: %w", ErrInvalidConfig, err) } if err := d.populate(uid); err != nil { os.RemoveAll(string(d)) - return "", &Error{Kind: ErrInvalidConfig, Op: "session directory", Err: err} + return "", fmt.Errorf("%w: session directory: %w", ErrInvalidConfig, err) } return d, nil } diff --git a/apps/daemon/internal/agenthost/view_linux_test.go b/apps/daemon/internal/agenthost/view_linux_test.go index 306e14c5f..ec3ad3454 100644 --- a/apps/daemon/internal/agenthost/view_linux_test.go +++ b/apps/daemon/internal/agenthost/view_linux_test.go @@ -252,7 +252,7 @@ func TestSessionRunsInAViewOverItsAttachment(t *testing.T) { cmd.Wait() }() until(t, "a zombie leader with a running thread", func() bool { return zombieLeaderHolds(id) }) - if got, err := allocUID(UIDRange{First: id, Count: 1}, procfs{}); !errors.Is(err, ErrCapacity) { + if got, err := allocUID(UIDRange{First: id, Count: 1}, taskUIDs); !errors.Is(err, ErrCapacity) { freeUID(got) t.Errorf("allocUID beside a running thread = %d, %v", got, err) } @@ -541,7 +541,7 @@ func checkReleased(t *testing.T, cfg Config) { if strings.Contains(string(mounts), cfg.StateDir) { t.Error("a mount under the state directory remains") } - if held, err := heldUIDs(procfs{}, cfg.UIDs); err != nil || len(held) != 0 { + if held, err := heldUIDs(taskUIDs, cfg.UIDs); err != nil || len(held) != 0 { t.Errorf("Session uids held: %v, %v", held, err) } } diff --git a/apps/daemon/internal/gateway/conns.go b/apps/daemon/internal/gateway/conns.go index ef6a555ce..39585fe6c 100644 --- a/apps/daemon/internal/gateway/conns.go +++ b/apps/daemon/internal/gateway/conns.go @@ -5,31 +5,88 @@ import ( "sync" ) -// sessionConns are the Harness's open connections to the gateway, hijacked -// and upgraded ones included. http.Server.Close does not close a hijacked -// connection, so the end of the Session aborts them here. -type sessionConns struct { - mu sync.Mutex - open map[*sessionConn]struct{} - ended bool +// conn is a connection the gateway accepts or makes: a TCP connection, or a +// connection through the sandbox. CloseWrite ends its write side in order and +// Reset aborts both sides. +type conn interface { + net.Conn + CloseWrite() error + Reset() error } -// abort resets every open connection and refuses later ones. -func (s *sessionConns) abort() { +// tcpConn is a TCP connection whose Reset sends a TCP reset. +type tcpConn struct{ *net.TCPConn } + +func (c tcpConn) Reset() error { + c.SetLinger(0) + return c.Close() +} + +// connSet holds the gateway's open connections in both directions, hijacked, +// upgraded, tunnelled and HTTP/2 ones included, which net/http neither tracks +// nor closes while they are in use, so that closeAll ends every one. +type connSet struct { + mu sync.Mutex + open map[*trackedConn]struct{} + closed bool +} + +// add records c, or resets it and returns net.ErrClosed once the set is +// closed. +func (s *connSet) add(c conn) (conn, error) { + t := &trackedConn{conn: c, set: s} + s.mu.Lock() + closed := s.closed + if !closed { + s.open[t] = struct{}{} + } + s.mu.Unlock() + if closed { + c.Reset() + return nil, net.ErrClosed + } + return t, nil +} + +// closeAll closes the set and resets each connection in it. It returns once +// every one has closed, those already closing included. +func (s *connSet) closeAll() { s.mu.Lock() open := s.open - s.open, s.ended = nil, true + s.open, s.closed = nil, true s.mu.Unlock() for c := range open { - c.SetLinger(0) - c.TCPConn.Close() + c.Reset() } } +// trackedConn is a connection in a connSet. It leaves the set once it has +// closed, and a concurrent Close or Reset returns only then. +type trackedConn struct { + conn + set *connSet + once sync.Once +} + +func (c *trackedConn) Close() error { return c.end(c.conn.Close) } + +func (c *trackedConn) Reset() error { return c.end(c.conn.Reset) } + +func (c *trackedConn) end(close func() error) error { + err := net.ErrClosed + c.once.Do(func() { + err = close() + c.set.mu.Lock() + delete(c.set.open, c) + c.set.mu.Unlock() + }) + return err +} + // sessionListener records each connection it accepts in conns. type sessionListener struct { *net.TCPListener - conns *sessionConns + conns *connSet } func (l sessionListener) Accept() (net.Conn, error) { @@ -37,30 +94,5 @@ func (l sessionListener) Accept() (net.Conn, error) { if err != nil { return nil, err } - c := &sessionConn{TCPConn: tc, conns: l.conns} - l.conns.mu.Lock() - ended := l.conns.ended - if !ended { - l.conns.open[c] = struct{}{} - } - l.conns.mu.Unlock() - if ended { - tc.SetLinger(0) - tc.Close() - return nil, net.ErrClosed - } - return c, nil -} - -// sessionConn is an accepted connection that leaves conns when it closes. -type sessionConn struct { - *net.TCPConn - conns *sessionConns -} - -func (c *sessionConn) Close() error { - c.conns.mu.Lock() - delete(c.conns.open, c) - c.conns.mu.Unlock() - return c.TCPConn.Close() + return l.conns.add(tcpConn{tc}) } diff --git a/apps/daemon/internal/gateway/gateway.go b/apps/daemon/internal/gateway/gateway.go index 702cb553c..3e8a1891e 100644 --- a/apps/daemon/internal/gateway/gateway.go +++ b/apps/daemon/internal/gateway/gateway.go @@ -1,11 +1,11 @@ // Package gateway is the Session gateway on the agent host. Inside the -// Session's loopback-only network namespace it serves one listener per frozen -// model upstream, one per MCP HTTP binding and, when the view has one, a +// Session's loopback-only network namespace it serves one listener for the +// frozen model upstream, one per MCP HTTP binding and, when the view has one, a // generic proxy, so the Harness never holds an upstream credential and has no // network route of its own. // // A listener's identity selects its upstream and credential; nothing is routed -// by hostname. A model listener relays the declared native routes of its +// by hostname. The model listener relays the declared native routes of its // protocol (internal/modelprovider) to the upstream from the agent host and // injects the credential. An MCP listener relays to its binding's server and // injects the binding's bearer token and HTTP headers: an environment-origin @@ -18,7 +18,7 @@ // through the sandbox's Network service. Redirects reach the Harness unchanged // and are never followed. Response header and trailer values that contain an // injected credential, header value or MCP query value are withheld; bodies -// pass unchanged. The end of the Session closes every connection, tunnels and +// pass unchanged. Stopping the gateway closes every connection, tunnels and // upgraded ones included. The gateway logs nothing. package gateway @@ -33,6 +33,7 @@ import ( "net/http" "os" "strconv" + "sync" "time" "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" @@ -44,9 +45,8 @@ import ( // Config is what one Session's gateway serves. It holds credentials: keep it // in memory and never log it. type Config struct { - // Models are the frozen model upstreams, each under the adapter's name for - // it. Names are unique. - Models []Model + // Model is the Session's frozen model upstream. + Model modelprovider.Provider // MCP are the Session's effective MCP bindings as // agent.ResolveMCPBindings returns them, bearer tokens and HTTP headers // included. Each is an HTTP binding, and server labels are unique. @@ -69,22 +69,12 @@ type Config struct { Proxy bool } -// Model is one frozen model upstream. -type Model struct { - Name string - Provider modelprovider.Provider -} - // Endpoints is what the Harness is given in place of upstreams and // credentials. Every URL is plain HTTP on the Session's loopback. type Endpoints struct { - // Placeholder is the credential a Harness sends to a model listener. It - // is not secret; the listener removes it. - Placeholder string - // Models maps each model upstream's name to its listener's base URL, - // http://127.0.0.1:, with no path. The listener adds the frozen - // base URL's path. - Models map[string]string + // Model is the model listener's base URL, http://127.0.0.1:, with + // no path. The listener adds the frozen base URL's path. + Model string // MCP maps each binding's server label to the URL the Harness uses: its // listener with the server URL's path and no query. MCP map[string]string @@ -93,14 +83,8 @@ type Endpoints struct { Proxy string } -// SessionNetwork is the Session's network namespace: the file sessionview's -// network hook receives. Start uses it only while it runs. -type SessionNetwork struct { - Namespace *os.File -} - // ProxyPort is the generic proxy's port in the Session's namespace. The model -// listeners take the following ports in Config order, then the MCP listeners, +// listener takes the following port, then the MCP listeners in Config order, // whether or not the proxy is served. // The namespace is the Session's own and the gateway listens before the // Harness starts, so the ports are free; fixing them lets the Harness's @@ -112,7 +96,7 @@ const maxListeners = 256 var ( // ErrInvalidConfig is a Config that Plan and Start reject. The message - // names the item by its name or label and never includes a credential. + // names the item by its label and never includes a credential. ErrInvalidConfig = errors.New("gateway: invalid configuration") // ErrNetwork is a failure to listen in the Session's network namespace. ErrNetwork = errors.New("gateway: session network") @@ -130,26 +114,23 @@ func Plan(cfg Config) (Endpoints, error) { return g.endpoints(g.fixedPorts()), nil } -// Start validates cfg, opens its listeners inside the Session's network -// namespace and serves them from the daemon until ctx ends. It returns the -// same Endpoints as Plan(cfg). It is meant to run in sessionview's network -// hook; nothing listens outside the namespace. The end of ctx closes the -// listeners and every connection. -func Start(ctx context.Context, n SessionNetwork, cfg Config) (Endpoints, error) { +// Start validates cfg, opens its listeners inside the network namespace ns at +// the Endpoints of Plan(cfg) and serves them from the daemon until stop is +// called. It is meant to run in sessionview's network hook; nothing listens +// outside the namespace. stop ends every request, and closes the listeners +// and every connection before it returns. +func Start(ns *os.File, cfg Config) (stop func(), err error) { + ctx, cancel := context.WithCancel(context.Background()) g, err := build(ctx, cfg) - if err != nil { - return Endpoints{}, err - } - if n.Namespace == nil { - return Endpoints{}, fmt.Errorf("%w: no namespace", ErrNetwork) + var lns []*net.TCPListener + if err == nil { + lns, err = listen(ns, g.fixedPorts()) } - ports := g.fixedPorts() - lns, err := listen(n.Namespace, ports) if err != nil { - return Endpoints{}, err + cancel() + return nil, err } - g.serve(ctx, lns) - return g.endpoints(ports), nil + return g.serve(ctx, cancel, lns), nil } // fixedPorts returns each listener's port: ProxyPort for the proxy, then the @@ -178,14 +159,14 @@ const ( // addresses it. type listener struct { role role - name string // model name or MCP server label + label string // MCP: the server label suffix string // MCP: the server URL's path handler http.Handler } type gateway struct { - listeners []listener // the proxy when served, then models, then MCP, in Config order - transports []*http.Transport + listeners []listener // the proxy when served, the model, then MCP in Config order + conns *connSet // every connection, accepted or dialed } func invalid(format string, args ...any) error { @@ -193,37 +174,29 @@ func invalid(format string, args ...any) error { } // build validates cfg and makes each listener's handler. Every upstream dial -// ends with session. +// ends with session, and every upstream connection joins g.conns. func build(session context.Context, cfg Config) (*gateway, error) { - if n := 1 + len(cfg.Models) + len(cfg.MCP); n > maxListeners { + if n := 2 + len(cfg.MCP); n > maxListeners { return nil, invalid("%d listeners, at most %d", n, maxListeners) } - host := relayTransport(cfg.RootCAs, sessionDial(session, (&net.Dialer{Timeout: dialTimeout, KeepAlive: 30 * time.Second}).DialContext)) - g := &gateway{transports: []*http.Transport{host}} - // The proxy forwards on its own transport; environment-origin MCP relays - // use a relay transport. + g := &gateway{conns: &connSet{open: map[*trackedConn]struct{}{}}} + host := relayTransport(cfg.RootCAs, sessionDial(session, g.conns, hostDial)) + // The proxy tunnels with the sandbox dial and forwards on its own + // transport; environment-origin MCP relays use a relay transport. + var dial dialFunc var forward, sandbox *http.Transport if cfg.OpenNetwork != nil { - dial := sessionDial(session, sandboxDialer(cfg.OpenNetwork)) + dial = sessionDial(session, g.conns, sandboxDialer(cfg.OpenNetwork)) forward, sandbox = newTransport(cfg.RootCAs, dial), relayTransport(cfg.RootCAs, dial) - g.transports = append(g.transports, forward, sandbox) } if cfg.Proxy { - g.listeners = append(g.listeners, listener{role: roleProxy, handler: newProxy(cfg.OpenNetwork, forward)}) + g.listeners = append(g.listeners, listener{role: roleProxy, handler: newProxy(dial, forward)}) } - - names := map[string]bool{} - for _, m := range cfg.Models { - if m.Name == "" || names[m.Name] { - return nil, invalid("model upstream name %q is empty or repeated", m.Name) - } - names[m.Name] = true - h, err := newModelRelay(m.Provider, host) - if err != nil { - return nil, invalid("model upstream %q: %v", m.Name, err) - } - g.listeners = append(g.listeners, listener{role: roleModel, name: m.Name, handler: h}) + model, err := newModelRelay(cfg.Model, host) + if err != nil { + return nil, invalid("model upstream: %v", err) } + g.listeners = append(g.listeners, listener{role: roleModel, handler: model}) labels := map[string]bool{} for _, b := range cfg.MCP { @@ -248,22 +221,22 @@ func build(session context.Context, cfg Config) (*gateway, error) { if err != nil { return nil, invalid("MCP server %q: %v", b.ServerLabel, err) } - g.listeners = append(g.listeners, listener{role: roleMCP, name: b.ServerLabel, suffix: suffix, handler: h}) + g.listeners = append(g.listeners, listener{role: roleMCP, label: b.ServerLabel, suffix: suffix, handler: h}) } return g, nil } func (g *gateway) endpoints(ports []int) Endpoints { - e := Endpoints{Placeholder: modelprovider.Placeholder, Models: map[string]string{}, MCP: map[string]string{}} + e := Endpoints{MCP: map[string]string{}} for i, l := range g.listeners { base := "http://" + net.JoinHostPort("127.0.0.1", strconv.Itoa(ports[i])) switch l.role { case roleProxy: e.Proxy = base case roleModel: - e.Models[l.name] = base + e.Model = base case roleMCP: - e.MCP[l.name] = base + l.suffix + e.MCP[l.label] = base + l.suffix } } return e @@ -273,12 +246,13 @@ func (g *gateway) endpoints(ports []int) Endpoints { // could carry more than the gateway chooses to reveal. var quiet = log.New(io.Discard, "", 0) -// serve serves each listener with its handler until ctx ends. The end of ctx -// cancels every request, which closes its upstream side, and aborts every -// connection the listeners accepted, which ends a relay blocked on a Harness -// that does not read. -func (g *gateway) serve(ctx context.Context, lns []*net.TCPListener) { - conns := &sessionConns{open: map[*sessionConn]struct{}{}} +// serve serves each listener with its handler and returns its stop. stop +// cancels ctx, which ends every request and dial, closes the listeners, waits +// until none is accepting and then resets every connection either way, which +// ends a relay blocked on a peer that does not read. It returns once all of +// them have closed. +func (g *gateway) serve(ctx context.Context, cancel context.CancelFunc, lns []*net.TCPListener) (stop func()) { + var accepting sync.WaitGroup for i, ln := range lns { srv := &http.Server{ Handler: g.listeners[i].handler, @@ -286,13 +260,14 @@ func (g *gateway) serve(ctx context.Context, lns []*net.TCPListener) { ErrorLog: quiet, BaseContext: func(net.Listener) context.Context { return ctx }, } - go srv.Serve(sessionListener{TCPListener: ln, conns: conns}) - context.AfterFunc(ctx, func() { srv.Close() }) + accepting.Go(func() { srv.Serve(sessionListener{TCPListener: ln, conns: g.conns}) }) } - context.AfterFunc(ctx, func() { - conns.abort() - for _, t := range g.transports { - t.CloseIdleConnections() + return func() { + cancel() + for _, ln := range lns { + ln.Close() } - }) + accepting.Wait() + g.conns.closeAll() + } } diff --git a/apps/daemon/internal/gateway/gateway_test.go b/apps/daemon/internal/gateway/gateway_test.go index 2925e0d20..359b0350f 100644 --- a/apps/daemon/internal/gateway/gateway_test.go +++ b/apps/daemon/internal/gateway/gateway_test.go @@ -4,6 +4,7 @@ import ( "bufio" "context" "crypto/x509" + "errors" "fmt" "io" "net" @@ -11,6 +12,8 @@ import ( "net/http/httptest" "net/netip" "net/url" + "os" + "strings" "sync" "sync/atomic" "testing" @@ -27,12 +30,15 @@ import ( const wait = 5 * time.Second +// model is a model upstream for tests that do not reach it. +var model = modelprovider.Provider{Protocol: modelprovider.Anthropic, BaseURL: "https://model.test", APIKey: upstreamKey} + // loopback is a gateway served on host loopback listeners at free ports, as // Start serves it in the Session's namespace. type loopback struct { Endpoints - // end ends the Session. - end context.CancelFunc + // stop stops the gateway. + stop func() // handlers counts the requests being handled. handlers sync.WaitGroup } @@ -41,12 +47,12 @@ type loopback struct { func serveOnLoopback(t *testing.T, cfg Config) *loopback { t.Helper() ctx, cancel := context.WithCancel(context.Background()) - t.Cleanup(cancel) g, err := build(ctx, cfg) if err != nil { + cancel() t.Fatal(err) } - l := &loopback{end: cancel} + l := &loopback{} lns := make([]*net.TCPListener, len(g.listeners)) ports := make([]int, len(lns)) for i := range lns { @@ -61,7 +67,8 @@ func serveOnLoopback(t *testing.T, cfg Config) *loopback { h.ServeHTTP(w, r) }) } - g.serve(ctx, lns) + l.stop = g.serve(ctx, cancel, lns) + t.Cleanup(l.stop) l.Endpoints = g.endpoints(ports) return l } @@ -160,7 +167,7 @@ func startSandbox(t *testing.T) *sandbox { func TestPlanKeepsPortsWithoutTheProxy(t *testing.T) { cfg := Config{ - Models: []Model{{Name: "main", Provider: modelprovider.Provider{Protocol: modelprovider.Anthropic, BaseURL: "https://upstream.test", APIKey: upstreamKey}}}, + Model: model, MCP: []agent.MCPBinding{{ConnectionOrigin: "service", ServerLabel: "tools", Transport: "http", ServerURL: "https://tools.test/mcp"}}, Prompt: proto.PromptRequestPayload{DisableExecutionEnvironment: true}, } @@ -170,7 +177,7 @@ func TestPlanKeepsPortsWithoutTheProxy(t *testing.T) { if err != nil { t.Fatal(err) } - want := Endpoints{Placeholder: modelprovider.Placeholder, Models: map[string]string{"main": "http://127.0.0.1:17101"}, MCP: map[string]string{"tools": "http://127.0.0.1:17102/mcp"}} + want := Endpoints{Model: "http://127.0.0.1:17101", MCP: map[string]string{"tools": "http://127.0.0.1:17102/mcp"}} if proxy { want.Proxy = "http://127.0.0.1:17100" } @@ -206,6 +213,7 @@ func TestSessionEndEndsBlockedRelays(t *testing.T) { })) defer srv.Close() gw := serveOnLoopback(t, Config{ + Model: model, MCP: []agent.MCPBinding{{ConnectionOrigin: "service", ServerLabel: "tools", Transport: "http", ServerURL: srv.URL + "/mcp"}}, Prompt: proto.PromptRequestPayload{DisableExecutionEnvironment: true}, }) @@ -227,7 +235,7 @@ func TestSessionEndEndsBlockedRelays(t *testing.T) { t.Fatal("the upstream never filled the Harness's side") } - gw.end() + gw.stop() returned := make(chan struct{}) go func() { gw.handlers.Wait() @@ -258,6 +266,7 @@ func TestRejectedUpgradeClosesTheUpstream(t *testing.T) { })) defer srv.Close() gw := serveOnLoopback(t, Config{ + Model: model, MCP: []agent.MCPBinding{{ConnectionOrigin: "service", ServerLabel: "tools", Transport: "http", ServerURL: srv.URL + "/mcp"}}, Prompt: proto.PromptRequestPayload{DisableExecutionEnvironment: true}, }) @@ -279,3 +288,65 @@ func TestRejectedUpgradeClosesTheUpstream(t *testing.T) { t.Fatal("the upstream connection is still open") } } + +func TestStopClosesEveryConnection(t *testing.T) { + // The upstream holds the request open over HTTP/2, whose connection the + // transport keeps when a request ends. + held, closed := make(chan struct{}), make(chan struct{}) + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(held) + <-r.Context().Done() + })) + srv.EnableHTTP2 = true + srv.Config.ConnState = func(_ net.Conn, s http.ConnState) { + if s == http.StateClosed { + close(closed) + } + } + srv.StartTLS() + defer srv.Close() + gw := serveOnLoopback(t, Config{Model: modelprovider.Provider{Protocol: modelprovider.Anthropic, BaseURL: srv.URL, APIKey: upstreamKey}, RootCAs: trust(srv)}) + + harness, err := net.Dial("tcp", strings.TrimPrefix(gw.Model, "http://")) + if err != nil { + t.Fatal(err) + } + defer harness.Close() + io.WriteString(harness, "POST /v1/messages HTTP/1.1\r\nHost: model\r\nContent-Length: 0\r\n\r\n") + select { + case <-held: + case <-time.After(wait): + t.Fatal("the request never reached the upstream") + } + + gw.stop() + harness.SetReadDeadline(time.Now().Add(wait)) + if _, err := io.Copy(io.Discard, harness); errors.Is(err, os.ErrDeadlineExceeded) { + t.Error("the Harness's connection is still open") + } + select { + case <-closed: + case <-time.After(wait): + t.Error("the upstream connection is still open") + } +} + +func TestClosedSetClosesNewConnections(t *testing.T) { + ln, err := net.ListenTCP("tcp4", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + defer ln.Close() + c, err := net.DialTCP("tcp4", nil, ln.Addr().(*net.TCPAddr)) + if err != nil { + t.Fatal(err) + } + var s connSet + s.closeAll() + if _, err := s.add(tcpConn{c}); !errors.Is(err, net.ErrClosed) { + t.Fatalf("add: %v", err) + } + if _, err := c.Write([]byte("x")); !errors.Is(err, net.ErrClosed) { + t.Errorf("the connection is open: %v", err) + } +} diff --git a/apps/daemon/internal/gateway/mcp_test.go b/apps/daemon/internal/gateway/mcp_test.go index 3c3c14de2..ad82809af 100644 --- a/apps/daemon/internal/gateway/mcp_test.go +++ b/apps/daemon/internal/gateway/mcp_test.go @@ -36,7 +36,7 @@ func TestMCPBrokersBothOrigins(t *testing.T) { dials int32 }{{"service", service, 0}, {"environment", environment, 1}} { before := sb.dials.Load() - eps := serveOnLoopback(t, Config{MCP: []agent.MCPBinding{binding(c.origin)}, Prompt: c.prompt, OpenNetwork: sb.open, RootCAs: trust(srv)}) + eps := serveOnLoopback(t, Config{Model: model, MCP: []agent.MCPBinding{binding(c.origin)}, Prompt: c.prompt, OpenNetwork: sb.open, RootCAs: trust(srv)}) if !strings.HasPrefix(eps.MCP["tools"], "http://127.0.0.1:") || !strings.HasSuffix(eps.MCP["tools"], "/mcp") { t.Fatalf("%s: Harness URL %q", c.origin, eps.MCP["tools"]) } @@ -65,7 +65,7 @@ func TestMCPBrokersBothOrigins(t *testing.T) { // Origin admission runs first: an environment binding needs an enabled // workspace network. - if _, err := Plan(Config{MCP: []agent.MCPBinding{binding("environment")}, Prompt: service, OpenNetwork: sb.open}); !errors.Is(err, ErrInvalidConfig) { + if _, err := Plan(Config{Model: model, MCP: []agent.MCPBinding{binding("environment")}, Prompt: service, OpenNetwork: sb.open}); !errors.Is(err, ErrInvalidConfig) { t.Errorf("Plan with a relocated binding: %v", err) } // The gateway relays HTTP only, and the bearer token owns Authorization. @@ -84,7 +84,7 @@ func TestMCPBrokersBothOrigins(t *testing.T) { b agent.MCPBinding prompt proto.PromptRequestPayload }{"stdio": {stdio, service}, "Authorization twice": {twice, service}, "headers over http": {plain, environment}, "a query over http": {query, environment}, "userinfo": {userinfo, service}} { - _, err := Plan(Config{MCP: []agent.MCPBinding{c.b}, Prompt: c.prompt, OpenNetwork: sb.open}) + _, err := Plan(Config{Model: model, MCP: []agent.MCPBinding{c.b}, Prompt: c.prompt, OpenNetwork: sb.open}) if !errors.Is(err, ErrInvalidConfig) || strings.Contains(err.Error(), "secret") { t.Errorf("Plan with %s: %v", name, err) } @@ -108,7 +108,7 @@ func TestMCPServesOnlyItsServerURL(t *testing.T) { defer srv.Close() const query = "tenant=acme&key=query%2Bsecret" b := agent.MCPBinding{ConnectionOrigin: "service", ServerLabel: "tools", Transport: "http", ServerURL: srv.URL + "/mcp?" + query} - eps := serveOnLoopback(t, Config{MCP: []agent.MCPBinding{b}, Prompt: proto.PromptRequestPayload{DisableExecutionEnvironment: true}, RootCAs: trust(srv)}) + eps := serveOnLoopback(t, Config{Model: model, MCP: []agent.MCPBinding{b}, Prompt: proto.PromptRequestPayload{DisableExecutionEnvironment: true}, RootCAs: trust(srv)}) harness := eps.MCP["tools"] if !strings.HasPrefix(harness, "http://127.0.0.1:") || !strings.HasSuffix(harness, "/mcp") || strings.Contains(harness, "?") { t.Fatalf("Harness URL %q", harness) diff --git a/apps/daemon/internal/gateway/model_test.go b/apps/daemon/internal/gateway/model_test.go index db6faf245..452010f6c 100644 --- a/apps/daemon/internal/gateway/model_test.go +++ b/apps/daemon/internal/gateway/model_test.go @@ -27,10 +27,10 @@ func startModel(t *testing.T, key string, handler http.HandlerFunc) (string, *ht srv := httptest.NewTLSServer(handler) t.Cleanup(srv.Close) eps := serveOnLoopback(t, Config{ - Models: []Model{{Name: "main", Provider: modelprovider.Provider{Protocol: modelprovider.Anthropic, BaseURL: srv.URL + "/anthropic", APIKey: key}}}, + Model: modelprovider.Provider{Protocol: modelprovider.Anthropic, BaseURL: srv.URL + "/anthropic", APIKey: key}, RootCAs: trust(srv), }) - return eps.Models["main"], srv + return eps.Model, srv } func TestModelInjectsTheKeyAndNeverThePlaceholder(t *testing.T) { diff --git a/apps/daemon/internal/gateway/proxy.go b/apps/daemon/internal/gateway/proxy.go index 05eec8fb0..40eac0d35 100644 --- a/apps/daemon/internal/gateway/proxy.go +++ b/apps/daemon/internal/gateway/proxy.go @@ -3,26 +3,22 @@ package gateway import ( "context" "io" - "net" "net/http" "net/http/httputil" "strings" "sync" - - "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" - "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxnet" ) // proxy is the generic proxy. It serves CONNECT tunnels and absolute-form // plain-HTTP requests, and connects only through a new Network stream to the // sandbox for each connection. It rejects every other request form. type proxy struct { - open func(context.Context) (sandboxlink.Stream, error) + dial dialFunc // through the sandbox; nil without a sandbox network forward *httputil.ReverseProxy } -func newProxy(open func(context.Context) (sandboxlink.Stream, error), sandbox *http.Transport) *proxy { - p := &proxy{open: open} +func newProxy(dial dialFunc, sandbox *http.Transport) *proxy { + p := &proxy{dial: dial} if sandbox != nil { // The request is relayed as the Harness addressed it; the reverse // proxy drops hop-by-hop headers, Proxy-Authorization among them. It @@ -37,7 +33,7 @@ func newProxy(open func(context.Context) (sandboxlink.Stream, error), sandbox *h func (p *proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { switch { - case p.open == nil: + case p.dial == nil: http.Error(w, "the Session has no sandbox network", http.StatusForbidden) case r.Method == http.MethodConnect && r.URL.Host != "" && r.URL.Path == "": p.tunnel(w, r) @@ -50,12 +46,7 @@ func (p *proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { // tunnel connects through the sandbox, answers 200 and then carries raw bytes. func (p *proxy) tunnel(w http.ResponseWriter, r *http.Request) { - host, port, err := splitHostPort(r.URL.Host) - if err != nil { - http.Error(w, "invalid CONNECT authority", http.StatusBadRequest) - return - } - remote, err := connectSandbox(r.Context(), p.open, host, port) + remote, err := p.dial(r.Context(), "tcp", r.URL.Host) if err != nil { status := statusOf(err) http.Error(w, http.StatusText(status), status) @@ -75,22 +66,19 @@ func (p *proxy) tunnel(w http.ResponseWriter, r *http.Request) { client.Close() return } - splice(r.Context(), client, pending, remote) + splice(r.Context(), client.(conn), pending, remote) } // splice carries bytes both ways between the Harness's connection and the // sandbox's. Each direction's orderly end reaches the other side as a // half-close after every byte before it, and starts no timeout. An error on // either side, or the end of ctx, aborts both. -func splice(ctx context.Context, client net.Conn, pending []byte, remote *sandboxnet.Conn) { +func splice(ctx context.Context, client conn, pending []byte, remote conn) { var end sync.Once abort := func() { end.Do(func() { remote.Reset() - if l, ok := client.(interface{ SetLinger(int) error }); ok { - l.SetLinger(0) - } - client.Close() + client.Reset() }) } stop := context.AfterFunc(ctx, abort) @@ -115,7 +103,7 @@ func splice(ctx context.Context, client net.Conn, pending []byte, remote *sandbo }() _, err := io.Copy(struct{ io.Writer }{client}, struct{ io.Reader }{remote}) if err == nil { - err = closeWrite(client) + err = client.CloseWrite() } if err != nil { abort() @@ -126,10 +114,3 @@ func splice(ctx context.Context, client net.Conn, pending []byte, remote *sandbo client.Close() }) } - -func closeWrite(c net.Conn) error { - if cw, ok := c.(interface{ CloseWrite() error }); ok { - return cw.CloseWrite() - } - return c.Close() -} diff --git a/apps/daemon/internal/gateway/proxy_test.go b/apps/daemon/internal/gateway/proxy_test.go index f6538aa29..dad072791 100644 --- a/apps/daemon/internal/gateway/proxy_test.go +++ b/apps/daemon/internal/gateway/proxy_test.go @@ -11,7 +11,7 @@ import ( func TestProxyConnectsOnlyThroughTheSandbox(t *testing.T) { sb := startSandbox(t) - eps := serveOnLoopback(t, Config{OpenNetwork: sb.open, Proxy: true}) + eps := serveOnLoopback(t, Config{Model: model, OpenNetwork: sb.open, Proxy: true}) hello := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "hello "+r.RequestURI) }) secure := httptest.NewTLSServer(hello) defer secure.Close() diff --git a/apps/daemon/internal/gateway/relay.go b/apps/daemon/internal/gateway/relay.go index c9f265477..35e94333c 100644 --- a/apps/daemon/internal/gateway/relay.go +++ b/apps/daemon/internal/gateway/relay.go @@ -35,7 +35,7 @@ const ( connectMargin = 5 * time.Second ) -type dialFunc func(ctx context.Context, network, addr string) (net.Conn, error) +type dialFunc func(ctx context.Context, network, addr string) (conn, error) // newTransport returns an upstream transport that relays requests as they // are: no proxy from the environment, no added compression and no redirects, @@ -43,7 +43,7 @@ type dialFunc func(ctx context.Context, network, addr string) (net.Conn, error) // when roots is nil, and always verifies the destination's hostname. func newTransport(roots *x509.CertPool, dial dialFunc) *http.Transport { return &http.Transport{ - DialContext: dial, + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { return dial(ctx, network, addr) }, TLSClientConfig: &tls.Config{RootCAs: roots}, ForceAttemptHTTP2: true, DisableCompression: true, @@ -66,23 +66,38 @@ func relayTransport(roots *x509.CertPool, dial dialFunc) *http.Transport { return t } -// sessionDial binds each dial to the Session as well as to its own context. -// http.Transport detaches a dial from the request that started it, and -// CloseIdleConnections cancels only dials that no request waits for. -func sessionDial(session context.Context, dial dialFunc) dialFunc { - return func(ctx context.Context, network, addr string) (net.Conn, error) { +// sessionDial binds each dial to the Session as well as to its own context, +// because http.Transport detaches a dial from the request that started it, +// and records each connection in conns. +func sessionDial(session context.Context, conns *connSet, dial dialFunc) dialFunc { + return func(ctx context.Context, network, addr string) (conn, error) { ctx, cancel := context.WithCancel(ctx) defer cancel() stop := context.AfterFunc(session, cancel) defer stop() - return dial(ctx, network, addr) + c, err := dial(ctx, network, addr) + if err != nil { + return nil, err + } + return conns.add(c) + } +} + +// hostDial connects from the agent host. +func hostDial(ctx context.Context, network, addr string) (conn, error) { + c, err := (&net.Dialer{Timeout: dialTimeout, KeepAlive: 30 * time.Second}).DialContext(ctx, network, addr) + if err != nil { + return nil, err } + return tcpConn{c.(*net.TCPConn)}, nil } // sandboxDialer connects through a new Network stream for each connection, so -// the sandbox resolves the name and the connection has sandbox origin. +// the sandbox resolves the name and the connection has sandbox origin. A local +// deadline bounds the stream's opening and the sandbox's answer, so a sandbox +// that never answers cannot hold the dial; the connection outlives it. func sandboxDialer(open func(context.Context) (sandboxlink.Stream, error)) dialFunc { - return func(ctx context.Context, network, addr string) (net.Conn, error) { + return func(ctx context.Context, network, addr string) (conn, error) { if network != "tcp" && network != "tcp4" && network != "tcp6" { return nil, &sandboxnet.Error{Code: sandboxnet.CodeUnsupportedNetwork, Effect: sandboxwire.EffectNone} } @@ -90,21 +105,18 @@ func sandboxDialer(open func(context.Context) (sandboxlink.Stream, error)) dialF if err != nil { return nil, err } - return connectSandbox(ctx, open, host, port) - } -} - -// connectSandbox opens a Network stream and connects through it. A local -// deadline bounds both, so a sandbox that never answers cannot hold the -// connection open; the returned Conn outlives it. -func connectSandbox(ctx context.Context, open func(context.Context) (sandboxlink.Stream, error), host string, port uint16) (*sandboxnet.Conn, error) { - ctx, cancel := context.WithTimeout(ctx, connectTimeout+connectMargin) - defer cancel() - s, err := open(ctx) - if err != nil { - return nil, err + ctx, cancel := context.WithTimeout(ctx, connectTimeout+connectMargin) + defer cancel() + s, err := open(ctx) + if err != nil { + return nil, err + } + c, err := sandboxnet.Connect(ctx, s, host, port, connectTimeout) + if err != nil { + return nil, err + } + return c, nil } - return sandboxnet.Connect(ctx, s, host, port, connectTimeout) } // splitHostPort splits an authority into an unbracketed host and a nonzero diff --git a/apps/daemon/internal/gateway/view_linux_test.go b/apps/daemon/internal/gateway/view_linux_test.go index 28438ec19..c0bc31cdd 100644 --- a/apps/daemon/internal/gateway/view_linux_test.go +++ b/apps/daemon/internal/gateway/view_linux_test.go @@ -17,13 +17,13 @@ import ( "testing" "time" - gofs "github.com/hanwen/go-fuse/v2/fs" - "github.com/hanwen/go-fuse/v2/fuse" "golang.org/x/sys/unix" "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/sessionview" "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/sessionview/sessionviewtest" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/worldfs" + "github.com/MiniMax-AI/OpenAgentCore/apps/sandboxio/fileservicetest" "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" "github.com/MiniMax-AI/OpenAgentCore/internal/modelprovider" ) @@ -69,7 +69,7 @@ func TestListenersExistOnlyInTheSession(t *testing.T) { port := strconv.Itoa(ln.Addr().(*net.TCPAddr).Port) cfg := Config{ - Models: []Model{{Name: "main", Provider: modelprovider.Provider{Protocol: modelprovider.Anthropic, BaseURL: "https://127.0.0.1:1", APIKey: upstreamKey}}}, + Model: modelprovider.Provider{Protocol: modelprovider.Anthropic, BaseURL: "https://127.0.0.1:1", APIKey: upstreamKey}, MCP: []agent.MCPBinding{{ConnectionOrigin: "service", ServerLabel: "tools", Transport: "http", ServerURL: "http://127.0.0.1:" + port + "/mcp"}}, Prompt: proto.PromptRequestPayload{DisableExecutionEnvironment: true}, OpenNetwork: startSandbox(t).open, @@ -82,21 +82,21 @@ func TestListenersExistOnlyInTheSession(t *testing.T) { } encoded, _ := json.Marshal(eps) - world, harness := t.TempDir(), t.TempDir() - for _, d := range []string{".oac/harness", ".oac/bin", "proc", "dev"} { - if err := os.MkdirAll(filepath.Join(world, d), 0o755); err != nil { - t.Fatal(err) - } + world, err := fileservicetest.New(t.TempDir()) + if err != nil { + t.Fatal(err) } + defer world.Close() + harness := t.TempDir() if err := os.Chmod(harness, 0o755); err != nil { t.Fatal(err) } copyExecutable(t, filepath.Join(harness, "harness")) - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() + stop := func() {} + defer func() { stop() }() v, err := sessionview.Start(context.Background(), sessionview.Spec{ - World: (&loopbackWorld{dir: world}).serve, + World: worldfs.New(fileservicetest.Export, world.Dial).Serve, StagingParent: t.TempDir(), CgroupParent: sessionviewtest.CgroupParent(t), Private: []sessionview.PrivateDir{{Name: "harness", HostDir: harness, Exec: true}}, @@ -104,8 +104,8 @@ func TestListenersExistOnlyInTheSession(t *testing.T) { Path: "/.oac/harness/harness", Args: []string{"harness"}, Dir: "/", UID: viewID, GID: viewID, Stderr: os.Stderr, Env: []string{harnessEnv + "=1", endpointsEnv + "=" + string(encoded), externalEnv + "=" + net.JoinHostPort(hostAddress(t), port)}, }, - Network: sessionview.Network{Setup: func(netns *os.File) error { - _, err := Start(ctx, SessionNetwork{Namespace: netns}, cfg) + Network: sessionview.Network{Setup: func(netns *os.File) (err error) { + stop, err = Start(netns, cfg) return err }}, }) @@ -131,7 +131,7 @@ func TestListenersExistOnlyInTheSession(t *testing.T) { } // The listeners still serve the Session's namespace; the host's has none. - for _, u := range []string{eps.Proxy, eps.Models["main"], eps.MCP["tools"]} { + for _, u := range []string{eps.Proxy, eps.Model, eps.MCP["tools"]} { parsed, _ := url.Parse(u) if c, err := net.DialTimeout("tcp", parsed.Host, time.Second); err == nil { c.Close() @@ -171,7 +171,7 @@ func runHarness() int { } checks := map[string]func() error{ // An undeclared route is answered by the listener itself. - "model listener": func() error { return answers(direct, eps.Models["main"]+"/", http.StatusNotFound, "") }, + "model listener": func() error { return answers(direct, eps.Model+"/", http.StatusNotFound, "") }, "MCP listener": func() error { return answers(direct, eps.MCP["tools"], http.StatusOK, "tools") }, "no direct route": func() error { c, err := net.DialTimeout("tcp", external, 2*time.Second) @@ -228,45 +228,3 @@ func copyExecutable(t *testing.T, dst string) { t.Fatal(err) } } - -// loopbackWorld serves a directory as the view's world. It presents each mountpoint at its declared path. -type loopbackWorld struct { - dir string - served chan struct{} -} - -func (w *loopbackWorld) serve(_ context.Context, dev *os.File, mount sessionview.WorldMount) (sessionview.WorldServer, sessionview.Presentation, error) { - fd, err := unix.Dup(int(dev.Fd())) - if err != nil { - return nil, sessionview.Presentation{}, err - } - root, err := gofs.NewLoopbackRoot(w.dir) - if err != nil { - unix.Close(fd) - return nil, sessionview.Presentation{}, err - } - srv, err := fuse.NewServer(gofs.NewNodeFS(root, &gofs.Options{}), fmt.Sprintf("/dev/fd/%d", fd), &fuse.MountOptions{}) - if err != nil { - unix.Close(fd) - return nil, sessionview.Presentation{}, err - } - w.served = make(chan struct{}) - go func() { - srv.Serve() - close(w.served) - }() - var p sessionview.Presentation - for _, m := range mount.Mountpoints { - p.Targets = append(p.Targets, m.Path) - } - return w, p, nil -} - -func (w *loopbackWorld) Stop() error { - select { - case <-w.served: - return nil - case <-time.After(10 * time.Second): - return errors.New("world still serving 10s after the view ended") - } -} diff --git a/apps/daemon/internal/worldfs/conn_linux.go b/apps/daemon/internal/worldfs/conn_linux.go index 603b8ded0..562c5111c 100644 --- a/apps/daemon/internal/worldfs/conn_linux.go +++ b/apps/daemon/internal/worldfs/conn_linux.go @@ -5,6 +5,7 @@ package worldfs import ( "context" "errors" + "fmt" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxfs" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" @@ -36,7 +37,7 @@ func (f *frontend) client(ctx context.Context, interrupt <-chan struct{}) (*sand select { case f.connTurn <- struct{}{}: case <-ctx.Done(): - return nil, &Error{Kind: ErrConnect, Op: "reconnect", Err: ctx.Err()} + return nil, fmt.Errorf("%w: reconnect: %w", ErrConnect, ctx.Err()) case <-interrupt: return nil, errInterrupted } @@ -45,7 +46,7 @@ func (f *frontend) client(ctx context.Context, interrupt <-chan struct{}) (*sand return nil, errDead } if f.ctx.Err() != nil { - return nil, &Error{Kind: ErrConnect, Op: "reconnect", Err: errStopped} + return nil, fmt.Errorf("%w: reconnect: %w", ErrConnect, errStopped) } if c := f.conn; c != nil { select { @@ -71,10 +72,10 @@ func (f *frontend) client(ctx context.Context, interrupt <-chan struct{}) (*sand return nil, errInterrupted case err != nil: f.observe(err) - return nil, &Error{Kind: ErrConnect, Op: "reconnect", Err: err} + return nil, fmt.Errorf("%w: reconnect: %w", ErrConnect, err) case d.ServerInstanceID != f.instance: c.Close() - f.lose(&Error{Kind: ErrInstanceChanged, Op: "reconnect"}, true) + f.lose(fmt.Errorf("%w: reconnect", ErrInstanceChanged), true) return nil, errDead } f.conn = c diff --git a/apps/daemon/internal/worldfs/dir_linux.go b/apps/daemon/internal/worldfs/dir_linux.go index cc8bfe058..5c271841f 100644 --- a/apps/daemon/internal/worldfs/dir_linux.go +++ b/apps/daemon/internal/worldfs/dir_linux.go @@ -88,8 +88,3 @@ func (f *frontend) FsyncDir(_ <-chan struct{}, in *fuse.FsyncIn) fuse.Status { _, err := call(f, f.ctx, (*sandboxfs.Client).Fsync, &sandboxfs.FsyncRequest{Handle: h.server, DataOnly: in.FsyncFlags&1 != 0}) return status(err) } - -// ReadDirPlus is never negotiated. -func (f *frontend) ReadDirPlus(<-chan struct{}, *fuse.ReadIn, *fuse.DirEntryList) fuse.Status { - return fuse.ENOSYS -} diff --git a/apps/daemon/internal/worldfs/errors.go b/apps/daemon/internal/worldfs/errors.go index aca4f82ab..6b2c63853 100644 --- a/apps/daemon/internal/worldfs/errors.go +++ b/apps/daemon/internal/worldfs/errors.go @@ -4,7 +4,6 @@ import ( "context" "errors" "io" - "strings" ) // Dial opens a new File stream to the attachment's service. It returns once ctx ends; ctx bounds the open, not the stream. @@ -22,33 +21,3 @@ var ( // ErrAttachmentDirty is a failed Serve whose cleanup Detach could not be sent or answered, so it cannot show the attachment holds nothing: the owner of the Link attachment must end it. ErrAttachmentDirty = errors.New("worldfs: the attachment may still hold state") ) - -// Error is a typed worldfs failure. It matches Kind and, when present, Err. -type Error struct { - Kind error - Op string - Path string - Err error -} - -func (e *Error) Error() string { - var b strings.Builder - b.WriteString(e.Kind.Error()) - if e.Op != "" { - b.WriteString(": " + e.Op) - } - if e.Path != "" { - b.WriteString(" " + e.Path) - } - if e.Err != nil { - b.WriteString(": " + e.Err.Error()) - } - return b.String() -} - -func (e *Error) Unwrap() []error { - if e.Err == nil { - return []error{e.Kind} - } - return []error{e.Kind, e.Err} -} diff --git a/apps/daemon/internal/worldfs/files_linux.go b/apps/daemon/internal/worldfs/files_linux.go index f2347df3a..c8135cce9 100644 --- a/apps/daemon/internal/worldfs/files_linux.go +++ b/apps/daemon/internal/worldfs/files_linux.go @@ -143,21 +143,6 @@ func (f *frontend) Release(_ <-chan struct{}, in *fuse.ReleaseIn) { } } -// Lseek has no File request. ENOSYS makes the kernel seek itself; SEEK_DATA and SEEK_HOLE then treat the file as one data extent. -func (f *frontend) Lseek(<-chan struct{}, *fuse.LseekIn, *fuse.LseekOut) fuse.Status { - return fuse.ENOSYS -} - -// Fallocate has no File request. ENOSYS makes the kernel stop asking and answer EOPNOTSUPP itself. -func (f *frontend) Fallocate(<-chan struct{}, *fuse.FallocateIn) fuse.Status { - return fuse.ENOSYS -} - -// CopyFileRange has no File request. ENOSYS makes the kernel copy with Read and Write. -func (f *frontend) CopyFileRange(<-chan struct{}, *fuse.CopyFileRangeIn) (uint32, fuse.Status) { - return 0, fuse.ENOSYS -} - // Ioctl has no File request: no world file takes ioctls. func (f *frontend) Ioctl(<-chan struct{}, *fuse.IoctlIn, []byte, *fuse.IoctlOut, []byte) fuse.Status { return errno(syscall.ENOTTY) diff --git a/apps/daemon/internal/worldfs/listing_linux.go b/apps/daemon/internal/worldfs/listing_linux.go index d4425f42a..d8964437d 100644 --- a/apps/daemon/internal/worldfs/listing_linux.go +++ b/apps/daemon/internal/worldfs/listing_linux.go @@ -3,6 +3,8 @@ package worldfs import ( + "fmt" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxfs" "github.com/hanwen/go-fuse/v2/fuse" ) @@ -62,7 +64,7 @@ func (f *frontend) readPresented(h *handle, in *fuse.ReadIn, out *fuse.DirEntryL c := n.fixed[name] if c != nil { if !c.synthetic() && e.Type != c.fileType() { - f.lose(&Error{Kind: ErrTopologyChanged, Op: "readdir", Path: c.path}, false) + f.lose(fmt.Errorf("%w: readdir %s", ErrTopologyChanged, c.path), false) } ent.Mode, ent.Ino = c.fileType(), f.ino(c) } @@ -82,7 +84,7 @@ func (f *frontend) readPresented(h *handle, in *fuse.ReadIn, out *fuse.DirEntryL if l.pass != nil { for _, name := range n.order { if c := n.fixed[name]; !c.synthetic() && !l.pass.names[name] { - f.lose(&Error{Kind: ErrTopologyChanged, Op: "readdir", Path: c.path}, false) + f.lose(fmt.Errorf("%w: readdir %s", ErrTopologyChanged, c.path), false) } } } diff --git a/apps/daemon/internal/worldfs/ops_linux.go b/apps/daemon/internal/worldfs/ops_linux.go index 57c57c5f7..c770af09b 100644 --- a/apps/daemon/internal/worldfs/ops_linux.go +++ b/apps/daemon/internal/worldfs/ops_linux.go @@ -3,6 +3,7 @@ package worldfs import ( + "fmt" "syscall" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxfs" @@ -10,10 +11,7 @@ import ( "golang.org/x/sys/unix" ) -func (f *frontend) String() string { return "oac-world" } -func (f *frontend) SetDebug(bool) {} -func (f *frontend) Init(*fuse.Server) {} -func (f *frontend) OnUnmount() {} +func (f *frontend) String() string { return "oac-world" } func (f *frontend) Lookup(_ <-chan struct{}, in *fuse.InHeader, name string, out *fuse.EntryOut) fuse.Status { p, st := f.node(in.NodeId) @@ -53,7 +51,7 @@ func (f *frontend) lookupPinned(p *inode, name string, c *inode, out *fuse.Entry case !isErrno(err, sandboxfs.ErrnoNotFound): return status(err) } - f.lose(&Error{Kind: ErrTopologyChanged, Op: "lookup", Path: c.path}, false) + f.lose(fmt.Errorf("%w: lookup %s", ErrTopologyChanged, c.path), false) f.lookedUp(c) f.entry(c, c.attr, out) return fuse.OK @@ -330,26 +328,3 @@ func (f *frontend) StatFs(_ <-chan struct{}, in *fuse.InHeader, out *fuse.Statfs } return fuse.OK } - -// The File protocol has no extended attributes. ENOSYS makes the kernel stop asking and answer EOPNOTSUPP itself. - -func (f *frontend) GetXAttr(<-chan struct{}, *fuse.InHeader, string, []byte) (uint32, fuse.Status) { - return 0, fuse.ENOSYS -} - -func (f *frontend) ListXAttr(<-chan struct{}, *fuse.InHeader, []byte) (uint32, fuse.Status) { - return 0, fuse.ENOSYS -} - -func (f *frontend) SetXAttr(<-chan struct{}, *fuse.SetXAttrIn, string, []byte) fuse.Status { - return fuse.ENOSYS -} - -func (f *frontend) RemoveXAttr(<-chan struct{}, *fuse.InHeader, string) fuse.Status { - return fuse.ENOSYS -} - -// Statx has no File request. ENOSYS makes the kernel fall back to GetAttr. -func (f *frontend) Statx(<-chan struct{}, *fuse.StatxIn, *fuse.StatxOut) fuse.Status { - return fuse.ENOSYS -} diff --git a/apps/daemon/internal/worldfs/present_linux.go b/apps/daemon/internal/worldfs/present_linux.go index 1d4ba16e8..5eb5f8474 100644 --- a/apps/daemon/internal/worldfs/present_linux.go +++ b/apps/daemon/internal/worldfs/present_linux.go @@ -223,7 +223,7 @@ func (r *resolver) at(name string) string { } func (r *resolver) fail(err error) error { - return &Error{Kind: ErrMountpoint, Op: "present", Path: r.mp.Path, Err: err} + return fmt.Errorf("%w: present %s: %w", ErrMountpoint, r.mp.Path, err) } // serverErr reports a sandbox errno as a presentation failure and anything else as a connection failure. @@ -232,7 +232,7 @@ func (r *resolver) serverErr(err error) error { if errors.As(err, &fail) && fail.Code == sandboxfs.CodeErrno { return r.fail(fmt.Errorf("%w (%w)", errnoOf(err), err)) } - return &Error{Kind: ErrConnect, Op: "present", Path: r.mp.Path, Err: err} + return fmt.Errorf("%w: present %s: %w", ErrConnect, r.mp.Path, err) } func isName(s string) bool { return s != "" && s != "." && s != ".." } diff --git a/apps/daemon/internal/worldfs/world_linux.go b/apps/daemon/internal/worldfs/world_linux.go index 2aa19da28..a8395cde9 100644 --- a/apps/daemon/internal/worldfs/world_linux.go +++ b/apps/daemon/internal/worldfs/world_linux.go @@ -39,6 +39,8 @@ func New(export sandboxlink.ExportID, dial Dial) *World { ctx, cancel := context.WithCancel(context.Background()) drainCtx, stopDrain := context.WithCancel(ctx) return &World{fs: &frontend{ + RawFileSystem: fuse.NewDefaultRawFileSystem(), + export: export, dial: dial, ctx: ctx, @@ -89,6 +91,9 @@ func (w *World) Err() error { // frontend is the FUSE file system. go-fuse calls it from its reader goroutines. type frontend struct { + // RawFileSystem answers what the File protocol has no request for, as the Mapping table in the package documentation lists. + fuse.RawFileSystem + export sandboxlink.ExportID dial Dial ctx context.Context @@ -141,7 +146,7 @@ type frontend struct { func (f *frontend) serve(ctx context.Context, dev *os.File, mount sessionview.WorldMount) (sessionview.Presentation, error) { if !f.started.CompareAndSwap(false, true) { - return sessionview.Presentation{}, &Error{Kind: ErrConnect, Op: "serve", Err: errors.New("the world already serves a view")} + return sessionview.Presentation{}, fmt.Errorf("%w: serve: the world already serves a view", ErrConnect) } f.view = sandboxfs.Identity{UID: mount.UID, GID: mount.GID} now := time.Now() @@ -161,23 +166,23 @@ func (f *frontend) serve(ctx context.Context, dev *os.File, mount sessionview.Wo func (f *frontend) attach(ctx context.Context, mps []sessionview.Mountpoint) (sessionview.Presentation, *fuse.MountOptions, error) { c, d, err := f.connect(ctx) if err != nil { - return sessionview.Presentation{}, nil, &Error{Kind: ErrConnect, Op: "describe", Err: err} + return sessionview.Presentation{}, nil, fmt.Errorf("%w: describe: %w", ErrConnect, err) } f.conn, f.instance, f.service, f.caps = c, d.ServerInstanceID, d.Identity, d.Capabilities maxIO := min(f.caps.MaxReadBytes, f.caps.MaxWriteBytes) &^ uint32(os.Getpagesize()-1) switch { case f.caps.PathProfile != sandboxfs.PathProfileLinuxBytes || f.caps.CacheProfile != sandboxfs.CacheProfileUncached: - return sessionview.Presentation{}, nil, &Error{Kind: ErrIncompatible, Op: "describe", Err: fmt.Errorf("path profile %d, cache profile %d", f.caps.PathProfile, f.caps.CacheProfile)} + return sessionview.Presentation{}, nil, fmt.Errorf("%w: describe: path profile %d, cache profile %d", ErrIncompatible, f.caps.PathProfile, f.caps.CacheProfile) case f.caps.ReadOnly: - return sessionview.Presentation{}, nil, &Error{Kind: ErrIncompatible, Op: "describe", Err: errors.New("the export is read-only")} + return sessionview.Presentation{}, nil, fmt.Errorf("%w: describe: the export is read-only", ErrIncompatible) case maxIO == 0 || f.caps.MaxWalkComponents == 0 || f.caps.MaxReadDirBytes == 0: - return sessionview.Presentation{}, nil, &Error{Kind: ErrIncompatible, Op: "describe", Err: errors.New("read, write, walk or directory limit too small")} + return sessionview.Presentation{}, nil, fmt.Errorf("%w: describe: read, write, walk or directory limit too small", ErrIncompatible) } a, err := c.Attach(ctx, &sandboxfs.AttachRequest{Export: f.export}) if err != nil { var fail *sandboxfs.Failure f.maybe = !errors.As(err, &fail) || fail.Effect != sandboxwire.EffectNone - return sessionview.Presentation{}, nil, &Error{Kind: ErrConnect, Op: "attach", Path: string(f.export), Err: err} + return sessionview.Presentation{}, nil, fmt.Errorf("%w: attach %s: %w", ErrConnect, f.export, err) } f.attached = true f.root = f.newInode(a.Root.Node) @@ -200,11 +205,11 @@ func (f *frontend) attach(ctx context.Context, mps []sessionview.Mountpoint) (se func (f *frontend) start(dev *os.File, opts *fuse.MountOptions) error { fd, err := unix.FcntlInt(dev.Fd(), unix.F_DUPFD_CLOEXEC, 3) if err != nil { - return &Error{Kind: ErrConnect, Op: "dup", Err: err} + return fmt.Errorf("%w: dup: %w", ErrConnect, err) } srv, err := fuse.NewServer(f, fmt.Sprintf("/dev/fd/%d", fd), opts) if err != nil { - return &Error{Kind: ErrConnect, Op: "init", Err: err} + return fmt.Errorf("%w: init: %w", ErrConnect, err) } go f.drain() go func() { @@ -226,7 +231,7 @@ func (f *frontend) abort(ctx context.Context, err error) error { ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), detachWait) defer cancel() if _, derr := call(f, ctx, (*sandboxfs.Client).Detach, &sandboxfs.DetachRequest{}); derr != nil && !f.dead.Load() { - return &Error{Kind: ErrAttachmentDirty, Op: "detach", Err: errors.Join(err, derr)} + return fmt.Errorf("%w: detach: %w", ErrAttachmentDirty, errors.Join(err, derr)) } return err } @@ -237,9 +242,9 @@ func (f *frontend) observe(err error) { if errors.As(err, &fail) { switch fail.Code { case sandboxfs.CodeInstanceChanged: - f.lose(&Error{Kind: ErrInstanceChanged, Err: err}, true) + f.lose(fmt.Errorf("%w: %w", ErrInstanceChanged, err), true) case sandboxfs.CodeStaleAttachment: - f.lose(&Error{Kind: ErrAttachmentLost, Err: err}, true) + f.lose(fmt.Errorf("%w: %w", ErrAttachmentLost, err), true) } return } @@ -247,9 +252,9 @@ func (f *frontend) observe(err error) { switch { case !ok || code.Retryable(): case code == sandboxlink.InstanceChanged: - f.lose(&Error{Kind: ErrInstanceChanged, Err: err}, true) + f.lose(fmt.Errorf("%w: %w", ErrInstanceChanged, err), true) default: - f.lose(&Error{Kind: ErrAttachmentLost, Err: err}, true) + f.lose(fmt.Errorf("%w: %w", ErrAttachmentLost, err), true) } } @@ -263,7 +268,7 @@ func linkCode(err error) (sandboxlink.Code, bool) { } // lose reports why the view must be rebuilt. dead also fails every later request. -func (f *frontend) lose(err *Error, dead bool) { +func (f *frontend) lose(err error, dead bool) { if dead { f.dead.Store(true) } @@ -282,7 +287,7 @@ func (f *frontend) stop() error { select { case <-f.served: case <-time.After(stopWait): - errs = append(errs, &Error{Kind: ErrConnect, Op: "stop", Err: errors.New("the view's mount still exists")}) + errs = append(errs, fmt.Errorf("%w: stop: the view's mount still exists", ErrConnect)) } f.closed.Store(true) ctx, cancel := context.WithTimeout(f.ctx, detachWait) @@ -294,7 +299,7 @@ func (f *frontend) stop() error { <-f.drained if !f.dead.Load() { if _, err := call(f, ctx, (*sandboxfs.Client).Detach, &sandboxfs.DetachRequest{}); err != nil { - errs = append(errs, &Error{Kind: ErrConnect, Op: "detach", Err: err}) + errs = append(errs, fmt.Errorf("%w: detach: %w", ErrConnect, err)) } } f.shutdown() diff --git a/apps/daemon/internal/worldfs/world_linux_test.go b/apps/daemon/internal/worldfs/world_linux_test.go index b86791336..6c88a03a6 100644 --- a/apps/daemon/internal/worldfs/world_linux_test.go +++ b/apps/daemon/internal/worldfs/world_linux_test.go @@ -129,6 +129,16 @@ func serve(t *testing.T, backing string, id uint32, mps ...sessionview.Mountpoin }) ws, p, err := m.world.Serve(context.Background(), dev, sessionview.WorldMount{UID: id, GID: id, Mountpoints: mps}) m.present = p + if err == nil { + // The kernel asks the server about a file's first poll, and the Go runtime polls each file this process opens without releasing its P, which the world needs to answer. + // Poll go-fuse's own file once with a call that releases the P, as fuse.Server.WaitMount does for other mountpoints, and the kernel stops asking. + fd, err := unix.Open(filepath.Join(mnt, ".go-fuse-epoll-hack"), unix.O_RDONLY, 0) + if err != nil { + t.Fatal(err) + } + unix.Poll([]unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}, 0) + unix.Close(fd) + } t.Cleanup(func() { if !m.stopped { if err := unix.Unmount(mnt, unix.MNT_DETACH); err != nil { @@ -169,6 +179,37 @@ func TestPOSIX(t *testing.T) { } } +// The operations TestPOSIX skips fail as the package documentation maps them. +func TestUnsupportedErrnos(t *testing.T) { + requireFUSE(t) + m, err := serve(t, t.TempDir(), 0) + if err != nil { + t.Fatalf("Serve: %v", err) + } + p := filepath.Join(m.dir, "f") + f, err := os.Create(p) + if err != nil { + t.Fatal(err) + } + defer f.Close() + fd := int(f.Fd()) + _, getErr := unix.Getxattr(p, "user.x", make([]byte, 64)) + _, listErr := unix.Listxattr(p, make([]byte, 64)) + for name, c := range map[string]struct{ err, want error }{ + "fallocate": {unix.Fallocate(fd, 0, 0, 4096), unix.EOPNOTSUPP}, + "fallocate keep size": {unix.Fallocate(fd, unix.FALLOC_FL_KEEP_SIZE, 0, 4096), unix.EOPNOTSUPP}, + "getxattr": {getErr, unix.EOPNOTSUPP}, + "listxattr": {listErr, unix.EOPNOTSUPP}, + "setxattr": {unix.Setxattr(p, "user.x", []byte("v"), 0), unix.EOPNOTSUPP}, + "removexattr": {unix.Removexattr(p, "user.x"), unix.EOPNOTSUPP}, + "fcntl lock": {unix.FcntlFlock(uintptr(fd), unix.F_SETLK, &unix.Flock_t{Type: unix.F_WRLCK}), unix.ENOLCK}, + } { + if !errors.Is(c.err, c.want) { + t.Errorf("%s = %v, want %v", name, c.err, c.want) + } + } +} + func TestOwnerMapping(t *testing.T) { requireFUSE(t) const view, other = 1000, 4242 @@ -213,15 +254,21 @@ func TestInstanceChanged(t *testing.T) { t.Fatal(err) } // A private mapping reads the file through READ when it faults; a shared one is refused. + // madvise faults the mapping in within a system call, which releases the P the world needs to answer; a fault on a load would hold it. mf, err := os.Open(f) if err != nil { t.Fatal(err) } - if b, err := unix.Mmap(int(mf.Fd()), 0, len("before"), unix.PROT_READ, unix.MAP_PRIVATE); err != nil || string(b) != "before" { - t.Errorf("private mapping = %q, %v", b, err) - } else { - unix.Munmap(b) + b, err := unix.Mmap(int(mf.Fd()), 0, len("before"), unix.PROT_READ, unix.MAP_PRIVATE) + if err != nil { + t.Fatalf("private mapping: %v", err) + } + if err := unix.Madvise(b, unix.MADV_POPULATE_READ); err != nil { + t.Errorf("fault the private mapping in: %v", err) + } else if string(b) != "before" { + t.Errorf("private mapping = %q", b) } + unix.Munmap(b) if _, err := unix.Mmap(int(mf.Fd()), 0, len("before"), unix.PROT_READ, unix.MAP_SHARED); !errors.Is(err, syscall.ENODEV) { t.Errorf("shared mapping: %v, want ENODEV", err) } diff --git a/apps/daemon/internal/worldfs/world_other.go b/apps/daemon/internal/worldfs/world_other.go index 4c0b9fb16..a822ad6a9 100644 --- a/apps/daemon/internal/worldfs/world_other.go +++ b/apps/daemon/internal/worldfs/world_other.go @@ -4,6 +4,7 @@ package worldfs import ( "context" + "fmt" "os" "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/sessionview" @@ -18,7 +19,7 @@ func New(sandboxlink.ExportID, Dial) *World { return &World{} } // Serve reports ErrUnsupported. func (w *World) Serve(context.Context, *os.File, sessionview.WorldMount) (sessionview.WorldServer, sessionview.Presentation, error) { - return nil, sessionview.Presentation{}, &Error{Kind: ErrUnsupported, Op: "serve"} + return nil, sessionview.Presentation{}, fmt.Errorf("%w: serve", ErrUnsupported) } // Stop does nothing: the world never served.