[fix][evaluation] fix export URL encoding, add upload retry, trajectory lower-bound, goroutine leak & long-conn target validation

backend/infra/backoff/backoff.go

@@ -58,7 +58,7 @@ func RetryWithElapsedTime(ctx context.Context, maxElapsedTime time.Duration, fn
 	return backoffFn(ctx, fn, policy)
 }
 
-func backoffFn(ctx context.Context, fn func() error, policy *backoff.ExponentialBackOff) error {
+func backoffFn(ctx context.Context, fn func() error, policy backoff.BackOff) error {
 	ctxWithCancel, cancelFn := context.WithCancel(ctx)
 	defer cancelFn()
 
@@ -70,3 +70,10 @@ func backoffFn(ctx context.Context, fn func() error, policy *backoff.Exponential
 func RetryWithMaxTimes(ctx context.Context, max int, fn func() error) error {
 	return backoff.Retry(fn, backoff.WithMaxRetries(&backoff.ZeroBackOff{}, uint64(max)))
 }
+
+// RetryWithMaxTimesAndInterval 以固定间隔 interval 最多重试 maxRetries 次（即总尝试次数为 maxRetries+1），
+// 适用于依赖偶发抖动（如对象存储/上传服务单实例迁移）时的止血重试：固定间隔而非指数退避，次数可控。
+func RetryWithMaxTimesAndInterval(ctx context.Context, maxRetries int, interval time.Duration, fn func() error) error {
+	policy := backoff.WithMaxRetries(backoff.NewConstantBackOff(interval), uint64(maxRetries))
+	return backoffFn(ctx, fn, policy)
+}

backend/infra/backoff/backoff_test.go

@@ -65,3 +65,52 @@ func TestRetryWithMaxTimes(t *testing.T) {
 		assert.Equal(t, 4, count)
 	})
 }
+
+func TestRetryWithMaxTimesAndInterval(t *testing.T) {
+	ctx := context.Background()
+
+	t.Run("succeeds after retries within max", func(t *testing.T) {
+		attempts := 0
+		err := RetryWithMaxTimesAndInterval(ctx, 2, 10*time.Millisecond, func() error {
+			attempts++
+			if attempts < 3 {
+				return fmt.Errorf("transient err")
+			}
+			return nil
+		})
+		assert.NoError(t, err)
+		assert.Equal(t, 3, attempts) // 1 initial + 2 retries
+	})
+
+	t.Run("fails after exhausting max retries", func(t *testing.T) {
+		attempts := 0
+		err := RetryWithMaxTimesAndInterval(ctx, 2, 10*time.Millisecond, func() error {
+			attempts++
+			return fmt.Errorf("always err")
+		})
+		assert.Error(t, err)
+		assert.Equal(t, 3, attempts) // 1 initial + 2 retries, then give up
+	})
+
+	t.Run("succeeds on first try, no retry", func(t *testing.T) {
+		attempts := 0
+		err := RetryWithMaxTimesAndInterval(ctx, 3, 10*time.Millisecond, func() error {
+			attempts++
+			return nil
+		})
+		assert.NoError(t, err)
+		assert.Equal(t, 1, attempts)
+	})
+
+	t.Run("respects the fixed interval between attempts", func(t *testing.T) {
+		start := time.Now()
+		attempts := 0
+		_ = RetryWithMaxTimesAndInterval(ctx, 2, 50*time.Millisecond, func() error {
+			attempts++
+			return fmt.Errorf("err")
+		})
+		// 2 retries × ~50ms 固定间隔,总耗时应 >= 80ms(留裕度)
+		assert.GreaterOrEqual(t, time.Since(start), 80*time.Millisecond)
+		assert.Equal(t, 3, attempts)
+	})
+}

backend/modules/evaluation/application/convertor/experiment/openapi.go

@@ -248,6 +248,40 @@ func IsSupportedOpenAPIEvalTargetType(t openapiEvalTarget.EvalTargetType) bool {
 	return err == nil
 }
 
+// ValidateOpenAPIEvalTargetClusterEnv validates the required cluster/env for the
+// long-connection custom_agent eval target. custom_agent is resolved to a live
+// RPC/frontier client at run time, so a missing cluster/env passes creation
+// silently and only fails later with an opaque RPC error; rejecting it up front
+// yields a clear param error with common values.
+//   - custom_agent: cluster and env are required, UNLESS an explicit AgentConnection
+//     (frontier direct-connect) is provided — direct-connect does not use cluster/env.
+//   - a2a_agent / custom_rpc_server: intentionally NOT validated for now.
+//
+// Returns nil when param/type is nil or the type is not custom_agent.
+func ValidateOpenAPIEvalTargetClusterEnv(param *openapi.SubmitExperimentEvalTargetParam) error {
+	if param == nil || param.EvalTargetType == nil {
+		return nil
+	}
+	// Only validate custom_agent (the in-house long-connection agent). a2a_agent /
+	// custom_rpc_server are intentionally left un-validated for now.
+	if *param.EvalTargetType == openapiEvalTarget.EvalTargetTypeCustomAgent {
+		// When an explicit AgentConnection (frontier direct-connect) is provided, the target
+		// is dispatched by the frontier tuple (ProductID/AppID/UserID/DeviceID) and cluster/env
+		// are not used at run time, so do not require them here — requiring them would wrongly
+		// reject the legitimate direct-connect path.
+		if param.IsSetAgentConnection() {
+			return nil
+		}
+		if param.GetCluster() == "" {
+			return fmt.Errorf("cluster is required for eval target type %s (e.g. \"default\")", *param.EvalTargetType)
+		}
+		if param.GetEnv() == "" {
+			return fmt.Errorf("env is required for eval target type %s (lane/env identifier, e.g. \"ppe_fornax_eval\")", *param.EvalTargetType)
+		}
+	}
+	return nil
+}
+
 func mapOpenAPIEvalTargetType(openapiType openapiEvalTarget.EvalTargetType) (domaindoEvalTarget.EvalTargetType, error) {
 	switch openapiType {
 	case openapiEvalTarget.EvalTargetTypeCozeBot:

backend/modules/evaluation/application/convertor/experiment/openapi_test.go

@@ -3830,3 +3830,59 @@ func TestDomainFilterLogicOpToOpenAPI(t *testing.T) {
 	assert.Equal(t, "or", domainFilterLogicOpToOpenAPI(domainExpt.FilterLogicOp_Or))
 	assert.Equal(t, "999", domainFilterLogicOpToOpenAPI(domainExpt.FilterLogicOp(999)))
 }
+
+func TestValidateOpenAPIEvalTargetClusterEnv(t *testing.T) {
+	tp := func(v openapiEvalTarget.EvalTargetType) *openapiEvalTarget.EvalTargetType { return &v }
+	tests := []struct {
+		name    string
+		param   *openapi.SubmitExperimentEvalTargetParam
+		wantErr string // 空=期望无错; 否则 err 应包含该子串
+	}{
+		{name: "nil param", param: nil},
+		{name: "nil type", param: &openapi.SubmitExperimentEvalTargetParam{}},
+		{
+			name:  "non long-connection type skipped (coze_loop_prompt)",
+			param: &openapi.SubmitExperimentEvalTargetParam{EvalTargetType: tp(openapiEvalTarget.EvalTargetTypeCozeLoopPrompt)},
+		},
+		{
+			name:    "custom_agent missing cluster",
+			param:   &openapi.SubmitExperimentEvalTargetParam{EvalTargetType: tp(openapiEvalTarget.EvalTargetTypeCustomAgent), Env: gptr.Of("cn")},
+			wantErr: "cluster is required",
+		},
+		{
+			name:    "custom_agent missing env",
+			param:   &openapi.SubmitExperimentEvalTargetParam{EvalTargetType: tp(openapiEvalTarget.EvalTargetTypeCustomAgent), Cluster: gptr.Of("default")},
+			wantErr: "env is required",
+		},
+		{
+			name:  "custom_agent both present",
+			param: &openapi.SubmitExperimentEvalTargetParam{EvalTargetType: tp(openapiEvalTarget.EvalTargetTypeCustomAgent), Cluster: gptr.Of("default"), Env: gptr.Of("cn")},
+		},
+		{
+			// direct-connect (AgentConnection) exempts cluster/env — must not require them.
+			name:  "custom_agent with AgentConnection exempts cluster/env",
+			param: &openapi.SubmitExperimentEvalTargetParam{EvalTargetType: tp(openapiEvalTarget.EvalTargetTypeCustomAgent), AgentConnection: &openapiEvalTarget.AgentConnection{}},
+		},
+		{
+			// a2a_agent is intentionally not validated for now → no error even without cluster/env.
+			name:  "a2a_agent not validated",
+			param: &openapi.SubmitExperimentEvalTargetParam{EvalTargetType: tp(openapiEvalTarget.EvalTargetTypeA2Agent)},
+		},
+		{
+			// custom_rpc_server is intentionally not validated for now → no error even without env.
+			name:  "custom_rpc_server not validated",
+			param: &openapi.SubmitExperimentEvalTargetParam{EvalTargetType: tp(openapiEvalTarget.EvalTargetTypeCustomRPCServer)},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := ValidateOpenAPIEvalTargetClusterEnv(tt.param)
+			if tt.wantErr == "" {
+				assert.NoError(t, err)
+			} else {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tt.wantErr)
+			}
+		})
+	}
+}

backend/modules/evaluation/application/eval_openapi_app.go

@@ -887,6 +887,7 @@ func (e *EvalOpenAPIApplication) ReportEvalTargetInvokeResult_(ctx context.Conte
 		Status:                  target.ToTargetRunStatsDO(req.GetStatus()),
 		Session:                 actx.Session,
 		EnableExtractTrajectory: actx.EnableExtractTrajectory,
+		AsyncUnixMS:             actx.AsyncUnixMS,
 	}); err != nil {
 		return nil, err
 	}
