Skip to content

Run a process broker per agent-host view #371

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SaladDay merged 2 commits into from
Oct 1, 2026
Merged

Run a process broker per agent-host view #371

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f4a4d12
Stop the world when a view's Start is cancelled during the build
SaladDay Oct 1, 2026
ba0079b
Run a process broker per agent-host view
SaladDay Oct 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 24 additions & 1 deletion apps/daemon/internal/agenthost/admit.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ import (

"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent"
"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/gateway"
"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/processbroker"
"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto"
"github.com/MiniMax-AI/OpenAgentCore/internal/modelprovider"
"github.com/MiniMax-AI/OpenAgentCore/internal/sandboxfs"
Expand All @@ -38,6 +39,9 @@ type plan struct {
// mcp and proxy are ViewSession.MCP and ViewSession.Proxy.
mcp []agent.MCPBinding
proxy string
// executables is each view's process broker table: each shim name runs
// that name on the sandbox PATH, and each shim path the same path.
executables processbroker.Executables
}

// checkConfig validates cfg and loads the roots in its CA directory.
Expand Down Expand Up @@ -115,6 +119,8 @@ func admit(cfg Config, roots *x509.CertPool, s Session, openNetwork func(context
case len(req.FunctionTools) > 0 || req.ToolSearch:
// A function call waits for a result that Input cannot deliver.
return nil, unsupported("function tools and their discovery")
case len(view.Shims) > 0 && !hasPATH(s.Environment):
return nil, invalidSession("the view's shims run names on the sandbox PATH, and the Environment sets no PATH")
}
raw, ok := req.AgentOptions["model_provider"]
if !ok {
Expand Down Expand Up @@ -148,7 +154,8 @@ func admit(cfg Config, roots *x509.CertPool, s Session, openNetwork func(context
if err != nil {
return nil, &Error{Kind: ErrInvalidSession, Op: "gateway", Err: err}
}
p := &plan{view: view, gateway: gw, proxy: endpoints.Proxy}
p := &plan{view: view, gateway: gw, 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
}
Expand Down Expand Up @@ -228,6 +235,22 @@ func checkSession(s Session) error {
return nil
}

// hasPATH reports whether a forwarded process receives PATH.
func hasPATH(env Environment) bool {
_, sandbox := env.Sandbox["PATH"]
_, tool := env.Tool["PATH"]
return sandbox || tool
}

// identity maps each of keys to itself.
func identity(keys []string) map[string]string {
m := make(map[string]string, len(keys))
for _, k := range keys {
m[k] = k
}
return m
}

// valid reports whether r is a nonempty range of nonzero uids.
func (r UIDRange) valid() bool {
return r.First != 0 && r.Count != 0 && uint64(r.First)+uint64(r.Count) <= math.MaxUint32
Expand Down
31 changes: 18 additions & 13 deletions apps/daemon/internal/agenthost/admit_linux_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,9 @@ import (
var errFactory = errors.New("factory reached")

// viewFixture registers "viewed", whose factory records what it receives,
// "masked", whose view masks an /etc file the agent host writes, and
// "plain", which declares no view.
// "masked", whose view masks an /etc file the agent host writes, "shimmed",
// whose view runs a shim name on the sandbox PATH, and "plain", which
// declares no view.
type viewFixture struct {
cfg Config
req proto.PromptRequestPayload
Expand Down Expand Up @@ -54,6 +55,9 @@ func newViewFixture(t *testing.T) *viewFixture {
masked := view
masked.Masks = []agent.ViewMask{{Path: "/etc/passwd"}}
register(reg, "masked", &masked)
shimmed := view
shimmed.Shims = []string{"git"}
register(reg, "shimmed", &shimmed)
register(reg, "plain", nil)
f.cfg = newConfig(t, reg, upstream.Certificate())
return f
Expand All @@ -70,13 +74,14 @@ func TestAdmissionRejectsBeforeAnyEffect(t *testing.T) {
"environment none": {func(r *proto.PromptRequestPayload) {
r.DisableExecutionEnvironment, r.LocalEnvironment = true, nil
}, []error{ErrUnsupported, agent.ErrUnsupportedOperation}},
"relative workspace": {func(r *proto.PromptRequestPayload) { r.LocalEnvironment.WorkspaceRoot = "workspace" }, []error{ErrInvalidSession}},
"no model provider": {func(r *proto.PromptRequestPayload) { delete(r.AgentOptions, "model_provider") }, []error{ErrUnsupported}},
"no strict resume": {func(r *proto.PromptRequestPayload) { r.StrictResume = false }, []error{ErrUnsupported}},
"capabilities": {func(r *proto.PromptRequestPayload) { r.LocalEnvironment.Capabilities = true }, []error{ErrUnsupported}},
"restricted network": {func(r *proto.PromptRequestPayload) { r.LocalEnvironment.NetworkAccess = "disabled" }, []error{ErrUnsupported}},
"allowed domains only": {func(r *proto.PromptRequestPayload) { r.LocalEnvironment.AllowedDomains = []string{"example.com"} }, []error{ErrUnsupported}},
"function tools": {func(r *proto.PromptRequestPayload) { r.FunctionTools = []proto.FunctionTool{{Name: "lookup"}} }, []error{ErrUnsupported, agent.ErrUnsupportedOperation}},
"shim name without PATH": {func(r *proto.PromptRequestPayload) { r.AgentKind = "shimmed" }, []error{ErrInvalidSession}},
"relative workspace": {func(r *proto.PromptRequestPayload) { r.LocalEnvironment.WorkspaceRoot = "workspace" }, []error{ErrInvalidSession}},
"no model provider": {func(r *proto.PromptRequestPayload) { delete(r.AgentOptions, "model_provider") }, []error{ErrUnsupported}},
"no strict resume": {func(r *proto.PromptRequestPayload) { r.StrictResume = false }, []error{ErrUnsupported}},
"capabilities": {func(r *proto.PromptRequestPayload) { r.LocalEnvironment.Capabilities = true }, []error{ErrUnsupported}},
"restricted network": {func(r *proto.PromptRequestPayload) { r.LocalEnvironment.NetworkAccess = "disabled" }, []error{ErrUnsupported}},
"allowed domains only": {func(r *proto.PromptRequestPayload) { r.LocalEnvironment.AllowedDomains = []string{"example.com"} }, []error{ErrUnsupported}},
"function tools": {func(r *proto.PromptRequestPayload) { r.FunctionTools = []proto.FunctionTool{{Name: "lookup"}} }, []error{ErrUnsupported, agent.ErrUnsupportedOperation}},
"stdio MCP": {func(r *proto.PromptRequestPayload) {
r.LocalEnvironment.MCP = []proto.EnvironmentMCP{{Server: agentplugin.MCPServer{Name: "tools", Type: "stdio", Command: "tools"}}}
}, []error{ErrUnsupported, agent.ErrUnsupportedOperation}},
Expand All @@ -85,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), broker: func() processBroker { return noBroker{} }, procs: &fakeProcesses{}})
err := run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), procs: &fakeProcesses{}})
for _, want := range c.want {
if !errors.Is(err, want) {
t.Errorf("%s: Run = %v, want %v", name, err, want)
Expand All @@ -107,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), broker: func() processBroker { return noBroker{} }, procs: &fakeProcesses{}})
err := run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), procs: &fakeProcesses{}})
if !errors.Is(err, ErrInvalidSession) || dials.Load() != 0 {
t.Errorf("%s: Run = %v after %d dials, want ErrInvalidSession", name, err, dials.Load())
}
Expand All @@ -125,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), broker: func() processBroker { return noBroker{} }, procs: &fakeProcesses{}})
err := run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), procs: &fakeProcesses{}})
if !errors.Is(err, ErrExecutor) || !errors.Is(err, errFactory) {
t.Fatalf("Run = %v, want the factory's error as ErrExecutor", err)
}
Expand Down Expand Up @@ -159,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), broker: func() processBroker { return noBroker{} }, procs: &fakeProcesses{}})
err = run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), procs: &fakeProcesses{}})
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)
}
Expand Down
7 changes: 5 additions & 2 deletions apps/daemon/internal/agenthost/agenthost.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,11 +120,14 @@ var (
ErrWorld = errors.New("agenthost: world lost")
// ErrLaunch is a view that could not be launched.
ErrLaunch = errors.New("agenthost: launch failed")
// ErrProcessBroker is a process broker that could not start.
// ErrProcessBroker is a view's process broker that could not start, or
// whose process relay was lost while the view ran
// (processbroker.ErrRelayLost).
ErrProcessBroker = errors.New("agenthost: process broker failed")
// ErrTurn is a Turn that failed or left its Executor unusable.
ErrTurn = errors.New("agenthost: turn failed")
// ErrTeardown is a Session resource that could not be released.
// ErrTeardown is a Session resource that could not be released, such as
// a view whose teardown did not finish (sessionview.ErrCleanup).
ErrTeardown = errors.New("agenthost: teardown incomplete")
)

