Let Subagents call function and MCP tools

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

@@ -14,9 +14,12 @@ import (
 	"github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto"
 )
 
-// This opt-in test uses the real pinned harness and an explicitly configured
-// Responses provider. Credentials are read from a private file, never arguments.
-func TestExecutorNativeReuse(t *testing.T) {
+// nativeAcceptanceConfig returns the pinned binary and the explicitly
+// configured Responses provider of an opt-in native test, and gives the test an
+// isolated Runtime home. Credentials are read from a private file, never
+// arguments.
+func nativeAcceptanceConfig(t *testing.T) (binary string, options map[string]any) {
+	t.Helper()
 	if os.Getenv("OAC_TEST_CODEX_EXECUTOR_NATIVE") != "1" {
 		t.Skip("requires an authorized native model acceptance environment")
 	}
@@ -45,13 +48,18 @@ func TestExecutorNativeReuse(t *testing.T) {
 	for _, name := range []string{"CODEX_EXEC_SERVER_URL", "CODEX_EXEC_SERVER_NOISE_REGISTRY_URL", "CODEX_EXEC_SERVER_NOISE_ENVIRONMENT_ID", "CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN"} {
 		t.Setenv(name, "")
 	}
+	return binary, map[string]any{"model": model, "model_provider": map[string]any{"base_url": endpoint, "protocol": "responses", "api_key": strings.TrimSpace(string(key))}}
+}
+
+func TestExecutorNativeReuse(t *testing.T) {
+	binary, options := nativeAcceptanceConfig(t)
 	cfg := defaultSessionConfig(testStateRoot(t))
 	cfg.codexBinary = binary
 	cfg.logger = slog.New(slog.DiscardHandler)
 	req := proto.PromptRequestPayload{
 		AgentKind: "codex", AgentStateKey: "executor-native",
 		DisableExecutionEnvironment: true, DisableSubagents: true, ObserveMessages: true,
-		AgentOptions:  map[string]any{"model": model, "model_provider": map[string]any{"base_url": endpoint, "protocol": "responses", "api_key": strings.TrimSpace(string(key))}},
+		AgentOptions:  options,
 		FunctionTools: []proto.FunctionTool{{Name: "hold", Description: "Wait until the host supplies a result.", Parameters: json.RawMessage("{\"type\":\"object\",\"properties\":{},\"additionalProperties\":false}")}},
 	}
 	ctx, cancel := context.WithTimeout(t.Context(), 4*time.Minute)

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

@@ -43,7 +43,7 @@ func (s *Session) settleExecutorTurn(startErr error) {
 	s.operationsClosed = true
 	s.operationMu.Unlock()
 	s.stopSteering()
-	s.stopFunctionCalls()
+	s.stopFunctionCalls(false)
 	if !s.nativeSettled.Load() || startErr != nil || !s.rpc.Alive() {
 		s.cancelFn()
 		s.settlementErr = errors.Join(s.settlementErr, s.rpc.Close())
@@ -117,7 +117,7 @@ func (s *Session) cancelExecutorTurn(ctx context.Context) error {
 		s.cancelReady = make(chan struct{})
 		go func() {
 			defer close(s.cancelReady)
-			s.stopFunctionCalls()
+			s.stopFunctionCalls(false)
 			turnID, active := s.stopSteering()
 			if !s.terminal.Load() && active {
 				if turnID == "" {
@@ -148,19 +148,8 @@ func (s *Session) cancelExecutorTurn(ctx context.Context) error {
 
 var _ agent.Turn = (*Session)(nil)
 
-// Server callbacks retain their originating Turn, whose ID must match exactly.
 func (s *Session) onServerRequest(method string, handler ServerRequestHandler) {
 	s.rpc.OnServerRequest(method, func(raw json.RawMessage, id any) (any, error) {
-		if s.executor != nil {
-			var scope struct {
-				ThreadID string `json:"threadId"`
-				TurnID   string `json:"turnId"`
-			}
-			if json.Unmarshal(raw, &scope) != nil || !s.isRootThread(scope.ThreadID) || s.terminal.Load() ||
-				scope.TurnID == "" || !s.isRootTurn(scope.ThreadID, scope.TurnID) {
-				return nil, errors.New("codex: request does not belong to the active turn")
-			}
-		}
 		if !s.beginOperation() {
 			return nil, errors.New("codex: turn has settled")
 		}

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

@@ -137,7 +137,7 @@ func TestFunctionReceiptResultIsFrozenAndSubmittedOnce(t *testing.T) {
 	if err := s.SubmitFunctionResult(t.Context(), result); err == nil {
 		t.Fatal("settled result admitted again")
 	}
-	s.stopFunctionCalls()
+	s.stopFunctionCalls(false)
 }
 
 func TestFunctionReceiptShutdownDoesNotConfirmApplication(t *testing.T) {
@@ -151,7 +151,7 @@ func TestFunctionReceiptShutdownDoesNotConfirmApplication(t *testing.T) {
 			case "run cancellation":
 				s.cancelFn()
 			case "terminal":
-				s.stopFunctionCalls()
+				s.stopFunctionCalls(false)
 			}
 			select {
 			case err := <-done:
@@ -161,7 +161,7 @@ func TestFunctionReceiptShutdownDoesNotConfirmApplication(t *testing.T) {
 			case <-time.After(time.Second):
 				t.Fatal("shutdown did not settle receipt")
 			}
-			s.stopFunctionCalls()
+			s.stopFunctionCalls(false)
 		})
 	}
 }
@@ -199,7 +199,7 @@ func TestFunctionReceiptDeadlineEndsUncertainExecution(t *testing.T) {
 	if s.cancelCtx.Err() == nil {
 		t.Fatal("uncertain native execution remains active")
 	}
-	s.stopFunctionCalls()
+	s.stopFunctionCalls(false)
 }
 
 // The receipt and deadline are both ready when the waiter resumes. Either select
@@ -211,7 +211,7 @@ func TestFunctionReceiptWinsSimultaneousDeadline(t *testing.T) {
 		s.setThreadID("thread")
 		s.startSteering("thread", "turn")
 		s.functions, _ = prepareFunctionTools(nil)
-		pending := &pendingFunction{turnID: "turn", name: "lookup", reply: &functionReply{Success: true, ContentItems: []functionContent{}}, receipt: make(chan error, 1)}
+		pending := &pendingFunction{thread: "thread", turnID: "turn", name: "lookup", reply: &functionReply{Success: true, ContentItems: []functionContent{}}, receipt: make(chan error, 1)}
 		s.functions.pending["call"] = pending
 		s.confirmFunctionResult(nativeFunctionReceipt("thread", "turn", "call", "lookup", "true", "completed", "[]"))
 		ctx, cancel := context.WithCancel(t.Context())
@@ -254,11 +254,11 @@ func TestFunctionReceiptSurvivesObservationBackpressure(t *testing.T) {
 				cancel()
 			case "cancel":
 				s.cancelled.Store(true)
-				s.stopFunctionCalls()
+				s.stopFunctionCalls(false)
 				s.cancelFn()
 			case "terminal":
 				s.terminal.Store(true)
-				s.stopFunctionCalls()
+				s.stopFunctionCalls(false)
 			}
 			select {
 			case err := <-receipt:

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

@@ -18,8 +18,11 @@ type functionReply struct {
 }
 
 type pendingFunction struct {
-	rpcID   any
-	turnID  string
+	rpcID  any
+	thread string
+	turnID string
+	// held is the Subagent call that waits for its child Turn fact.
+	held    *proto.Envelope
 	name    string
 	reply   *functionReply
 	receipt chan error
@@ -51,7 +54,7 @@ func (s *Session) SubmitFunctionResult(ctx context.Context, result proto.Functio
 	}
 	s.functions.mu.Lock()
 	pending := s.functions.pending[result.CallID]
-	if pending == nil || pending.reply != nil || s.functions.closed || s.cancelled.Load() || s.terminal.Load() {
+	if pending == nil || pending.reply != nil || s.functions.closed || s.cancelled.Load() || (pending.thread == s.currentThreadID() && s.terminal.Load()) {
 		s.functions.mu.Unlock()
 		return agent.ErrUnknownFunctionCall
 	}
@@ -129,13 +132,13 @@ func (s *Session) confirmFunctionResult(raw json.RawMessage) {
 		TurnID   string                `json:"turnId"`
 		Item     toolObservationSource `json:"item"`
 	}
-	if json.Unmarshal(raw, &event) != nil || event.Item.Type != "dynamicToolCall" || !s.isRootTurn(event.ThreadID, event.TurnID) {
+	if json.Unmarshal(raw, &event) != nil || event.Item.Type != "dynamicToolCall" {
 		return
 	}
 	s.functions.mu.Lock()
 	defer s.functions.mu.Unlock()
 	pending := s.functions.pending[event.Item.ID]
-	if pending == nil || pending.reply == nil || s.functions.closed || s.cancelled.Load() {
+	if pending == nil || pending.thread != event.ThreadID || pending.turnID != event.TurnID || pending.reply == nil || s.functions.closed || s.cancelled.Load() {
 		return
 	}
 	item := event.Item
@@ -144,7 +147,7 @@ func (s *Session) confirmFunctionResult(raw json.RawMessage) {
 		status = "failed"
 	}
 	var err error
-	if pending.turnID != event.TurnID || pending.name != item.Tool || item.Namespace != "" || item.Status != status || item.Success == nil || *item.Success != pending.reply.Success || item.ContentItems == nil || !reflect.DeepEqual(*item.ContentItems, pending.reply.ContentItems) {
+	if pending.name != item.Tool || item.Namespace != "" || item.Status != status || item.Success == nil || *item.Success != pending.reply.Success || item.ContentItems == nil || !reflect.DeepEqual(*item.ContentItems, pending.reply.ContentItems) {
 		err = errors.New("codex: native function receipt does not match submitted result")
 	}
 	delete(s.functions.pending, item.ID)

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

@@ -1,6 +1,7 @@
 package codex
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"strings"
@@ -51,21 +52,38 @@ func (s *Session) handleFunctionCall(raw json.RawMessage, rpcID any) (any, error
 	if err := json.Unmarshal(raw, &call); err != nil {
 		return nil, err
 	}
-	if s.functions == nil || !s.functions.names[call.Tool] || call.Namespace != nil || call.CallID == "" || !s.isRootTurn(call.ThreadID, call.TurnID) || !json.Valid(call.Arguments) {
+	root := s.isRootTurn(call.ThreadID, call.TurnID)
+	child := !root && s.subagents != nil && call.TurnID != "" && call.ThreadID != "" && !s.isRootThread(call.ThreadID)
+	if s.functions == nil || !s.functions.names[call.Tool] || call.Namespace != nil || call.CallID == "" || !(root || child) || !json.Valid(call.Arguments) {
 		return nil, errors.New("unexpected function call")
 	}
+	payload := proto.FunctionCallPayload{CallID: call.CallID, Name: call.Tool, Arguments: call.Arguments}
+	if child {
+		payload.NativeID, payload.TurnID = call.ThreadID, call.TurnID
+	}
+	env, err := proto.NewEnvelope(proto.TypeFunctionCall, s.runID, payload)
+	if err != nil {
+		return nil, err
+	}
 	s.functions.mu.Lock()
-	if s.functions.closed || s.cancelled.Load() || s.terminal.Load() || s.functions.pending[call.CallID] != nil || len(s.functions.pending) >= 64 {
+	if s.functions.closed || s.cancelled.Load() || (root && s.terminal.Load()) || s.functions.pending[call.CallID] != nil || len(s.functions.pending) >= 64 {
 		s.functions.mu.Unlock()
 		return nil, errors.New("function call cannot be admitted")
 	}
-	s.functions.pending[call.CallID] = &pendingFunction{rpcID: rpcID, turnID: call.TurnID, name: call.Tool, receipt: make(chan error, 1)}
+	pending := &pendingFunction{rpcID: rpcID, thread: call.ThreadID, turnID: call.TurnID, name: call.Tool, receipt: make(chan error, 1)}
+	if child {
+		pending.held = &env
+	}
+	s.functions.pending[call.CallID] = pending
 	s.functions.mu.Unlock()
-	env, err := proto.NewEnvelope(proto.TypeFunctionCall, s.runID, proto.FunctionCallPayload{CallID: call.CallID, Name: call.Tool, Arguments: call.Arguments})
-	if err == nil {
-		err = s.sendFunctionCall(env)
+	if child {
+		select {
+		case s.subagents.wake <- struct{}{}:
+		default:
+		}
+		return DeferReply, nil
 	}
-	if err != nil {
+	if err = s.sendFunctionCall(env); err != nil {
 		s.functions.mu.Lock()
 		delete(s.functions.pending, call.CallID)
 		s.functions.mu.Unlock()
@@ -98,14 +116,72 @@ type functionContent struct {
 	ImageURL *string `json:"imageUrl,omitempty"`
 }
 
-func (s *Session) stopFunctionCalls() {
+// stopFunctionCalls ends every pending call. With rootOnly it ends the root's
+// calls and keeps accepting those of Subagents, which outlive the root Turn.
+func (s *Session) stopFunctionCalls(rootOnly bool) {
 	if s.functions != nil {
 		s.functions.mu.Lock()
-		s.functions.closed = true
-		for _, pending := range s.functions.pending {
-			pending.receipt <- agent.ErrUnknownFunctionCall
+		s.functions.closed = s.functions.closed || !rootOnly
+		for id, pending := range s.functions.pending {
+			if !rootOnly || pending.thread == s.currentThreadID() {
+				pending.receipt <- agent.ErrUnknownFunctionCall
+				delete(s.functions.pending, id)
+			}
 		}
-		clear(s.functions.pending)
 		s.functions.mu.Unlock()
 	}
 }
+
+// withdrawChildCalls removes the pending calls of a child Turn that ended and
+// reports them to Core as withdrawn. Native drops their requests without a
+// reply, so no result can be applied. A call still held was never reported.
+func (s *Session) withdrawChildCalls(ctx context.Context, thread, turn string) error {
+	if s.functions == nil {
+		return nil
+	}
+	var withdrawn []proto.Envelope
+	s.functions.mu.Lock()
+	for id, pending := range s.functions.pending {
+		if pending.thread != thread || pending.turnID != turn {
+			continue
+		}
+		delete(s.functions.pending, id)
+		pending.receipt <- agent.ErrUnknownFunctionCall
+		if pending.held != nil {
+			continue
+		}
+		env, err := proto.NewEnvelope(proto.TypeFunctionCallWithdrawn, s.runID, proto.FunctionCallWithdrawnPayload{CallID: id, NativeID: thread, TurnID: turn})
+		if err != nil {
+			s.functions.mu.Unlock()
+			return err
+		}
+		withdrawn = append(withdrawn, env)
+	}
+	s.functions.mu.Unlock()
+	for _, env := range withdrawn {
+		if !s.sendWithin(ctx, env) {
+			return errors.New("codex: function call withdrawal delivery unavailable")
+		}
+	}
+	return nil
+}
+
+// sendHeldFunctionCalls sends the Subagent calls whose child Turn Core already
+// has, because Core resolves a child's call through its identity.
+func (s *Session) sendHeldFunctionCalls(ctx context.Context) {
+	if s.functions == nil {
+		return
+	}
+	var ready []proto.Envelope
+	s.functions.mu.Lock()
+	for _, pending := range s.functions.pending {
+		if _, sent := s.subagents.sent[proto.TypeSubagentTurn+":"+pending.thread+":"+pending.turnID]; sent && pending.held != nil {
+			ready = append(ready, *pending.held)
+			pending.held = nil
+		}
+	}
+	s.functions.mu.Unlock()
+	for _, env := range ready {
+		s.sendWithin(ctx, env)
+	}
+}