From 988a7767fe161a3ff7bf3a1dbe2fb512a50af79a Mon Sep 17 00:00:00 2001 From: Haitao Pan Date: Mon, 6 Apr 2026 14:52:27 +0800 Subject: [PATCH] fix: freeze external provider routing for single-agent runs --- go/go_core/internal/acp/execution.go | 39 ++++++++++ .../internal/acp/providers_sync_test.go | 51 +++++++++++++ go/go_core/internal/acp/server.go | 74 ++++++++++++++++--- 3 files changed, 152 insertions(+), 12 deletions(-) diff --git a/go/go_core/internal/acp/execution.go b/go/go_core/internal/acp/execution.go index 0f9ba939..a5b6f059 100644 --- a/go/go_core/internal/acp/execution.go +++ b/go/go_core/internal/acp/execution.go @@ -15,6 +15,12 @@ import ( "xworkmate/go_core/internal/shared" ) +const ( + externalProviderEndpointKey = "externalProviderEndpoint" + externalProviderAuthorizationHeaderKey = "externalProviderAuthorizationHeader" + externalProviderLabelKey = "externalProviderLabel" +) + func buildResolvedExecutionParams( params map[string]any, resolved router.Result, @@ -49,6 +55,25 @@ func buildResolvedExecutionParams( return next } +func injectResolvedExternalProviderParams( + params map[string]any, + provider syncedProvider, +) map[string]any { + if params == nil { + params = map[string]any{} + } + if endpoint := strings.TrimSpace(provider.Endpoint); endpoint != "" { + params[externalProviderEndpointKey] = endpoint + } + if authorization := strings.TrimSpace(provider.AuthorizationHeader); authorization != "" { + params[externalProviderAuthorizationHeaderKey] = authorization + } + if label := strings.TrimSpace(provider.Label); label != "" { + params[externalProviderLabelKey] = label + } + return params +} + func (s *Server) runGateway( ctx context.Context, method string, @@ -125,6 +150,20 @@ func (s *Server) runSingleAgentViaExternalProvider( ) } +func externalProviderFromParams(params map[string]any) (syncedProvider, bool) { + endpoint := strings.TrimSpace(shared.StringArg(params, externalProviderEndpointKey, "")) + if endpoint == "" { + return syncedProvider{}, false + } + return syncedProvider{ + ProviderID: strings.TrimSpace(shared.StringArg(params, "provider", "")), + Label: strings.TrimSpace(shared.StringArg(params, externalProviderLabelKey, "")), + Endpoint: endpoint, + AuthorizationHeader: strings.TrimSpace(shared.StringArg(params, externalProviderAuthorizationHeaderKey, "")), + Enabled: true, + }, true +} + func requestExternalACP( ctx context.Context, endpoint, diff --git a/go/go_core/internal/acp/providers_sync_test.go b/go/go_core/internal/acp/providers_sync_test.go index bbe25e6e..de78591e 100644 --- a/go/go_core/internal/acp/providers_sync_test.go +++ b/go/go_core/internal/acp/providers_sync_test.go @@ -1,6 +1,7 @@ package acp import ( + "context" "encoding/json" "net/http" "net/http/httptest" @@ -148,3 +149,53 @@ func TestExecuteSessionTaskUsesSyncedExternalProvider(t *testing.T) { t.Fatalf("expected resolved provider claude, got %#v", response) } } + +func TestRunSingleAgentUsesFrozenExternalProviderParams(t *testing.T) { + externalServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/acp/rpc" { + http.NotFound(w, r) + return + } + defer r.Body.Close() + var request map[string]any + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + t.Fatalf("decode request: %v", err) + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "jsonrpc": "2.0", + "id": request["id"], + "result": map[string]any{ + "success": true, + "output": "frozen-provider-ok", + "turnId": "turn-frozen", + "provider": "custom-agent-1", + "mode": "single-agent", + }, + }) + })) + defer externalServer.Close() + + server := NewServer() + session := server.getOrCreateSession("session-frozen", "thread-frozen") + result := server.runSingleAgent( + context.Background(), + "session.start", + session, + map[string]any{ + "provider": "custom-agent-1", + "taskPrompt": "hello", + "workingDirectory": t.TempDir(), + externalProviderEndpointKey: externalServer.URL, + externalProviderAuthorizationHeaderKey: "Bearer test", + externalProviderLabelKey: "Codex", + }, + "turn-frozen", + func(map[string]any) {}, + ) + if result.err != nil { + t.Fatalf("expected success, got rpc error: %v", result.err) + } + if got := result.response["output"]; got != "frozen-provider-ok" { + t.Fatalf("expected frozen provider output, got %#v", result.response) + } +} diff --git a/go/go_core/internal/acp/server.go b/go/go_core/internal/acp/server.go index 83b78d1f..b6fda2b3 100644 --- a/go/go_core/internal/acp/server.go +++ b/go/go_core/internal/acp/server.go @@ -44,10 +44,10 @@ type taskResult struct { } type Server struct { - mu sync.Mutex - sessions map[string]*session - queues map[string]chan task - gateway *gatewayruntime.Manager + mu sync.Mutex + sessions map[string]*session + queues map[string]chan task + gateway *gatewayruntime.Manager providerCatalog map[string]syncedProvider } @@ -70,7 +70,7 @@ func Serve(args []string) error { server := NewServer() httpServer := &http.Server{ - Addr: strings.TrimSpace(*listen), + Addr: strings.TrimSpace(*listen), Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case "/acp/rpc": @@ -95,9 +95,9 @@ func Serve(args []string) error { func NewServer() *Server { return &Server{ - sessions: make(map[string]*session), - queues: make(map[string]chan task), - gateway: gatewayruntime.NewManager(), + sessions: make(map[string]*session), + queues: make(map[string]chan task), + gateway: gatewayruntime.NewManager(), providerCatalog: make(map[string]syncedProvider), } } @@ -556,10 +556,10 @@ func (s *Server) executeSessionTask(task task) (map[string]any, *shared.RPCError threadID := strings.TrimSpace(shared.StringArg(params, "threadId", sessionID)) if resolvedRouting.Unavailable { response := mergeRoutingResponse(map[string]any{ - "success": false, - "error": resolvedRouting.UnavailableMessage, - "unavailable": true, - "unavailableCode": resolvedRouting.UnavailableCode, + "success": false, + "error": resolvedRouting.UnavailableMessage, + "unavailable": true, + "unavailableCode": resolvedRouting.UnavailableCode, "unavailableMessage": resolvedRouting.UnavailableMessage, }, resolvedRouting) return response, nil @@ -567,6 +567,14 @@ func (s *Server) executeSessionTask(task task) (map[string]any, *shared.RPCError executionParams := buildResolvedExecutionParams(params, resolvedRouting) mode := strings.TrimSpace(shared.StringArg(executionParams, "mode", "single-agent")) provider := strings.TrimSpace(shared.StringArg(executionParams, "provider", "")) + if provider != "" { + if syncedProvider, ok := s.syncedProviderByID(provider); ok { + executionParams = injectResolvedExternalProviderParams( + executionParams, + syncedProvider, + ) + } + } session := s.getOrCreateSession(sessionID, threadID) session.mode = mode @@ -658,6 +666,48 @@ func (s *Server) runSingleAgent( prompt := strings.TrimSpace(shared.StringArg(params, "taskPrompt", "")) prompt = shared.AugmentPromptWithAttachments(prompt, params) + if syncedProvider, ok := externalProviderFromParams(params); ok { + response, err := s.runSingleAgentViaExternalProvider( + ctx, + syncedProvider, + method, + params, + notify, + ) + if err == nil { + result := asMap(response["result"]) + if len(result) == 0 { + result = response + } + if _, exists := result["provider"]; !exists { + result["provider"] = provider + } + if _, exists := result["mode"]; !exists { + result["mode"] = "single-agent" + } + if _, exists := result["turnId"]; !exists { + result["turnId"] = turnID + } + return taskResult{response: result} + } + s.emitSessionUpdate(session, notify, turnID, map[string]any{ + "type": "status", + "event": "completed", + "message": err.Error(), + "pending": false, + "error": true, + }) + return taskResult{ + response: map[string]any{ + "success": false, + "error": err.Error(), + "turnId": turnID, + "mode": "single-agent", + "provider": provider, + }, + } + } + if syncedProvider, ok := s.syncedProviderByID(provider); ok { response, err := s.runSingleAgentViaExternalProvider( ctx,