Bind the Runtime to its Environment and serve one Session at a time

apps/daemon/internal/agent/codex/preparation_router_test.go

@@ -39,7 +39,7 @@ func TestPreparationRouterRetainsActualNativeChild(t *testing.T) {
 			if err := os.MkdirAll(workspace, 0700); err != nil {
 				t.Fatal(err)
 			}
-			binding, err := localworkspace.NewWithCapabilityDirectory(environment, session, workspace, filepath.Join(t.TempDir(), "capabilities"))
+			binding, err := localworkspace.NewWithCapabilityDirectory(environment, workspace, filepath.Join(t.TempDir(), "capabilities"))
 			if err != nil {
 				t.Fatal(err)
 			}

apps/daemon/internal/agent/mcode/tool_environment_test.go

@@ -44,7 +44,7 @@ func TestWorkspaceCredentialsRemainInRuntimeSnapshotAcrossReconnect(t *testing.T
 			req.AgentSessionID = "native-1"
 			write(source, `{"MCP_TEST_TOKEN":"changed","TOOL_SECRET":"changed"}`)
 		}
-		binding, err := localworkspace.NewWithCapabilityDirectory(environment, session, config.Directory, capabilityDirectory)
+		binding, err := localworkspace.NewWithCapabilityDirectory(environment, config.Directory, capabilityDirectory)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -97,7 +97,7 @@ func TestWorkspaceCredentialsRemainInRuntimeSnapshotAcrossReconnect(t *testing.T
 	if err := os.Remove(filepath.Join(initialization, "tool-env.json")); err != nil {
 		t.Fatal(err)
 	}
-	binding, err := localworkspace.NewWithCapabilityDirectory(environment, session, config.Directory, capabilityDirectory)
+	binding, err := localworkspace.NewWithCapabilityDirectory(environment, config.Directory, capabilityDirectory)
 	if err != nil {
 		t.Fatal(err)
 	}

apps/daemon/internal/agent/runtime_paths.go

@@ -2,6 +2,7 @@ package agent
 
 import (
 	"fmt"
+	"os"
 	"path/filepath"
 	"strings"
 )
@@ -49,3 +50,41 @@ func safeRuntimePathPart(value string) string {
 	}
 	return value
 }
+
+// RemoveSessionState deletes the state directory of every key but keepKey under
+// both roots, for every agent kind. It lists directories instead of tracking
+// Sessions, so it also covers state left by earlier processes and checkpoints.
+func RemoveSessionState(stateRoot, runtimeRoot, keepKey string) error {
+	keep := safeRuntimePathParts(keepKey)
+	if len(keep) != 1 {
+		return fmt.Errorf("agent: invalid agent state key %q", keepKey)
+	}
+	for _, root := range []string{stateRoot, runtimeRoot} {
+		if !filepath.IsAbs(root) || filepath.Clean(root) != root {
+			return fmt.Errorf("agent: state root must be a clean absolute directory")
+		}
+		kinds, err := os.ReadDir(filepath.Join(root, "runtime"))
+		if err != nil && !os.IsNotExist(err) {
+			return err
+		}
+		for _, kind := range kinds {
+			if !kind.IsDir() {
+				continue
+			}
+			states := filepath.Join(root, "runtime", kind.Name(), "state")
+			entries, err := os.ReadDir(states)
+			if err != nil && !os.IsNotExist(err) {
+				return err
+			}
+			for _, entry := range entries {
+				if entry.Name() == keep[0] {
+					continue
+				}
+				if err := os.RemoveAll(filepath.Join(states, entry.Name())); err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}

apps/daemon/internal/agent/runtime_paths_test.go

@@ -1,6 +1,7 @@
 package agent
 
 import (
+	"os"
 	"path/filepath"
 	"testing"
 )
@@ -31,3 +32,42 @@ func TestStateDirRequiresAgentState(t *testing.T) {
 		t.Fatalf("sanitized state = %q, %v; want %q", got, err, want)
 	}
 }
+
+func TestRemoveSessionStateKeepsOnlyTheNamedKey(t *testing.T) {
+	stateRoot, runtimeRoot := t.TempDir(), t.TempDir()
+	var kept, removed []string
+	for _, root := range []string{stateRoot, runtimeRoot} {
+		for _, kind := range []string{"codex", "claude-sdk"} {
+			keep, err := StateDir(root, kind, "keep")
+			if err != nil {
+				t.Fatal(err)
+			}
+			other, _ := StateDir(root, kind, "other")
+			for _, dir := range []string{keep, other} {
+				if err := os.MkdirAll(filepath.Join(dir, "history"), 0o700); err != nil {
+					t.Fatal(err)
+				}
+			}
+			kept, removed = append(kept, keep), append(removed, other)
+		}
+	}
+	if err := os.WriteFile(filepath.Join(stateRoot, "runtime", ".DS_Store"), nil, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if err := RemoveSessionState(stateRoot, runtimeRoot, "keep"); err != nil {
+		t.Fatal(err)
+	}
+	for _, dir := range kept {
+		if _, err := os.Stat(dir); err != nil {
+			t.Fatalf("kept state removed: %v", err)
+		}
+	}
+	for _, dir := range removed {
+		if _, err := os.Stat(dir); !os.IsNotExist(err) {
+			t.Fatalf("other state kept: %s", dir)
+		}
+	}
+	if err := RemoveSessionState(t.TempDir(), t.TempDir(), "keep"); err != nil {
+		t.Fatalf("empty roots: %v", err)
+	}
+}

apps/daemon/internal/cli/connect_bootstrap.go

@@ -19,7 +19,7 @@ func bootstrapProfile(path string) (*auth.Profile, bool, error) {
 	if err != nil {
 		return nil, false, err
 	}
-	if err = localworkspace.Bind(input.EnvironmentID, input.SessionID, input.Workspace, localworkspace.CapabilityDirectory); err != nil {
+	if err = localworkspace.Bind(input.EnvironmentID, input.Workspace, localworkspace.CapabilityDirectory); err != nil {
 		return nil, false, errors.New("connect: Runtime workspace binding unavailable")
 	}
 	return &auth.Profile{ServerURL: input.CoreURL, RuntimeID: input.DeviceID, RunnerCredential: input.Credential}, input.SuspendControl, nil

apps/daemon/internal/cli/connect_environment_binding.go

@@ -62,9 +62,10 @@ func bindEnvironmentRuntime(remote string, bound machineapi.Enrollment, workspac
 	if bound.WorkspaceDirectory != runtimebootstrap.PublicWorkspace && bound.WorkspaceDirectory != workspace {
 		return errors.New("connect: enrollment does not match the Runtime workspace")
 	}
-	if err := localworkspace.Bind(bound.EnvironmentID, bound.SessionID, workspace, capabilityDirectory); err != nil {
+	if err := localworkspace.Bind(bound.EnvironmentID, workspace, capabilityDirectory); err != nil {
 		return errors.New("connect: local Runtime layout unavailable")
 	}
+	bound.SessionID = "" // The Runtime binds to its Environment; the receipt's Session is not part of it.
 	return saveEnvironmentBinding(root, environmentBinding{RemoteURL: remote, Enrollment: bound, LocalWorkspace: workspace, CapabilityDirectory: capabilityDirectory})
 }
 

apps/daemon/internal/cli/connect_environment_binding_test.go

@@ -44,11 +44,14 @@ func TestBindEnvironmentRuntimeUsesCanonicalOperatorWorkspace(t *testing.T) {
 				t.Fatal(err)
 			}
 			var receipt environmentBinding
-			if decodeEnvironmentJSON(raw, &receipt) != nil || receipt.LocalWorkspace != workspace || receipt.Enrollment != bound {
+			want := bound
+			want.SessionID = ""
+			if decodeEnvironmentJSON(raw, &receipt) != nil || receipt.LocalWorkspace != workspace || receipt.Enrollment != want {
 				t.Fatal("workspace identity changed", receipt)
 			}
+			bound.SessionID = uuid.NewString()
 			if err := bindEnvironmentRuntime(receipt.RemoteURL, bound, workspace, localworkspace.CapabilityDirectory); err != nil {
-				t.Fatal("same binding refused", err)
+				t.Fatal("same Environment refused for another Session", err)
 			}
 			replacement := filepath.Join(filepath.Dir(workspace), "replacement")
 			if err := os.Mkdir(replacement, 0700); err != nil {

apps/daemon/internal/cli/connect_suspend_test.go

@@ -342,7 +342,7 @@ func TestQuiesceRetainsExecutorAcrossReconnectAndShutdownConfirmsCleanup(t *test
 	environment, session := uuid.NewString(), uuid.NewString()
 	t.Setenv("OAC_RUNTIME_HOME", t.TempDir())
 	t.Setenv("OAC_RUNTIME_STATE_DIRECTORY", t.TempDir())
-	if err := localworkspace.Bind(environment, session, t.TempDir(), t.TempDir()); err != nil {
+	if err := localworkspace.Bind(environment, t.TempDir(), t.TempDir()); err != nil {
 		t.Fatal(err)
 	}
 	t.Cleanup(localworkspace.Unbind)

apps/daemon/internal/cli/suspend_control_linux_test.go

@@ -22,7 +22,7 @@ func TestResumeControlRejectsStaleIdentityAndDiscardsPreParkSignals(t *testing.T
 	}
 	path := filepath.Join(dir, "control.json")
 	environment := uuid.NewString()
-	if err := localworkspace.Bind(environment, uuid.NewString(), t.TempDir(), t.TempDir()); err != nil {
+	if err := localworkspace.Bind(environment, t.TempDir(), t.TempDir()); err != nil {
 		t.Fatal(err)
 	}
 	t.Cleanup(localworkspace.Unbind)

apps/daemon/internal/dispatch/executor.go

@@ -101,8 +101,18 @@ func (r *Router) handleExecutorPrepare(ctx context.Context, env proto.Envelope,
 		return r.rejectPreparation(env, "preparation_capacity")
 	}
 	active := 0
+	// A Runtime bound to an Environment serves one Session at a time; the new
+	// executor closes the other Sessions' idle executors before it prepares.
+	var others []*executorState
 	for _, candidate := range r.executors {
-		if candidate.sessionID != input.SessionID && candidate.stateKey == req.AgentStateKey {
+		switch {
+		case candidate.sessionID == input.SessionID:
+		case r.localWorkspace != nil && (candidate.preparing || candidate.admission != nil || candidate.run != nil):
+			r.mu.Unlock()
+			return r.rejectPreparation(env, "resource_unavailable")
+		case r.localWorkspace != nil:
+			others = append(others, candidate)
+		case candidate.stateKey == req.AgentStateKey:
 			r.mu.Unlock()
 			return r.rejectPreparation(env, "session_binding_conflict")
 		}
@@ -169,12 +179,15 @@ func (r *Router) handleExecutorPrepare(ctx context.Context, env proto.Envelope,
 	if reused {
 		r.publishPreparation(p, status)
 	} else {
-		go r.prepareExecutor(p, req, factory)
+		go r.prepareExecutor(p, req, factory, others)
 	}
 	return nil
 }
 
-func (r *Router) prepareExecutor(p *preparationState, req proto.PromptRequestPayload, factory agent.ExecutorFactory) {
+// prepareExecutor confirms the close of the other Sessions' executors and
+// removes their native state, which also covers state left by expired
+// executors and earlier processes, before it prepares the new executor.
+func (r *Router) prepareExecutor(p *preparationState, req proto.PromptRequestPayload, factory agent.ExecutorFactory, others []*executorState) {
 	defer r.shutdownWG.Done()
 	owner := p.executor
 	started := time.Now()
@@ -184,15 +197,21 @@ func (r *Router) prepareExecutor(p *preparationState, req proto.PromptRequestPay
 	if !r.sendPreparation(p.requestID, p.trace, initial) {
 		r.abandonExecutorAdmission(p, "failed", "status_delivery_failed", false)
 	}
-	var native agent.Executor
-	var err error
-	if owner.ctx.Err() == nil {
-		req, err = r.localWorkspace.Prepare(owner.ctx, req)
-		if err == nil && owner.ctx.Err() == nil {
-			native, err = factory(owner.ctx, req)
+	err := owner.ctx.Err()
+	for _, other := range others {
+		if err == nil {
+			err = r.closeExecutor(other)
 		}
-	} else {
-		err = owner.ctx.Err()
+	}
+	if err == nil {
+		err = r.localWorkspace.RemoveOtherSessionState(req.AgentStateKey)
+	}
+	if err == nil {
+		req, err = r.localWorkspace.Prepare(owner.ctx, req)
+	}
+	var native agent.Executor
+	if err == nil && owner.ctx.Err() == nil {
+		native, err = factory(owner.ctx, req)
 	}
 	r.mu.Lock()
 	owner.native, owner.preparing = native, false

apps/daemon/internal/dispatch/executor_switch_test.go

@@ -0,0 +1,51 @@
+package dispatch_test
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent"
+	"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto"
+	"github.com/google/uuid"
+)
+
+func TestBoundRuntimeSwitchesSessionOnlyWhenOtherSessionIsIdle(t *testing.T) {
+	var made []*controlledPreparation
+	sender := &recSender{}
+	r := preparationRouter(t, sender, time.Minute, func(context.Context, proto.PromptRequestPayload) (preparedFixture, error) {
+		p := &controlledPreparation{closed: make(chan struct{})}
+		made = append(made, p)
+		return p, nil
+	})
+	root := os.Getenv("OAC_RUNTIME_HOME")
+	first, next := preparationRequest(), preparationRequest()
+	next.SessionID = uuid.NewString()
+	next.Configuration.AgentStateKey = next.SessionID
+	ready := executorAdmission(t, r, sender, "first", first)
+	var dirs [2]string
+	for i, id := range []string{first.SessionID, next.SessionID} {
+		dir, err := agent.StateDir(root, "codex", id)
+		if err != nil || os.MkdirAll(dir, 0o700) != nil {
+			t.Fatal(err)
+		}
+		dirs[i] = dir
+	}
+	_ = r.Handle(t.Context(), mustEnv(t, proto.TypeExecutionPrepare, "busy", next))
+	if got := waitPreparationStatus(t, sender, "busy", "rejected", ""); got.ErrorCode != "resource_unavailable" || len(made) != 1 {
+		t.Fatalf("admitted Session was displaced: %+v", got)
+	}
+	if err := r.Handle(t.Context(), mustEnv(t, proto.TypeExecutionRelease, "first", proto.ExecutionReleasePayload{Handle: ready.Handle})); err != nil {
+		t.Fatal(err)
+	}
+	waitPreparationStatus(t, sender, "first", "released", "")
+	executorAdmission(t, r, sender, "switch", next)
+	waitPreparationClosed(t, made[0])
+	if _, err := os.Stat(dirs[0]); !os.IsNotExist(err) {
+		t.Fatal("old Session state kept")
+	}
+	if _, err := os.Stat(dirs[1]); err != nil {
+		t.Fatal("new Session state removed")
+	}
+}

apps/daemon/internal/dispatch/executor_test.go

@@ -332,6 +332,16 @@ func TestExecutorRejectsSessionStateScopeMismatch(t *testing.T) {
 	}
 }
 
+func TestUnboundRuntimeKeepsSessionsSideBySide(t *testing.T) {
+	owner := &reusableExecutor{}
+	r, s, calls := executorRouter(t, owner, time.Minute)
+	executorAdmission(t, r, s, "first", executorRequest())
+	executorAdmission(t, r, s, "second", noEnvironmentPreparation("next", proto.PromptRequestPayload{AgentKind: "reusable"}))
+	if owner.closes.Load() != 0 || calls.Load() != 2 {
+		t.Fatalf("unbound Runtime closes=%d prepares=%d, want 0 and 2", owner.closes.Load(), calls.Load())
+	}
+}
+
 func TestExecutorRejectsUnsupportedConfigurationBeforeFactory(t *testing.T) {
 	for name, change := range map[string]func(*proto.PromptRequestPayload){
 		"no environment":       func(r *proto.PromptRequestPayload) { r.DisableExecutionEnvironment = false },

apps/daemon/internal/dispatch/local_directory_test.go

@@ -23,7 +23,7 @@ func TestLocalDirectoryPreparationNeedsNoHarnessAndRejectsOtherOwners(t *testing
 	workspace := t.TempDir()
 
 	environment, session := uuid.NewString(), uuid.NewString()
-	binding, err := localworkspace.NewWithCapabilityDirectory(environment, session, workspace, t.TempDir())
+	binding, err := localworkspace.NewWithCapabilityDirectory(environment, workspace, t.TempDir())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -66,11 +66,6 @@ func TestLocalDirectoryPreparationNeedsNoHarnessAndRejectsOtherOwners(t *testing
 	if err := r.Handle(t.Context(), mustEnv(t, proto.TypeExecutionStart, "idle", proto.ExecutionStartPayload{Handle: ready.Handle, ExecutorID: ready.ExecutorID, RunID: "forbidden", Input: proto.TextInput("work")})); err == nil {
 		t.Fatal("read preparation admitted execution")
 	}
-	bad := request
-	bad.AgentStateKey = uuid.NewString()
-	if err := r.Handle(t.Context(), mustEnv(t, proto.TypeExecutionPrepare, "wrong-session", proto.ExecutionPreparePayload{Configuration: bad})); err == nil {
-		t.Fatal("wrong Session accepted")
-	}
 	_ = r.Handle(t.Context(), mustEnv(t, proto.TypeExecutionRelease, "idle", proto.ExecutionReleasePayload{Handle: ready.Handle}))
 	waitPreparationStatus(t, sender, "idle", "released", "")
 	if harnessCalls.Load() != 0 || r.ActiveRuns() != 0 {
@@ -103,7 +98,7 @@ func TestLocalDirectoryKeepsNotDirectorySeparateFromFailures(t *testing.T) {
 		t.Fatal(err)
 	}
 	environment, session := uuid.NewString(), uuid.NewString()
-	binding, err := localworkspace.NewWithCapabilityDirectory(environment, session, workspace, t.TempDir())
+	binding, err := localworkspace.NewWithCapabilityDirectory(environment, workspace, t.TempDir())
 	if err != nil {
 		t.Fatal(err)
 	}

apps/daemon/internal/dispatch/native_file_results_test.go

@@ -38,7 +38,7 @@ func TestLocalUploadUnknownRetainsOwner(t *testing.T) {
 		t.Fatal(got)
 	}
 	r.workspaceWrite = &workspaceUpload{envelope: proto.Envelope{ID: uuid.NewString()}, finished: true, uncertain: true}
-	request := proto.WorkspaceWritePayload{Step: "begin", EnvironmentID: uuid.NewString(), SessionID: uuid.NewString(), Path: "file", SizeBytes: 0, SHA256: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"}
+	request := proto.WorkspaceWritePayload{Step: "begin", EnvironmentID: uuid.NewString(), Path: "file", SizeBytes: 0, SHA256: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"}
 	env, _ := proto.NewEnvelope(proto.TypeWorkspaceWrite, uuid.NewString(), request)
 	if err = r.Handle(t.Context(), env); err != nil {
 		t.Fatal(err)

apps/daemon/internal/dispatch/preparation_test.go

@@ -83,7 +83,7 @@ func preparationWorkspace(t *testing.T) *localworkspace.Binding {
 	}
 	t.Setenv("OAC_RUNTIME_HOME", runtimeHome)
 	preparationDirectories.workspace, preparationDirectories.capabilities = t.TempDir(), t.TempDir()
-	binding, err := localworkspace.NewWithCapabilityDirectory(preparationEnvironmentID, preparationSessionID, preparationDirectories.workspace, preparationDirectories.capabilities)
+	binding, err := localworkspace.NewWithCapabilityDirectory(preparationEnvironmentID, preparationDirectories.workspace, preparationDirectories.capabilities)
 	if err != nil {
 		t.Fatal(err)
 	}

apps/daemon/internal/dispatch/runtime_preparation.go

@@ -60,7 +60,7 @@ func (r *Router) handleRuntimePrepare(ctx context.Context, env proto.Envelope) e
 			}
 			return r.sendRuntimePrepareResult(ctx, env.ID, rejectedRuntimePreparation("runtime_preparation_capacity"))
 		}
-		if r.localWorkspace == nil || !r.localWorkspace.Matches(request.EnvironmentID, request.SessionID) || r.runtimePreparationResourcesBusyLocked() {
+		if r.localWorkspace == nil || !r.localWorkspace.Matches(request.EnvironmentID) || r.runtimePreparationResourcesBusyLocked() {
 			r.mu.Unlock()
 			return r.sendRuntimePrepareResult(ctx, env.ID, rejectedRuntimePreparation("resource_unavailable"))
 		}