@@ -1064,6 +1065,13 @@ func (e *EvalOpenAPIApplication) SubmitExperimentOApi(ctx context.Context, req *
 		}
 	}
 
+	// Long-connection eval targets (custom_agent / a2a_agent / custom_rpc_server) require
+	// cluster/env to resolve a live client at run time. Validate up front so a missing value
+	// fails with a clear param error here instead of an opaque RPC error during experiment run.
+	if err := experiment_convertor.ValidateOpenAPIEvalTargetClusterEnv(req.EvalTargetParam); err != nil {
+		return nil, errorx.NewByCode(errno.CommonInvalidParamCode, errorx.WithExtraMsg(err.Error()))
+	}
+
 	createEvalTargetParam, err := experiment_convertor.OpenAPICreateEvalTargetParamDTO2Domain(req.EvalTargetParam)
 	if err != nil {
 		return nil, errorx.NewByCode(errno.CommonInvalidParamCode, errorx.WithExtraMsg(err.Error()))

backend/modules/evaluation/domain/entity/param.go

@@ -457,6 +457,10 @@ type ReportTargetRecordParam struct {
 
 	Session                 *Session
 	EnableExtractTrajectory *bool
+	// AsyncUnixMS 异步评测对象「请求发起」的时间(unix ms),即提交异步调用之前的时刻。
+	// 用作抽取 trajectory 的时间下界:record.BaseInfo.CreatedAt 是异步返回后才 stamp 的,偏晚会漏掉
+	// 请求发起到返回之间的 span。为 0 时回退到 record.BaseInfo.CreatedAt。
+	AsyncUnixMS int64
 }
 
 type DebugTargetParam struct {

backend/modules/evaluation/domain/service/expt_run_item_turn_impl.go

@@ -416,6 +416,10 @@ func (e *DefaultExptTurnEvaluationImpl) callEvaluators(ctx context.Context, exec
 	if err != nil {
 		return nil, err
 	}
+	// 立即绑定释放到 pool 生命周期:即使下方循环中途提前 return(如 conf 为 nil / buildEvaluatorInputData 失败)
+	// 跳过了 ExecAll,也能释放底层 ants pool 及其常驻协程(purge / ticktock),避免 goroutine 泄漏。
+	// Release 幂等,与 ExecAll 内部的释放不冲突。
+	defer pool.Release()
 
 	for idx := range expt.Evaluators {
 		ev := expt.Evaluators[idx]

backend/modules/evaluation/domain/service/expt_run_item_turn_impl_test.go

@@ -5,6 +5,7 @@ package service
 import (
 	"context"
 	"errors"
+	"runtime"
 	"strconv"
 	"sync"
 	"testing"
@@ -4292,3 +4293,92 @@ func TestDefaultExptTurnEvaluationImpl_callTarget_CustomAgentAndA2AAgent(t *test
 		})
 	}
 }
+
+// TestDefaultExptTurnEvaluationImpl_CallEvaluators_NoGoroutineLeak 在真实泄漏点 callEvaluators 直接验证:
+// 当 evaluator conf 缺失导致中途提前 return(跳过 pool.ExecAll)时,协程池仍被释放,不泄漏 goroutine。
+// 修复前(NewPool 后无 defer pool.Release()),每次调用会泄漏底层 ants pool 的 2 个常驻协程(purge/ticktock),
+// 此测试会因协程数持续增长而失败;修复后应保持平稳。
+func TestDefaultExptTurnEvaluationImpl_CallEvaluators_NoGoroutineLeak(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	// evalTargetService 在提前 return 前不会被调用到(OutputFields 无 omitted 大字段),这里可留空 mock。
+	// benefitService 在 callEvaluators 之前的权益校验会被调用,mock 成功放行。
+	mockBenefit := benefitmocks.NewMockIBenefitService(ctrl)
+	mockBenefit.EXPECT().CheckAndDeductEvalBenefit(gomock.Any(), gomock.Any()).
+		Return(&benefit.CheckAndDeductEvalBenefitResult{}, nil).AnyTimes()
+	service := &DefaultExptTurnEvaluationImpl{
+		evalTargetService: svcmocks.NewMockIEvalTargetService(ctrl),
+		benefitService:    mockBenefit,
+		metric:            metricsmocks.NewMockExptMetric(ctrl),
+	}
+
+	target := &entity.EvalTargetRecord{
+		EvalTargetOutputData: &entity.EvalTargetOutputData{
+			OutputFields: map[string]*entity.Content{
+				"field1": {Text: gptr.Of("v")},
+			},
+		},
+	}
+
+	// 构造一个会触发 "evaluator conf not found" 提前 return 的 etec:
+	// Expt.Evaluators 里有 versionID=1 的 evaluator,但 EvaluatorsConf.EvaluatorConf 为空 → GetEvaluatorConf(1)=nil。
+	newEtec := func() *entity.ExptTurnEvalCtx {
+		return &entity.ExptTurnEvalCtx{
+			ExptItemEvalCtx: &entity.ExptItemEvalCtx{
+				EvalSetItem: &entity.EvaluationSetItem{ID: 1, ItemID: 2},
+				Event: &entity.ExptItemEvalEvent{
+					Session: &entity.Session{UserID: "u"},
+					ExptID:  1,
+					SpaceID: 2,
+				},
+				Expt: &entity.Experiment{
+					ID:      1,
+					SpaceID: 2,
+					Evaluators: []*entity.Evaluator{
+						{
+							ID:                     1,
+							EvaluatorType:          entity.EvaluatorTypePrompt,
+							PromptEvaluatorVersion: &entity.PromptEvaluatorVersion{ID: 1},
+						},
+					},
+					EvalConf: &entity.EvaluationConfiguration{
+						ItemConcurNum: gptr.Of(1),
+						ConnectorConf: entity.Connector{
+							EvaluatorsConf: &entity.EvaluatorsConf{
+								EvaluatorConcurNum: gptr.Of(1),
+								// 故意不放 versionID=1 的 conf → 触发 :433 提前 return
+								EvaluatorConf: []*entity.EvaluatorConf{},
+							},
+						},
+					},
+				},
+			},
+			ExptTurnRunResult: &entity.ExptTurnRunResult{},
+			Turn:              &entity.Turn{FieldDataList: []*entity.FieldData{}},
+		}
+	}
+
+	// 先跑一次预热(确保触发的是提前 return 路径),并让运行时协程稳定。
+	if _, err := service.CallEvaluators(context.Background(), newEtec(), target); err == nil {
+		t.Fatalf("expected evaluator-conf-not-found error to hit the early-return path")
+	}
+	time.Sleep(200 * time.Millisecond)
+	runtime.GC()
+	before := runtime.NumGoroutine()
+
+	const N = 50
+	for i := 0; i < N; i++ {
+		_, err := service.CallEvaluators(context.Background(), newEtec(), target)
+		assert.Error(t, err) // 每次都走 conf-not-found 提前 return
+	}
+
+	// 等底层 ants pool 的常驻协程随 Release 退出。
+	time.Sleep(500 * time.Millisecond)
+	runtime.GC()
+	after := runtime.NumGoroutine()
+
+	// 修复前:N 次泄漏 ≈ 2N=100 个常驻协程;修复后应基本持平,留少量裕度。
+	assert.Less(t, after-before, 20,
+		"goroutine leak in callEvaluators early-return path: before=%d after=%d (delta=%d)", before, after, after-before)
+}

backend/modules/evaluation/domain/service/target_impl.go

@@ -45,6 +45,10 @@ type EvalTargetServiceImpl struct {
 
 const evalTargetRecordPersistTimeout = 5 * time.Second
 
+// trajectoryStartTimeBufferMS 抽取 trajectory 时，时间下界额外向前预留的 buffer(1 分钟)，
+// 用于吸收请求发起时间与实际 span 上报时间之间可能的时钟/延迟误差，避免漏掉最早的 span。
+const trajectoryStartTimeBufferMS = int64(60 * 1000)
+
 func NewEvalTargetServiceImpl(evalTargetRepo repo.IEvalTargetRepo,
 	idgen idgen.IIDGenerator,
 	metric metrics.EvalTargetMetrics,
@@ -445,6 +449,11 @@ func (e *EvalTargetServiceImpl) ExtractTrajectory(ctx context.Context, spaceID i
 	if len(traceID) == 0 {
 		return nil, errorx.New("ExtractTrajectory with null traceID")
 	}
+	// 时间下界默认向前多减 1 分钟 buffer:防御请求发起时间与实际 span 时间之间可能存在的时钟/上报误差,
+	// 避免因下界略偏晚而漏掉最早的 span 导致轨迹拉不全。trace 查询按 traceID 精确匹配,放宽下界不会引入无关数据。
+	if startTimeMS != nil {
+		startTimeMS = gptr.Of(*startTimeMS - trajectoryStartTimeBufferMS)
+	}
 	trajectories, err := e.trajectoryAdapter.ListTrajectory(ctx, spaceID, []string{traceID}, startTimeMS)
 	if err != nil {
 		return nil, err
@@ -725,7 +734,11 @@ func (e *EvalTargetServiceImpl) ReportInvokeRecords(ctx context.Context, param *
 
 	recordTrajectory := func() error {
 		var sms *int64
-		if record.BaseInfo != nil {
+		// 优先用「请求发起时间」作为抽取 trajectory 的时间下界;它比 record.BaseInfo.CreatedAt(异步返回后才 stamp)
+		// 更早,避免漏掉请求发起到返回之间的 span。为 0(未透传)时回退到 CreatedAt,保持向前兼容。
+		if param.AsyncUnixMS > 0 {
+			sms = gptr.Of(param.AsyncUnixMS)
+		} else if record.BaseInfo != nil {
 			sms = record.BaseInfo.CreatedAt
 		}
 		trajectory, err := e.ExtractTrajectory(ctx, param.SpaceID, record.TraceID, sms)