feat(rest-api): dispatch every Flow endpoint through the gRPC proxy

rest-api/AGENTS.md

@@ -274,9 +274,11 @@ main patterns:
 - Flow-backed inventory and task APIs use Flow request/response protobufs in the
   API model layer and keep target-shape helpers next to the model or handler
   that owns the REST shape. Use Rack, Tray, Task, and Task Rule as references.
-  Thin unary Flow pass-throughs should use `handler/util/common.ExecuteFlowGRPC`
-  rather than a bespoke Temporal workflow per method. Switching an existing
-  endpoint over spans releases; see the skill for the required order.
+  Every Flow-backed endpoint dispatches through the generic proxy, so use
+  `handler/util/common.ProxyFlowGRPC` (or `ExecuteFlowGRPC` where the caller
+  must return a plain `error`) rather than adding a bespoke Temporal workflow
+  per method. Retiring the bespoke workflows spans releases; see the skill for
+  the required order.
 - Curated REST endpoints that call NICo Core `forge.Forge` unary methods should
   use `handler/util/common.ExecuteCoreGRPC` with a typed protobuf request. Do
   not create a bespoke Temporal workflow for a simple unary Core call. BMC

rest-api/api/pkg/api/handler/grpcproxy_test.go

@@ -0,0 +1,82 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package handler
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	tClient "go.temporal.io/sdk/client"
+	tmocks "go.temporal.io/sdk/mocks"
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/NVIDIA/infra-controller/rest-api/common/pkg/grpcproxy"
+)
+
+// testFlowProxyReply makes run resolve to msg the way the site's Flow proxy
+// activity does, as protojson inside a grpcproxy.Response rather than as the
+// bare response proto.
+func testFlowProxyReply(t *testing.T, run *tmocks.WorkflowRun, msg proto.Message) {
+	t.Helper()
+
+	responseJSON, err := protojson.Marshal(msg)
+	require.NoError(t, err)
+
+	run.Mock.On("Get", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
+		args.Get(1).(*grpcproxy.Response).ResponseJSON = responseJSON
+	}).Return(nil)
+}
+
+// testFlowProxyDispatch mocks the Flow proxy workflow start, asserting that the
+// handler asked for fullMethod, and returns a pointer to the options it was
+// started with so a test can also assert the derived workflow ID and conflict
+// policy. Checking the method here is what catches an endpoint wired to the
+// wrong Flow call, which the workflow ID alone cannot show.
+func testFlowProxyDispatch(t *testing.T, mockTC *tmocks.Client, run *tmocks.WorkflowRun, fullMethod string, execErr error) *tClient.StartWorkflowOptions {
+	t.Helper()
+
+	started := &tClient.StartWorkflowOptions{}
+	mockTC.Mock.On("ExecuteWorkflow", mock.Anything, mock.Anything, grpcproxy.Flow.WorkflowName, mock.Anything).
+		Run(func(args mock.Arguments) {
+			*started = args.Get(1).(tClient.StartWorkflowOptions)
+			assert.Equal(t, fullMethod, args.Get(3).(grpcproxy.Request).FullMethod)
+		}).
+		Return(run, execErr)
+	return started
+}
+
+// testFlowProxyMethodDispatch mocks the Flow proxy workflow start for one Flow
+// method, routing on the method instead of asserting it. A handler that makes
+// several Flow calls needs each to resolve to its own reply, which
+// testFlowProxyDispatch cannot do because it matches any method. inspect, when
+// non-nil, runs on the dispatched arguments so a test can decode the request.
+func testFlowProxyMethodDispatch(t *testing.T, mockTC *tmocks.Client, run *tmocks.WorkflowRun, fullMethod string, inspect func(mock.Arguments)) *tClient.StartWorkflowOptions {
+	t.Helper()
+
+	started := &tClient.StartWorkflowOptions{}
+	mockTC.Mock.On("ExecuteWorkflow", mock.Anything, mock.Anything, grpcproxy.Flow.WorkflowName,
+		mock.MatchedBy(func(req grpcproxy.Request) bool { return req.FullMethod == fullMethod })).
+		Run(func(args mock.Arguments) {
+			*started = args.Get(1).(tClient.StartWorkflowOptions)
+			if inspect != nil {
+				inspect(args)
+			}
+		}).
+		Return(run, nil)
+	return started
+}
+
+// testFlowProxyRequest decodes the Flow request the handler sent through the
+// proxy into req, so tests can assert on the fields the transport carries as
+// protojson rather than as a typed proto argument.
+func testFlowProxyRequest(t *testing.T, args mock.Arguments, req proto.Message) {
+	t.Helper()
+
+	proxyReq, ok := args.Get(3).(grpcproxy.Request)
+	require.True(t, ok, "workflow arg must be a grpcproxy.Request")
+	require.NoError(t, protojson.Unmarshal(proxyReq.RequestJSON, req))
+}