Expand Down
18 changes: 10 additions & 8 deletions apps/daemon/internal/agenthost/agenthost_linux_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"testing"

"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent"
"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/processshim"
"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/sessionview"
"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto"
"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto/prototest"
Expand All @@ -22,10 +23,17 @@ import (
"github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire"
)

// The test binary is also the privileged suite's Harness inside the view and
// its process with a zombie leader.
// The test binary is also the privileged suite's Harness inside the view,
// the view's shim and process relay, and its process with a zombie leader.
func TestMain(m *testing.M) {
sessionview.Init()
if processshim.Relaying() {
os.Exit(processshim.Relay())
}
// The shim runs with the Harness's environment, so it comes first.
if filepath.Base(os.Args[0]) == "sh" {
os.Exit(processshim.Run(processshim.SocketPath))
}
if os.Getenv(harnessEnv) != "" {
os.Exit(runHarness(os.Args[1:]))
}
Expand Down Expand Up @@ -97,12 +105,6 @@ func countingDial(n *atomic.Int32) dialFunc {
}
}

// noBroker is a process broker that serves nothing.
type noBroker struct{}

func (noBroker) Start(brokerConfig) error { return nil }
func (noBroker) Close() error { return nil }

// leftSessions lists what remains under the state directory's sessions.
func leftSessions(t *testing.T, cfg Config) []os.DirEntry {
t.Helper()
Expand Down
49 changes: 0 additions & 49 deletions apps/daemon/internal/agenthost/broker.go
View file
Open in desktop

This file was deleted.

45 changes: 27 additions & 18 deletions apps/daemon/internal/agenthost/doc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,18 +3,24 @@
// Session's Link attachment. It needs Linux; elsewhere Run returns
// ErrUnsupported.
//
// Run runs one Session. It admits the Session before any effect: the kind
// must declare an agent.View, and the request must use only what a view runs
// and no function tools, whose results Input cannot carry. It then allocates
// the Session uid, skipping each uid that a running thread holds as its real,
// effective, saved or file-system uid; this check only detects a conflict and
// never ends a process. It creates the Session directory under
// Config.StateDir, rewrites the request so the model provider and HTTP MCP
// reach the network only through the Session's gateway, and calls the view's
// Executor factory. The first ViewSession.Launch starts the process broker;
// each Launch builds one sessionview view, of which one at a time is live,
// over the world that worldfs serves from the attachment's File service, with
// the gateway listening in the view's network namespace.
// Run runs one Session. It admits the Session before any effect: the kind must
// declare an agent.View, the request must use only what a view runs and no
// function tools, whose results Input cannot carry, and when the view declares
// shim names, which run on the sandbox PATH, the Session's Environment must
// set PATH. It then allocates the Session uid, skipping each uid that a
// running thread holds as its real, effective, saved or file-system uid; this
// check only detects a conflict and never ends a process. It creates the
// Session directory under Config.StateDir, rewrites the request so the model
// provider and HTTP MCP reach the network only through the Session's gateway,
// and calls the view's Executor factory. Each ViewSession.Launch builds one
// sessionview view, of which one at a time is live, over the world that
// worldfs serves from the attachment's File service, with the gateway
// listening in the view's network namespace. A view with a shim gets its own
// process broker, started once the view runs and closed once it has ended. The
// broker runs the shims' commands over the attachment's Process service in the
// strongest scope the service declares, with the view's ForwardEnv and the
// Session's Environment, and cancels a forwarded process whose shim is lost
// with the launch's kill timeout as its grace.
//
// Each view presents the closure directories read-only and executable, the
// Session home read-write and noexec, the agent host's /etc/passwd, group,
Expand All @@ -25,19 +31,22 @@
// The agent host owns the Session's Link attachment: it opens each stream
// with the Session's binding, renews the lease and fails the Session when the
// relay closes the attachment, a Link request fails in a way that is not
// retryable, or the world is lost or did not stop cleanly, which leaves what
// the attachment holds uncertain. A failure cancels the running Turn and
// closes the live view. A view's end is settled before its clirunner.Process
// reports it: the gateway has stopped, the world's end is recorded and the
// view slot is free. Teardown releases, in order, the Executor, the view, the
// retryable, the world is lost or did not stop cleanly, which leaves what the
// attachment holds uncertain, a view's process relay is lost while the view
// runs, or a view's teardown does not finish within sessionview's bound. A
// failure cancels the running Turn and closes the live view. A view's end is
// settled before its clirunner.Process reports it: the gateway and the
// process broker have stopped, the world's end is recorded and the view slot
// is free. Teardown releases, in order, the Executor, the view with its
// process broker, the Link attachment, the Session directory and the uid;
// Run decides its result only afterwards, so a failure recorded during
// teardown counts, and from the close of the attachment on, what the Link
// reports changes nothing. When Executor.Close fails, teardown ends the
// views, which kills their processes, and retries Close once. If Close
// fails again, the Executor may still use the Session directory: Run returns
// ErrTeardown and keeps the directory, and the uid stays in use until the
// agent host exits.
// agent host exits. A view whose teardown did not finish keeps both the same
// way, because its processes may still run.
//
// Run drives each Turn as the daemon's dispatch drives a prepared execution.
// One output consumer starts before StartTurn and forwards the Turn's
Expand Down
Loading
Loading