[fix][evaluation] await sandbox cleanup before experiment deletion

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

@@ -0,0 +1,217 @@
+package service
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/bytedance/gg/gptr"
+	"github.com/coze-dev/coze-loop/backend/modules/evaluation/consts"
+	"github.com/coze-dev/coze-loop/backend/modules/evaluation/domain/component/rpc"
+	"github.com/coze-dev/coze-loop/backend/modules/evaluation/domain/entity"
+	"github.com/coze-dev/coze-loop/backend/modules/evaluation/domain/repo"
+	"github.com/stretchr/testify/require"
+)
+
+type deleteCoreTrace struct {
+	sync.Mutex
+	events []string
+}
+
+func (t *deleteCoreTrace) add(s string) { t.Lock(); defer t.Unlock(); t.events = append(t.events, s) }
+
+type deleteCoreExpts struct {
+	repo.IExperimentRepo
+	row     *entity.Experiment
+	trace   *deleteCoreTrace
+	deleted int
+}
+
+func (r *deleteCoreExpts) GetByID(context.Context, int64, int64) (*entity.Experiment, error) {
+	return r.row, nil
+}
+func (r *deleteCoreExpts) MGetByID(context.Context, []int64, int64) ([]*entity.Experiment, error) {
+	return []*entity.Experiment{r.row}, nil
+}
+func (r *deleteCoreExpts) Delete(context.Context, int64, int64) error {
+	r.deleted++
+	r.trace.add("delete")
+	return nil
+}
+func (r *deleteCoreExpts) MDelete(context.Context, []int64, int64) error {
+	r.deleted++
+	r.trace.add("delete")
+	return nil
+}
+
+type deleteCoreSelection struct {
+	ExptResultService
+	calls int
+	err   error
+}
+
+func (r *deleteCoreSelection) GetIncompleteTurns(context.Context, int64, int64, *entity.Session) ([]*entity.ItemTurnID, error) {
+	r.calls++
+	return []*entity.ItemTurnID{{ItemID: 1, TurnID: 1}}, r.err
+}
+
+type deleteCoreTurns struct {
+	repo.IExptTurnResultRepo
+	err error
+}
+
+func (r *deleteCoreTurns) MGetItemTurnRunLogs(context.Context, int64, int64, []int64, int64) ([]*entity.ExptTurnResultRunLog, error) {
+	return []*entity.ExptTurnResultRunLog{{TargetResultID: 10}}, r.err
+}
+
+type deleteCoreRuns struct{ repo.IExptRunLogRepo }
+
+func (*deleteCoreRuns) Get(context.Context, int64, int64) (*entity.ExptRunLog, error) {
+	return &entity.ExptRunLog{ExptID: 701}, nil
+}
+
+type deleteCoreRecords struct {
+	repo.IEvalTargetRepo
+	record *entity.EvalTargetRecord
+	trace  *deleteCoreTrace
+	saves  int
+}
+
+func (r *deleteCoreRecords) ListEvalTargetRecordByIDsAndSpaceID(context.Context, int64, []int64) ([]*entity.EvalTargetRecord, error) {
+	copy := *r.record
+	return []*entity.EvalTargetRecord{&copy}, nil
+}
+func (*deleteCoreRecords) BatchGetEvalTargetVersion(context.Context, int64, []int64) ([]*entity.EvalTarget, error) {
+	return []*entity.EvalTarget{{EvalTargetType: entity.EvalTargetTypeSandboxAgent, EvalTargetVersion: &entity.EvalTargetVersion{ID: 20}}}, nil
+}
+func (r *deleteCoreRecords) SaveEvalTargetRecord(_ context.Context, record *entity.EvalTargetRecord, _ *bool) error {
+	copy := *record
+	r.record = &copy
+	r.saves++
+	r.trace.add("save")
+	return nil
+}
+
+type deleteCoreScheduler struct {
+	rpc.ISandboxSchedulerAdapter
+	trace                  *deleteCoreTrace
+	calls                  int
+	requests               []rpc.SandboxDestroyRequest
+	failID                 string
+	failFirst              bool
+	entered, proceed, done chan struct{}
+}
+
+func (s *deleteCoreScheduler) Destroy(_ context.Context, r *rpc.SandboxDestroyRequest) (*rpc.SandboxDestroyResponse, error) {
+	s.calls++
+	s.requests = append(s.requests, *r)
+	s.trace.add("cancel:" + r.ExecuteIDs[0])
+	if s.entered != nil {
+		close(s.entered)
+		<-s.proceed
+		close(s.done)
+	}
+	if (s.failFirst && s.calls == 1) || s.failID == r.ExecuteIDs[0] {
+		return nil, errors.New("destroy failed")
+	}
+	return &rpc.SandboxDestroyResponse{AffectedCount: 1}, nil
+}
+func deleteCoreFixture() (*ExptMangerImpl, *deleteCoreExpts, *deleteCoreSelection, *deleteCoreTurns, *deleteCoreRecords, *deleteCoreScheduler) {
+	trace := &deleteCoreTrace{}
+	expts := &deleteCoreExpts{row: &entity.Experiment{ID: 701, SpaceID: 11, TargetSpaceID: 22, LatestRunID: 801, TargetType: entity.EvalTargetTypeSandboxAgent}, trace: trace}
+	selection := &deleteCoreSelection{}
+	turns := &deleteCoreTurns{}
+	records := &deleteCoreRecords{trace: trace, record: &entity.EvalTargetRecord{ID: 10, SpaceID: 22, TargetVersionID: 20, ExperimentRunID: 801, Status: gptr.Of(entity.EvalTargetRunStatusAsyncInvoking), EvalTargetOutputData: &entity.EvalTargetOutputData{Ext: map[string]string{consts.OutputDataExtKeySandboxExecuteIDs: `["10-agent","10-orch"]`, "retained": "value"}}}}
+	scheduler := &deleteCoreScheduler{trace: trace}
+	target := &EvalTargetServiceImpl{evalTargetRepo: records, exptRunLogRepo: &deleteCoreRuns{}, sandboxSchedulerAdapter: scheduler, sandboxDestroyRetryBudget: time.Nanosecond}
+	return &ExptMangerImpl{exptRepo: expts, exptResultService: selection, turnResultRepo: turns, evalTargetService: target}, expts, selection, turns, records, scheduler
+}
+func TestDeleteSandboxCoreCallsExistingCleanupBeforeSoftDelete(t *testing.T) {
+	for _, batch := range []bool{false, true} {
+		t.Run(map[bool]string{false: "single", true: "batch"}[batch], func(t *testing.T) {
+			mgr, expts, _, _, records, scheduler := deleteCoreFixture()
+			var err error
+			if batch {
+				err = mgr.MDelete(context.Background(), []int64{701}, 11, &entity.Session{})
+			} else {
+				err = mgr.Delete(context.Background(), 701, 11, &entity.Session{})
+			}
+			require.NoError(t, err)
+			require.Equal(t, 1, expts.deleted)
+			require.Equal(t, 2, scheduler.calls)
+			for _, request := range scheduler.requests {
+				require.EqualValues(t, 22, request.WorkspaceID)
+				require.Equal(t, "701", request.TaskID)
+			}
+			require.Equal(t, []string{"cancel:10-agent", "cancel:10-orch", "save", "delete"}, expts.trace.events)
+			require.Equal(t, entity.EvalTargetRunStatusFail, *records.record.Status)
+		})
+	}
+}
+func TestDeleteSandboxCoreFailureRetainsRecordsAndRetriesOriginalEntry(t *testing.T) {
+	mgr, expts, _, _, records, scheduler := deleteCoreFixture()
+	scheduler.failID = "10-orch"
+	original := records.record.EvalTargetOutputData.Ext
+	require.Error(t, mgr.Delete(context.Background(), 701, 11, &entity.Session{}))
+	require.Zero(t, expts.deleted)
+	require.Zero(t, records.saves)
+	require.Equal(t, entity.EvalTargetRunStatusAsyncInvoking, *records.record.Status)
+	require.Equal(t, original, records.record.EvalTargetOutputData.Ext)
+	scheduler.failID = ""
+	require.NoError(t, mgr.Delete(context.Background(), 701, 11, &entity.Session{}))
+	require.Equal(t, 1, expts.deleted)
+}
+func TestDeleteSandboxCoreSelectionFailureDoesNotDelete(t *testing.T) {
+	for _, turnQuery := range []bool{false, true} {
+		t.Run(map[bool]string{false: "selection", true: "turns"}[turnQuery], func(t *testing.T) {
+			mgr, expts, selection, turns, _, scheduler := deleteCoreFixture()
+			if turnQuery {
+				turns.err = errors.New("turn lookup failed")
+			} else {
+				selection.err = errors.New("selection failed")
+			}
+			require.Error(t, mgr.MDelete(context.Background(), []int64{701}, 11, &entity.Session{}))
+			require.Zero(t, expts.deleted)
+			require.Zero(t, scheduler.calls)
+		})
+	}
+}
+func TestDeleteSandboxCoreNonSandboxDoesNotLoadCleanup(t *testing.T) {
+	mgr, expts, selection, _, _, scheduler := deleteCoreFixture()
+	expts.row.TargetType = entity.EvalTargetTypeCustomRPCServer
+	selection.err = errors.New("must not query")
+	require.NoError(t, mgr.Delete(context.Background(), 701, 11, &entity.Session{}))
+	require.Zero(t, selection.calls)
+	require.Zero(t, scheduler.calls)
+}
+func TestBaselineDestroyRetryAcceptsReturnedErrors(t *testing.T) {
+	_, _, _, _, records, scheduler := deleteCoreFixture()
+	scheduler.failFirst = true
+	target := &EvalTargetServiceImpl{evalTargetRepo: records, sandboxSchedulerAdapter: scheduler, sandboxDestroyRetryBudget: 500 * time.Millisecond}
+	target.destroySandboxExecuteSync(context.Background(), "701", 22, []string{"10-agent"}, false)
+	require.Equal(t, 2, scheduler.calls)
+}
+func TestDeleteSandboxCoreDefaultTerminationRemainsAsync(t *testing.T) {
+	mgr, _, _, _, records, scheduler := deleteCoreFixture()
+	records.record.EvalTargetOutputData.Ext[consts.OutputDataExtKeySandboxExecuteIDs] = `["10-agent"]`
+	scheduler.entered = make(chan struct{})
+	scheduler.proceed = make(chan struct{})
+	scheduler.done = make(chan struct{})
+	returned := make(chan struct{})
+	go func() {
+		mgr.evalTargetService.TerminateAsyncRecordsAndDestroySandbox(context.Background(), 22, []int64{10}, 1, "cancel", false)
+		close(returned)
+	}()
+	select {
+	case <-returned:
+	case <-time.After(time.Second):
+		close(scheduler.proceed)
+		t.Fatal("default cancellation became synchronous")
+	}
+	<-scheduler.entered
+	require.Equal(t, entity.EvalTargetRunStatusFail, *records.record.Status)
+	close(scheduler.proceed)
+	<-scheduler.done
+}

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