rest-api/api/pkg/api/handler/machinepower.go

@@ -23,11 +23,7 @@ import (
 )
 
 func logAPIError(logger zerolog.Logger, apiErr *cutil.APIError, msg string) {
-	if apiErr.Data != nil {
-		logger.Error().Err(apiErr.Data).Msg(msg)
-		return
-	}
-	logger.Error().Err(apiErr).Msg(msg)
+	logger.Error().Err(apiErr.Diagnosis()).Msg(msg)
 }
 
 type MachinePowerControlHandler struct {

rest-api/api/pkg/api/handler/rack.go

@@ -4,7 +4,6 @@
 package handler
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -17,7 +16,6 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	temporalEnums "go.temporal.io/api/enums/v1"
 	tClient "go.temporal.io/sdk/client"
-	tp "go.temporal.io/sdk/temporal"
 
 	"github.com/NVIDIA/infra-controller/rest-api/api/internal/config"
 	"github.com/NVIDIA/infra-controller/rest-api/api/pkg/api/handler/util/common"
@@ -29,7 +27,6 @@ import (
 	cdb "github.com/NVIDIA/infra-controller/rest-api/db/pkg/db"
 	cdbm "github.com/NVIDIA/infra-controller/rest-api/db/pkg/db/model"
 	flowv1 "github.com/NVIDIA/infra-controller/rest-api/proto/flow/gen/v1"
-	"github.com/NVIDIA/infra-controller/rest-api/workflow/pkg/queue"
 )
 
 // ~~~~~ Get Rack Handler ~~~~~ //
@@ -161,35 +158,15 @@ func (grh GetRackHandler) Handle(c echo.Context) error {
 	}
 
 	// Execute workflow
-	workflowOptions := tClient.StartWorkflowOptions{
-		ID:                       fmt.Sprintf("rack-get-%s", rackStrID),
-		WorkflowIDReusePolicy:    temporalEnums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
-		WorkflowIDConflictPolicy: temporalEnums.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
-		WorkflowExecutionTimeout: cutil.WorkflowExecutionTimeout,
-		TaskQueue:                queue.SiteTaskQueue,
-	}
-
-	ctx, cancel := context.WithTimeout(ctx, cutil.WorkflowContextTimeout)
-	defer cancel()
-
-	we, err := stc.ExecuteWorkflow(ctx, workflowOptions, "GetRack", flowRequest)
-	if err != nil {
-		logger.Error().Err(err).Msg("failed to execute GetRack workflow")
-		return cutil.NewAPIErrorResponse(c, http.StatusInternalServerError, "Failed to get Rack details", nil)
-	}
-
-	// Get workflow result
 	var flowResponse flowv1.GetRackInfoResponse
-	err = we.Get(ctx, &flowResponse)
-	if err != nil {
-		var timeoutErr *tp.TimeoutError
-		if errors.As(err, &timeoutErr) || err == context.DeadlineExceeded || ctx.Err() != nil {
-			return common.TerminateWorkflowOnTimeOut(c, logger, stc, fmt.Sprintf("rack-get-%s", rackStrID), err, "Rack", "GetRack")
-		}
-		code, err := common.UnwrapWorkflowError(err)
-		logger.Error().Err(err).Msg("failed to get result from GetRack workflow")
-
-		return cutil.NewAPIErrorResponse(c, code, fmt.Sprintf("Failed to get Rack details: %s", err), nil)
+	proxyErr := common.ProxyFlowGRPC(
+		ctx, c, logger, stc,
+		flowv1.Flow_GetRackInfoByID_FullMethodName,
+		flowRequest, &flowResponse,
+		common.FlowWorkflowID(fmt.Sprintf("rack-get-%s", rackStrID)), temporalEnums.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
+	)
+	if proxyErr != nil {
+		return proxyErr
 	}
 
 	// Convert to API model
@@ -367,35 +344,15 @@ func (garh GetAllRackHandler) Handle(c echo.Context) error {
 	workflowID := fmt.Sprintf("rack-get-all-%s", common.QueryParamHash(apiRequest.QueryValues()))
 
 	// Execute workflow
-	workflowOptions := tClient.StartWorkflowOptions{
-		ID:                       workflowID,
-		WorkflowIDReusePolicy:    temporalEnums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
-		WorkflowIDConflictPolicy: temporalEnums.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
-		WorkflowExecutionTimeout: cutil.WorkflowExecutionTimeout,
-		TaskQueue:                queue.SiteTaskQueue,
-	}
-
-	ctx, cancel := context.WithTimeout(ctx, cutil.WorkflowContextTimeout)
-	defer cancel()
-
-	we, err := stc.ExecuteWorkflow(ctx, workflowOptions, "GetRacks", flowRequest)
-	if err != nil {
-		logger.Error().Err(err).Msg("failed to execute GetRacks workflow")
-		return cutil.NewAPIErrorResponse(c, http.StatusInternalServerError, "Failed to get Racks", nil)
-	}
-
-	// Get workflow result
 	var flowResponse flowv1.GetListOfRacksResponse
-	err = we.Get(ctx, &flowResponse)
-	if err != nil {
-		var timeoutErr *tp.TimeoutError
-		if errors.As(err, &timeoutErr) || err == context.DeadlineExceeded || ctx.Err() != nil {
-			return common.TerminateWorkflowOnTimeOut(c, logger, stc, workflowID, err, "Rack", "GetRacks")
-		}
-		code, err := common.UnwrapWorkflowError(err)
-		logger.Error().Err(err).Msg("failed to get result from GetRacks workflow")
-
-		return cutil.NewAPIErrorResponse(c, code, fmt.Sprintf("Failed to get Racks: %s", err), nil)
+	proxyErr := common.ProxyFlowGRPC(
+		ctx, c, logger, stc,
+		flowv1.Flow_GetListOfRacks_FullMethodName,
+		flowRequest, &flowResponse,
+		common.FlowWorkflowID(workflowID), temporalEnums.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
+	)
+	if proxyErr != nil {
+		return proxyErr
 	}
 
 	// Convert to API model
@@ -553,35 +510,15 @@ func (vrh ValidateRackHandler) Handle(c echo.Context) error {
 	}
 
 	// Execute workflow
-	workflowOptions := tClient.StartWorkflowOptions{
-		ID:                       fmt.Sprintf("rack-validate-%s", rackStrID),
-		WorkflowIDReusePolicy:    temporalEnums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
-		WorkflowIDConflictPolicy: temporalEnums.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
-		WorkflowExecutionTimeout: cutil.WorkflowExecutionTimeout,
-		TaskQueue:                queue.SiteTaskQueue,
-	}
-
-	ctx, cancel := context.WithTimeout(ctx, cutil.WorkflowContextTimeout)
-	defer cancel()
-
-	we, err := stc.ExecuteWorkflow(ctx, workflowOptions, "ValidateRackComponents", flowRequest)
-	if err != nil {
-		logger.Error().Err(err).Msg("failed to execute ValidateComponents workflow")
-		return cutil.NewAPIErrorResponse(c, http.StatusInternalServerError, "Failed to validate Rack", nil)
-	}
-
-	// Get workflow result
 	var flowResponse flowv1.ValidateComponentsResponse
-	err = we.Get(ctx, &flowResponse)
-	if err != nil {
-		var timeoutErr *tp.TimeoutError
-		if errors.As(err, &timeoutErr) || err == context.DeadlineExceeded || ctx.Err() != nil {
-			return common.TerminateWorkflowOnTimeOut(c, logger, stc, fmt.Sprintf("rack-validate-%s", rackStrID), err, "Rack", "ValidateRackComponents")
-		}
-		code, err := common.UnwrapWorkflowError(err)
-		logger.Error().Err(err).Msg("failed to get result from ValidateComponents workflow")
-
-		return cutil.NewAPIErrorResponse(c, code, fmt.Sprintf("Failed to validate Rack: %s", err), nil)
+	proxyErr := common.ProxyFlowGRPC(
+		ctx, c, logger, stc,
+		flowv1.Flow_ValidateComponents_FullMethodName,
+		flowRequest, &flowResponse,
+		common.FlowWorkflowID(fmt.Sprintf("rack-validate-%s", rackStrID)), temporalEnums.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
+	)
+	if proxyErr != nil {
+		return proxyErr
 	}
 
 	// Convert to API model
@@ -718,35 +655,15 @@ func (vrsh ValidateRacksHandler) Handle(c echo.Context) error {
 	workflowID := fmt.Sprintf("rack-validate-all-%s", common.QueryParamHash(apiRequest.QueryValues()))
 
 	// Execute workflow
-	workflowOptions := tClient.StartWorkflowOptions{
-		ID:                       workflowID,
-		WorkflowIDReusePolicy:    temporalEnums.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
-		WorkflowIDConflictPolicy: temporalEnums.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
-		WorkflowExecutionTimeout: cutil.WorkflowExecutionTimeout,
-		TaskQueue:                queue.SiteTaskQueue,
-	}
-
-	ctx, cancel := context.WithTimeout(ctx, cutil.WorkflowContextTimeout)
-	defer cancel()
-
-	we, err := stc.ExecuteWorkflow(ctx, workflowOptions, "ValidateRackComponents", flowRequest)
-	if err != nil {
-		logger.Error().Err(err).Msg("failed to execute ValidateComponents workflow")
-		return cutil.NewAPIErrorResponse(c, http.StatusInternalServerError, "Failed to validate Racks", nil)
-	}
-
-	// Get workflow result
 	var flowResponse flowv1.ValidateComponentsResponse
-	err = we.Get(ctx, &flowResponse)
-	if err != nil {
-		var timeoutErr *tp.TimeoutError
-		if errors.As(err, &timeoutErr) || err == context.DeadlineExceeded || ctx.Err() != nil {
-			return common.TerminateWorkflowOnTimeOut(c, logger, stc, workflowID, err, "Rack", "ValidateRackComponents")
-		}
-		code, err := common.UnwrapWorkflowError(err)
-		logger.Error().Err(err).Msg("failed to get result from ValidateComponents workflow")
-
-		return cutil.NewAPIErrorResponse(c, code, fmt.Sprintf("Failed to validate Racks: %s", err), nil)
+	proxyErr := common.ProxyFlowGRPC(
+		ctx, c, logger, stc,
+		flowv1.Flow_ValidateComponents_FullMethodName,
+		flowRequest, &flowResponse,
+		common.FlowWorkflowID(workflowID), temporalEnums.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
+	)
+	if proxyErr != nil {
+		return proxyErr
 	}
 
 	// Convert to API model