@@ -1380,14 +1380,17 @@ func (e *ExptMangerImpl) listItemTurnIDs(ctx context.Context, spaceID, exptID in
 //
 // targetSourceSpaceID: 跨空间共享场景下评测对象的来源空间 (Expt.TargetSpaceID, 0=同消费方空间);
 // EvalTargetRecord 落库时 SpaceID = 来源空间, 用消费方 spaceID 查会漏, 沙箱清不掉。
-func (e *ExptMangerImpl) terminateSandboxExecutesForCancelledItems(ctx context.Context, spaceID, targetSourceSpaceID, exptID int64, exptRunID *int64, terminatedItemIDs []int64) {
+func (e *ExptMangerImpl) terminateSandboxExecutesForCancelledItems(ctx context.Context, spaceID, targetSourceSpaceID, exptID int64, exptRunID *int64, terminatedItemIDs []int64, waitForDestroy ...bool) error {
 	if e.evalTargetService == nil || exptRunID == nil || *exptRunID <= 0 || len(terminatedItemIDs) == 0 {
-		return
+		if len(terminatedItemIDs) > 0 && len(waitForDestroy) > 0 && waitForDestroy[0] {
+			return fmt.Errorf("sandbox cleanup unavailable")
+		}
+		return nil
 	}
 	turnRunLogs, err := e.turnResultRepo.MGetItemTurnRunLogs(ctx, exptID, *exptRunID, terminatedItemIDs, spaceID)
 	if err != nil {
 		logs.CtxWarn(ctx, "[SandboxDestroy] MGetItemTurnRunLogs fail on cancel, expt_id=%d, expt_run_id=%d, err=%v", exptID, *exptRunID, err)
-		return
+		return err
 	}
 	recordIDSet := make(map[int64]struct{})
 	for _, rl := range turnRunLogs {
@@ -1397,7 +1400,7 @@ func (e *ExptMangerImpl) terminateSandboxExecutesForCancelledItems(ctx context.C
 		recordIDSet[rl.TargetResultID] = struct{}{}
 	}
 	if len(recordIDSet) == 0 {
-		return
+		return nil
 	}
 	recordIDs := make([]int64, 0, len(recordIDSet))
 	for id := range recordIDSet {
@@ -1406,13 +1409,14 @@ func (e *ExptMangerImpl) terminateSandboxExecutesForCancelledItems(ctx context.C
 	// 跨空间: EvalTargetRecord.SpaceID = 来源空间, TerminateAsyncRecordsAndDestroySandbox 的
 	// List/Version/Destroy 全链路都用它，否则 space 过滤把 record 过滤空、Destroy RPC 发不出。
 	targetSpaceID := resolveLoadSpaceID(spaceID, targetSourceSpaceID)
-	e.evalTargetService.TerminateAsyncRecordsAndDestroySandbox(
+	return e.evalTargetService.TerminateAsyncRecordsAndDestroySandbox(
 		ctx,
 		targetSpaceID,
 		recordIDs,
 		int32(errno.AsyncEvalTargetTerminatedCode),
 		"async eval target terminated: experiment cancelled",
 		false,
+		waitForDestroy...,
 	)
 }
 

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

@@ -461,6 +461,9 @@ func (e *ExptMangerImpl) MDelete(ctx context.Context, exptIDs []int64, spaceID i
 		if expt == nil {
 			continue
 		}
+		if err := e.cancelSandboxBeforeDelete(ctx, expt, session); err != nil {
+			return err
+		}
 		e.releaseCentralQuotaForIncompleteItems(ctx, expt, nil, entity.ExptStatus_Terminated)
 	}
 
@@ -484,6 +487,27 @@ func (e *ExptMangerImpl) MDelete(ctx context.Context, exptIDs []int64, spaceID i
 	return nil
 }
 
+// Resolve existing cleanup references while deletion has not hidden the experiment.
+func (e *ExptMangerImpl) cancelSandboxBeforeDelete(ctx context.Context, expt *entity.Experiment, session *entity.Session) error {
+	if expt == nil || expt.LatestRunID <= 0 || (expt.TargetType != entity.EvalTargetTypeSandboxAgent && !expt.IsSandboxAgentTarget()) {
+		return nil
+	}
+	turns, err := e.exptResultService.GetIncompleteTurns(ctx, expt.ID, expt.SpaceID, session)
+	if err != nil {
+		return err
+	}
+	seen := make(map[int64]bool, len(turns))
+	items := make([]int64, 0, len(turns))
+	for _, turn := range turns {
+		if turn == nil || seen[turn.ItemID] {
+			continue
+		}
+		seen[turn.ItemID] = true
+		items = append(items, turn.ItemID)
+	}
+	return e.terminateSandboxExecutesForCancelledItems(ctx, expt.SpaceID, expt.TargetSpaceID, expt.ID, &expt.LatestRunID, items, true)
+}
+
 func (e *ExptMangerImpl) makeExptMutexLockKey(exptID int64) string {
 	return fmt.Sprintf("expt_run_mutex_lock:%d", exptID)
 }
@@ -1842,6 +1866,9 @@ func (e *ExptMangerImpl) Delete(ctx context.Context, exptID, spaceID int64, sess
 	// ★ 与 MDelete 同理：软删前必须归还中心调度额度，且必须在删除之前。
 	// 完整论证见 MDelete 里那段注释（三条独立原因 + 为什么顺序不能反）。
 	// 两个删除入口都要改 —— 只改一个等于留一半泄漏。
+	if err := e.cancelSandboxBeforeDelete(ctx, expt, session); err != nil {
+		return err
+	}
 	e.releaseCentralQuotaForIncompleteItems(ctx, expt, nil, entity.ExptStatus_Terminated)
 
 	// 删除实验

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

@@ -37,7 +37,8 @@ type IEvalTargetService interface {
 	// 并以 best-effort 方式触发沙箱 Execute 销毁。非 SandboxAgent / 非 AsyncInvoking 的 record 会被忽略。
 	// errCode 用于写入 EvalTargetRunError，区分 zombie timeout / 手动取消等场景。
 	// zombieTimeout=true 时，Destroy 会带上 SandboxAgent 收尾命令 EndCmd（expt_id/invoke_id 由内部拼接）。
-	TerminateAsyncRecordsAndDestroySandbox(ctx context.Context, spaceID int64, recordIDs []int64, errCode int32, errMessage string, zombieTimeout bool)
+	// waitForDestroy 仅供删除前等待现有重试结果；默认仍在后台销毁。
+	TerminateAsyncRecordsAndDestroySandbox(ctx context.Context, spaceID int64, recordIDs []int64, errCode int32, errMessage string, zombieTimeout bool, waitForDestroy ...bool) error
 	// CheckSandboxTerminated 查询 SandboxAgent 类型 record 关联的沙箱 execute 是否已提前进入终态（Failed/Canceled）。
 	// 沙箱状态查询失败（含开源 stub 的 "not implement"）会在内部 warn 并跳过该 record，不向外抛错。
 	// 仅当 record 仍处于 AsyncInvoking 且沙箱返回 Failed/Canceled 时命中；Succeeded/Running/Pending 不视为终态，