fix(lastleg): persist bounded test collection session budgets

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRequestContext.java

@@ -5,6 +5,10 @@ public record SelfopRequestContext(String sourceAccount,String executionPurpose,
     String sessionId,String exclusiveGrantId,Long taskId,Long runId) {
     public static SelfopRequestContext forTask(SelfopSyncTask task,SelfopSyncRun run,SelfopAnalyticsProperties properties) {
         Long root="VALIDATION".equals(task.getExecutionPurpose())?task.getValidationRootTaskId():task.getId();
+        if("BUSINESS".equals(task.getExecutionPurpose())&&"test".equals(properties.getEnvironment())) {
+            root=task.getBusinessRootTaskId();
+            SelfopSupport.require(root!=null,"BUSINESS_SESSION_ROOT_REQUIRED");
+        }
         return new SelfopRequestContext(task.getSourceAccount(),task.getExecutionPurpose(),root,
             "selfop-"+root,properties.getExclusiveGrantId(),task.getId(),run.getId());
     }

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopSyncTask.java

@@ -3,6 +3,8 @@ package com.baiyue.lastleg.selfoperated.analytics.model;
 import com.baomidou.mybatisplus.annotation.IdType;
 import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.FieldStrategy;
 import lombok.Data;
 import java.math.BigDecimal;
 import java.time.LocalDate;
@@ -15,6 +17,11 @@ public class SelfopSyncTask {
     private String executionPurpose;
     private Long validationRootTaskId;
     private String validationBudgetJson;
+    @TableField(updateStrategy = FieldStrategy.NEVER)
+    private Long businessRootTaskId;
+    // Only locked explicit budget SQL may update this counter; collector entities can be stale.
+    @TableField(updateStrategy = FieldStrategy.NEVER)
+    private String businessBudgetJson;
     private Long resultPublishedRunId;
     private String progressJson;
     private LocalDateTime progressUpdatedAtUtc;

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java

@@ -235,6 +235,9 @@ public class CollectionQueryService {
         String disabled=executionBlock(task.warehouseCode(),protection);if(disabled!=null)return disabled;
         if(!"BUSINESS".equals(config.getCollectionMode()))return "COLLECTION_MODE_NOT_BUSINESS";
         if("test".equals(config.getEnvironment())&&"SCHEDULED".equals(task.taskType()))return "TEST_MANUAL_TASK_REQUIRED";
+        if("test".equals(config.getEnvironment())) {
+            String budgetBlock=businessBudgetBlock(task);if(budgetBlock!=null)return budgetBlock;
+        }
         if(!Set.of("DEAD","RETRY_WAIT").contains(task.status()))return "TASK_NOT_RETRYABLE";
         if(scope==null||!Objects.equals(scope.scopeHash(),task.scopeHash()))return "SCOPE_CHANGED";
         if(scope.leaseOwner()!=null&&scope.leaseUntilUtc()!=null&&clock.instant().isBefore(utc(scope.leaseUntilUtc())))return "SCOPE_LEASE_BUSY";
@@ -242,6 +245,19 @@ public class CollectionQueryService {
         if(task.upstreamTaskId()!=null&&(parent==null||!"SUCCEEDED".equals(parent.status())))return "UPSTREAM_NOT_READY";
         return null;
     }
+    private String businessBudgetBlock(Task task) {
+        JsonNode budget=boundedObject(task.businessRootBudgetJson());
+        Long maximum=number(budget,"maxHttp"),reserved=number(budget,"http"),confirmed=number(budget,"confirmedHttp");
+        Instant started=budgetInstant(budget,"startedAtUtc"),deadline=budgetInstant(budget,"deadlineAtUtc");
+        if(budget==null||maximum==null||maximum<=0||maximum>Integer.MAX_VALUE||reserved==null||reserved>maximum
+                ||confirmed==null||confirmed>reserved||started==null||deadline==null||!deadline.isAfter(started))
+            return "BUSINESS_SESSION_BUDGET_MISSING";
+        JsonNode stopped=budget.get("stopReason");
+        if(stopped!=null&&!stopped.isNull())
+            return "BUSINESS_SESSION_STOPPED";
+        if(!clock.instant().isBefore(deadline)&&!Boolean.TRUE.equals(task.businessLocalWorkReady()))return "BUSINESS_SESSION_DEADLINE";
+        return null;
+    }
     private CollectionResponses.Attempt attemptView(Attempt run,Task task,boolean retryAllowed) {
         boolean terminal=TERMINAL.contains(run.status());
         boolean same=task!=null&&Objects.equals(run.attemptNo(),task.attemptCount());
@@ -251,11 +267,13 @@ public class CollectionQueryService {
         String health=terminal?"TERMINATED":heartbeat==null||!clock.instant().isBefore(heartbeat.plusSeconds(config.getLeaseSeconds()))?"UNCONFIRMED":"HEALTHY";
         Long elapsed=started==null?null:Math.max(0,Duration.between(started,finished==null?clock.instant():finished).getSeconds());
         return new CollectionResponses.Attempt(run.id(),run.taskNo(),run.attemptNo(),run.status(),run.lastPhase(),run.executionPurpose(),
-                started,finished,progressInstant(progress,"stageStartedAt"),updated,heartbeat,health,elapsed,run.errorCode(),safeMessage(run.errorMessage()),
+                started,finished,progressInstant(progress,"stageStartedAt"),updated,heartbeat,health,elapsed,run.errorCode(),safeMessage(run.errorCode(),run.errorMessage()),
                 retryAllowed&&same&&"BUSINESS".equals(run.executionPurpose())&&Set.of("FAILED","ABORTED","YIELDED").contains(run.status()),
                 run.coverageStatus(),utc(run.coveredFromUtc()),utc(run.coveredUntilUtc()),utc(run.publishedAtUtc()),progress);
     }
-    private String safeMessage(String error) {
+    private String safeMessage(String code,String error) {
+        if("BUSINESS_SESSION_HTTP_LIMIT".equals(code))return "本次正式采集会话的请求预算已耗尽，根任务及关联子任务已停止，不能恢复原任务";
+        if("BUSINESS_SESSION_DEADLINE".equals(code))return "本次正式采集会话已到截止时间，根任务及关联子任务已停止，不能恢复原任务";
         if(error==null)return null;
         // Errors are already sanitized by the write side; avoid exposing arbitrary payloads here.
         return "采集尝试失败，请根据错误码查看受控诊断记录";

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java

@@ -41,17 +41,30 @@ public interface CollectionReadMapper {
             + " THEN JSON_UNQUOTE(JSON_EXTRACT(t.query_scope_json,'$.warehouses[0]')) ELSE t.warehouse_code END";
     String TASK_FAMILY = "CASE WHEN " + VALIDATION_ROOT + " THEN 'OUTBOUND' ELSE s.data_family END";
     String TASK_MEMBERSHIP = "((t.task_role='WAREHOUSE_SYNC' AND s.id IS NOT NULL) OR " + VALIDATION_ROOT + ")";
+    String BUSINESS_LOCAL_WORK_READY = "CASE WHEN t.task_role='WAREHOUSE_SYNC' AND s.data_family='OUTBOUND'"
+            + " AND (t.task_type='REBUILD' OR t.apply_state='APPLIED' OR ("
+            + "JSON_EXTRACT(t.routing_manifest_json,'$.directoryComplete')=CAST('true' AS JSON)"
+            + " AND JSON_TYPE(JSON_EXTRACT(t.routing_manifest_json,'$.objects'))='OBJECT'"
+            + " AND NOT EXISTS (SELECT 1 FROM JSON_TABLE(t.routing_manifest_json,'$.objects.*'"
+            + " COLUMNS(entry JSON PATH '$')) manifest_item"
+            + " WHERE COALESCE(JSON_TYPE(JSON_EXTRACT(manifest_item.entry,'$.effectiveStageId')),'NULL') NOT IN ('INTEGER')"
+            + " OR (CAST(JSON_UNQUOTE(JSON_EXTRACT(manifest_item.entry,'$.effectiveStageId')) AS DECIMAL(65,0)) BETWEEN 1 AND 9223372036854775807) IS NOT TRUE)))"
+            + " THEN TRUE ELSE FALSE END";
     String TASK = "t.id,t.task_no,t.source_account," + TASK_WAREHOUSE + " AS warehouse_code," + TASK_FAMILY + " AS data_family,t.task_role,t.task_type,"
             + "t.execution_purpose,t.scope_hash,t.status,t.apply_state,t.attempt_count,t.query_from_utc,t.query_until_utc,"
             + "t.next_retry_at_utc,t.upstream_task_id,t.validation_root_task_id,t.result_published_run_id,"
-            + "t.progress_json,t.progress_updated_at_utc,t.validation_budget_json,t.created_at,t.updated_at";
+            + "t.progress_json,t.progress_updated_at_utc,t.validation_budget_json,b.business_budget_json AS business_root_budget_json,"
+            + BUSINESS_LOCAL_WORK_READY + " AS business_local_work_ready,t.created_at,t.updated_at";
     String ATTEMPT = "r.id,r.task_id,t.task_no,r.attempt_no,r.status,r.last_phase,r.execution_purpose,"
             + "r.fetch_started_at_utc,r.fetch_finished_at_utc,r.build_started_at_utc,r.finished_at_utc,"
             + "r.published_at_utc,r.heartbeat_at_utc,r.progress_snapshot_json,r.coverage_status,"
             + "r.covered_from_utc,r.covered_until_utc,r.error_code,r.error_message,r.created_at,"
             + "CASE WHEN " + VALIDATION_ROOT + " THEN " + TASK_WAREHOUSE + " ELSE r.warehouse_code END AS warehouse_code,r.scope_id,r.scope_hash";
+    String BUSINESS_ROOT_JOIN = " LEFT JOIN ll_selfop_sync_task b ON b.id=t.business_root_task_id"
+            + " AND b.source_account=t.source_account AND t.execution_purpose='BUSINESS'"
+            + " AND b.execution_purpose='BUSINESS' AND b.business_root_task_id=b.id ";
     String TASK_FROM = " FROM ll_selfop_sync_task t LEFT JOIN ll_selfop_sync_scope s ON s.id=t.scope_id "
-            + "AND s.source_account=t.source_account AND s.warehouse_code=t.warehouse_code ";
+            + "AND s.source_account=t.source_account AND s.warehouse_code=t.warehouse_code " + BUSINESS_ROOT_JOIN;
     String RUN_FROM = " FROM ll_selfop_sync_run r JOIN ll_selfop_sync_task t ON t.id=r.task_id "
             + "AND t.source_account=r.source_account AND r.execution_purpose=t.execution_purpose AND r.run_role=t.task_role"
             + " AND (t.warehouse_code=r.warehouse_code OR (t.warehouse_code IS NULL AND r.warehouse_code IS NULL)) "
@@ -96,8 +109,7 @@ public interface CollectionReadMapper {
     @Select("SELECT api_cooldown_until_utc,api_pause_reason,api_budget_json FROM ll_selfop_outbound_discovery_scope WHERE source_account=#{account} LIMIT 1")
     Protection protection(@Param("account") String account);
 
-    @Select("<script>SELECT " + TASK + " FROM ll_selfop_sync_task t LEFT JOIN ll_selfop_sync_scope s ON s.id=t.scope_id"
-            + " AND s.source_account=t.source_account AND s.warehouse_code=t.warehouse_code"
+    @Select("<script>SELECT " + TASK + TASK_FROM
             + " WHERE t.source_account=#{account} AND t.id IN <foreach collection='ids' item='id' open='(' close=')' separator=','>#{id}</foreach></script>")
     List<Task> tasksByIds(@Param("account") String account,@Param("ids") List<Long> ids);
     @Select("<script>SELECT " + ATTEMPT + " FROM ll_selfop_sync_run r JOIN ll_selfop_sync_task t ON t.id=r.task_id"
@@ -132,6 +144,7 @@ public interface CollectionReadMapper {
     // The one account active task is separately related to warehouses by actual persisted windows.
     @Select("SELECT " + TASK + " FROM ll_selfop_outbound_discovery_scope d JOIN ll_selfop_sync_task t ON t.id=d.active_task_id"
             + " AND t.source_account=d.source_account LEFT JOIN ll_selfop_sync_scope s ON s.id=t.scope_id"
+            + BUSINESS_ROOT_JOIN
             + " WHERE d.source_account=#{account} AND t.task_role='ACCOUNT_DISCOVERY' LIMIT 1")
     Task sharedActiveAccount(@Param("account") String account);
     @Select("<script>SELECT DISTINCT s.warehouse_code FROM ll_selfop_sync_scope s WHERE s.source_account=#{account}"

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadRows.java

@@ -20,7 +20,8 @@ public final class CollectionReadRows {
             String scopeHash, String status, String applyState, Integer attemptCount,
             LocalDateTime queryFromUtc, LocalDateTime queryUntilUtc, LocalDateTime nextRetryAtUtc,
             Long upstreamTaskId, Long validationRootTaskId, Long resultPublishedRunId,
-            String progressJson, LocalDateTime progressUpdatedAtUtc, String validationBudgetJson,
+            String progressJson, LocalDateTime progressUpdatedAtUtc, String validationBudgetJson, String businessRootBudgetJson,
+            Boolean businessLocalWorkReady,
             LocalDateTime createdAt, LocalDateTime updatedAt) {}
     public record Attempt(Long id, Long taskId, String taskNo, Integer attemptNo,
             String status, String lastPhase, String executionPurpose,

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopBusinessBudget.java

@@ -0,0 +1,36 @@
+package com.baiyue.lastleg.selfoperated.analytics.request;
+
+import com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport;
+import java.time.LocalDateTime;
+
+/** TEST manual BUSINESS session ceiling, frozen on its root and shared by descendants. */
+public final class SelfopBusinessBudget {
+    public int maxHttp, http, confirmedHttp;
+    public LocalDateTime startedAtUtc, deadlineAtUtc;
+    public String stopReason;
+
+    public static String freeze(LocalDateTime now, int maxHttp, int seconds) {
+        if (now == null || maxHttp < 1 || seconds < 1)
+            throw new IllegalArgumentException("BUSINESS_SESSION_BUDGET_INVALID");
+        var budget = new SelfopBusinessBudget();
+        budget.maxHttp = maxHttp;
+        budget.startedAtUtc = now;
+        budget.deadlineAtUtc = now.plusSeconds(seconds);
+        return SelfopSupport.json(budget);
+    }
+
+    public void validate() {
+        if (maxHttp < 1 || http < 0 || http > maxHttp || confirmedHttp < 0 || confirmedHttp > http
+            || startedAtUtc == null || deadlineAtUtc == null || !deadlineAtUtc.isAfter(startedAtUtc)
+            || stopReason != null && !stopReason.matches("[A-Z0-9_]{1,64}"))
+            throw new EgressRejectedException("BUSINESS_SESSION_BUDGET_INVALID", false, null);
+    }
+
+    public String limit(LocalDateTime now) {
+        validate();
+        if (stopReason != null) return stopReason;
+        if (!now.isBefore(deadlineAtUtc)) return "BUSINESS_SESSION_DEADLINE";
+        if (http >= maxHttp) return "BUSINESS_SESSION_HTTP_LIMIT";
+        return null;
+    }
+}

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressMapper.java

@@ -9,6 +9,8 @@ public interface SelfopEgressMapper {
  @Select("SELECT * FROM ll_selfop_sync_run WHERE id=#{id} FOR UPDATE") SelfopSyncRun lockRun(Long id);
  @Select("SELECT COUNT(*) FROM ll_selfop_sync_scope WHERE id=#{scopeId} AND fence_token=#{fence} AND lease_until_utc>#{now}") int warehouseFence(@Param("scopeId") Long scopeId,@Param("fence") Long fence,@Param("taskId") Long taskId,@Param("now") LocalDateTime now);
  @Update("UPDATE ll_selfop_outbound_discovery_scope SET api_budget_json=#{apiBudgetJson},api_cooldown_until_utc=#{apiCooldownUntilUtc},api_pause_reason=#{apiPauseReason},updated_at=UTC_TIMESTAMP(3) WHERE id=#{id}") int saveAccount(SelfopOutboundDiscoveryScope row);
- @Update("UPDATE ll_selfop_sync_task SET validation_budget_json=#{validationBudgetJson},status=#{status},next_retry_at_utc=#{nextRetryAtUtc},updated_at=UTC_TIMESTAMP(3) WHERE id=#{id}") int saveRoot(SelfopSyncTask row);
+ @Update("UPDATE ll_selfop_sync_task SET validation_budget_json=#{validationBudgetJson},business_budget_json=#{businessBudgetJson},status=#{status},next_retry_at_utc=#{nextRetryAtUtc},updated_at=UTC_TIMESTAMP(3) WHERE id=#{id}") int saveRoot(SelfopSyncTask row);
+ @Update("UPDATE ll_selfop_sync_task SET status='DEAD',next_retry_at_utc=NULL,updated_at=UTC_TIMESTAMP(3) WHERE business_root_task_id=#{root} AND source_account=#{account} AND execution_purpose='BUSINESS' AND status IN ('WAITING_SOURCE','READY','RUNNING','RETRY_WAIT')") int terminateBusinessTasks(@Param("root") Long root,@Param("account") String account);
+ @Update("UPDATE ll_selfop_sync_run SET status='FAILED',error_code=#{reason},finished_at_utc=#{now},progress_snapshot_json=(SELECT progress_json FROM ll_selfop_sync_task WHERE id=task_id) WHERE source_account=#{account} AND execution_purpose='BUSINESS' AND task_id IN (SELECT id FROM ll_selfop_sync_task WHERE business_root_task_id=#{root} AND source_account=#{account} AND execution_purpose='BUSINESS') AND status IN ('FETCHING','VALIDATED','YIELDED','READY','BUILDING')") int terminateBusinessRuns(@Param("root") Long root,@Param("account") String account,@Param("reason") String reason,@Param("now") LocalDateTime now);
  @Update("UPDATE ll_selfop_sync_run SET status=#{status},error_code=#{reason},finished_at_utc=#{now},progress_snapshot_json=(SELECT progress_json FROM ll_selfop_sync_task WHERE id=task_id) WHERE task_id IN (SELECT id FROM ll_selfop_sync_task WHERE validation_root_task_id=#{root}) AND status IN ('FETCHING','VALIDATED','YIELDED')") int terminateValidationRuns(@Param("root") Long root,@Param("status") String status,@Param("reason") String reason,@Param("now") LocalDateTime now);
 }

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressService.java

@@ -19,7 +19,10 @@ public class SelfopEgressService {
   LocalDateTime now=now();gate(context.sourceAccount(),context.executionPurpose(),context.exclusiveGrantId(),now);
   if(operation==null||requestHash==null||!requestHash.matches("[a-f0-9]{64}"))throw reject("EGRESS_REQUEST_INVALID");
   var account=account(context.sourceAccount());var budget=budget(account,now);save(account,budget);var root=root(context.rootTaskId(),context.executionPurpose(),context.sourceAccount());
-  var task=mapper.lockTask(context.taskId());var run=mapper.lockRun(context.runId());verify(task,run,root,account,now);
+  var task=mapper.lockTask(context.taskId());var run=mapper.lockRun(context.runId());
+  SelfopBusinessBudget business=testBusiness(context.executionPurpose())?business(root):null;
+  if(business!=null){verifyBusinessTask(task,root);String limit=business.limit(now);if(limit!=null){stopBusiness(root,business,limit,now);throw reject(limit);}}
+  verify(task,run,root,account,now);
   checkAccount(account,budget,now);
   SelfopValidationBudget validation=null;
   if("VALIDATION".equals(context.executionPurpose())){
@@ -41,8 +44,10 @@ public class SelfopEgressService {
   var pending=new SelfopApiBudget.Pending();pending.context=outbound;pending.operation=operation;pending.internalRequestHash=requestHash;pending.taskId=context.taskId();pending.runId=context.runId();pending.expiresAtUtc=now.plusSeconds(Math.max(1,Math.min(120,config.getEgressLeaseSeconds())));
   if(pending.expiresAtUtc.isAfter(config.getExclusiveGrantExpiresAtUtc()))pending.expiresAtUtc=config.getExclusiveGrantExpiresAtUtc();
   if(validation!=null&&pending.expiresAtUtc.isAfter(validation.deadlineAtUtc))pending.expiresAtUtc=validation.deadlineAtUtc;
+  if(business!=null&&pending.expiresAtUtc.isAfter(business.deadlineAtUtc))pending.expiresAtUtc=business.deadlineAtUtc;
   budget.pending=pending;budget.reservedHttp++;budget.lastRequestAtUtc=now;
   if(validation!=null){validation.consume(operation,warehouseCode,detailKey);root.setValidationBudgetJson(SelfopSupport.json(validation));mapper.saveRoot(root);}
+  if(business!=null){business.http++;saveBusiness(root,business);}
   save(account,budget);return outbound;
  }
  @Transactional(noRollbackFor=EgressRejectedException.class)
@@ -50,7 +55,16 @@ public class SelfopEgressService {
   requireRequest(request);var c=request.context();var now=now();gate(c.sourceAccount(),c.executionPurpose(),c.exclusiveGrantId(),now);
   var account=account(c.sourceAccount());var budget=budget(account,now);var pending=pendingForAuthorization(budget,c,request.operation());
   var root=root(parseId(c.rootTaskId()),c.executionPurpose(),c.sourceAccount());
-  var task=mapper.lockTask(pending.taskId);var run=mapper.lockRun(pending.runId);verify(task,run,root,account,now);
+  var task=mapper.lockTask(pending.taskId);var run=mapper.lockRun(pending.runId);
+  if(testBusiness(c.executionPurpose())){
+   var b=business(root);verifyBusinessTask(task,root);
+   // The last reservation already spent its HTTP allowance. Recheck time/stop,
+   // not http>=maxHttp, or that paid reservation could never be authorized.
+   String limit=b.stopReason!=null?b.stopReason:!now.isBefore(b.deadlineAtUtc)?"BUSINESS_SESSION_DEADLINE":null;
+   if(limit!=null){stopBusiness(root,b,limit,now);throw reject(limit);}
+   if(pending.expiresAtUtc.isAfter(b.deadlineAtUtc))pending.expiresAtUtc=b.deadlineAtUtc;
+  }
+  verify(task,run,root,account,now);
   if(pending.consumed||pending.completed||!now.isBefore(pending.expiresAtUtc))throw reject("EGRESS_PERMIT_REPLAY_OR_EXPIRED");
   if(account.getApiPauseReason()!=null||account.getApiCooldownUntilUtc()!=null&&now.isBefore(account.getApiCooldownUntilUtc()))throw reject("ACCOUNT_PAUSED");
   if("VALIDATION".equals(c.executionPurpose())){var v=validation(root);if(v.stopReason!=null||!now.isBefore(v.deadlineAtUtc)){stop(root,v,v.stopReason==null?"VALIDATION_DEADLINE":v.stopReason,false,now);throw new EgressRejectedException("VALIDATION_STOPPED",true,null);}}
@@ -72,6 +86,7 @@ public class SelfopEgressService {
   var root=root(parseId(c.rootTaskId()),c.executionPurpose(),c.sourceAccount());
   SelfopValidationBudget validation="VALIDATION".equals(c.executionPurpose())?validation(root):null;
   if(validation!=null){validation.confirmedHttp+=request.actualHttpAttempts();root.setValidationBudgetJson(SelfopSupport.json(validation));mapper.saveRoot(root);}
+  if(testBusiness(c.executionPurpose())){var b=business(root);b.confirmedHttp+=request.actualHttpAttempts();b.validate();saveBusiness(root,b);}
   boolean auth=Integer.valueOf(401).equals(request.upstreamStatus())||Integer.valueOf(403).equals(request.upstreamStatus())
    ||"LINGXING_WMS_AUTHORIZATION_REJECTED".equals(request.errorCode())||"LINGXING_WMS_SIGNATURE_REJECTED".equals(request.errorCode());
   boolean rate=!auth&&(Integer.valueOf(429).equals(request.upstreamStatus())||Integer.valueOf(31001).equals(request.sourceCode())
@@ -146,12 +161,25 @@ public class SelfopEgressService {
  }
  private SelfopSyncTask root(Long id,String purpose,String account){var t=mapper.lockTask(id);if(t==null||!Objects.equals(t.getSourceAccount(),account)||!Objects.equals(t.getExecutionPurpose(),purpose))throw reject("EGRESS_ROOT_INVALID");if("VALIDATION".equals(purpose)&&!Objects.equals(id,t.getValidationRootTaskId()))throw reject("VALIDATION_ROOT_INVALID");return t;}
  private void verify(SelfopSyncTask task,SelfopSyncRun run,SelfopSyncTask root,SelfopOutboundDiscoveryScope account,LocalDateTime now){
+  if(testBusiness(root.getExecutionPurpose())&&"DEAD".equals(root.getStatus()))throw reject("BUSINESS_SESSION_ROOT_STOPPED");
   if(task==null||run==null||!Objects.equals(run.getTaskId(),task.getId())||!Objects.equals(run.getSourceAccount(),root.getSourceAccount())||!Objects.equals(task.getSourceAccount(),root.getSourceAccount())||!Objects.equals(task.getExecutionPurpose(),root.getExecutionPurpose())||!Objects.equals(run.getExecutionPurpose(),task.getExecutionPurpose())||!"RUNNING".equals(task.getStatus())||!Set.of("FETCHING","VALIDATED").contains(run.getStatus()))throw reject("EGRESS_TASK_NOT_ACTIVE");
   if("VALIDATION".equals(root.getExecutionPurpose())&&(!Objects.equals(task.getValidationRootTaskId(),root.getId())||Set.of("DEAD","SAMPLE_DONE").contains(root.getStatus())))throw reject("VALIDATION_ROOT_STOPPED");
   if("ACCOUNT_DISCOVERY".equals(task.getTaskRole())){if(!Objects.equals(task.getDiscoveryScopeId(),account.getId())||!Objects.equals(account.getFenceToken(),run.getFenceToken())||account.getLeaseUntilUtc()==null||!now.isBefore(account.getLeaseUntilUtc()))throw reject("EGRESS_FENCE_LOST");}
   else if(!Objects.equals(task.getScopeId(),run.getScopeId())||!Objects.equals(task.getWarehouseCode(),run.getWarehouseCode())||mapper.warehouseFence(task.getScopeId(),run.getFenceToken(),task.getId(),now)!=1)throw reject("EGRESS_FENCE_LOST");
  }
  private SelfopValidationBudget validation(SelfopSyncTask root){if(root==null||!"VALIDATION".equals(root.getExecutionPurpose())||!Objects.equals(root.getId(),root.getValidationRootTaskId())||root.getValidationBudgetJson()==null)throw reject("VALIDATION_ROOT_INVALID");var b=SelfopSupport.read(root.getValidationBudgetJson(),SelfopValidationBudget.class);b.validate();return b;}
+ private boolean testBusiness(String purpose){return "test".equals(config.getEnvironment())&&"BUSINESS".equals(purpose);}
+ private SelfopBusinessBudget business(SelfopSyncTask root){
+  if(root==null||!"BUSINESS".equals(root.getExecutionPurpose())||!Objects.equals(root.getId(),root.getBusinessRootTaskId())||root.getBusinessBudgetJson()==null)throw reject("BUSINESS_SESSION_ROOT_INVALID");
+  SelfopBusinessBudget b;try{b=SelfopSupport.read(root.getBusinessBudgetJson(),SelfopBusinessBudget.class);}catch(RuntimeException e){throw reject("BUSINESS_SESSION_BUDGET_INVALID");}
+  if(b==null)throw reject("BUSINESS_SESSION_BUDGET_INVALID");b.validate();return b;
+ }
+ private void verifyBusinessTask(SelfopSyncTask task,SelfopSyncTask root){if(task==null||!Objects.equals(task.getBusinessRootTaskId(),root.getId())||!Objects.equals(task.getSourceAccount(),root.getSourceAccount())||!"BUSINESS".equals(task.getExecutionPurpose()))throw reject("BUSINESS_SESSION_ROOT_INVALID");}
+ private void stopBusiness(SelfopSyncTask root,SelfopBusinessBudget b,String reason,LocalDateTime now){
+  b.stopReason=reason;root.setStatus("DEAD");root.setNextRetryAtUtc(null);saveBusiness(root,b);
+  mapper.terminateBusinessTasks(root.getId(),root.getSourceAccount());mapper.terminateBusinessRuns(root.getId(),root.getSourceAccount(),reason,now);
+ }
+ private void saveBusiness(SelfopSyncTask root,SelfopBusinessBudget b){root.setBusinessBudgetJson(SelfopSupport.json(b));if(mapper.saveRoot(root)!=1)throw new IllegalStateException("BUSINESS_SESSION_BUDGET_WRITE_FAILED");}
  private SelfopApiBudget.Pending pendingForAuthorization(SelfopApiBudget budget,EgressContext c,Operation op){var p=budget.pending;if(p==null||!Objects.equals(p.context,c)||p.operation!=op)throw reject("EGRESS_PERMIT_BINDING_INVALID");return p;}
  private SelfopApiBudget.Pending pending(SelfopApiBudget budget,EgressContext c,Operation op,String hash){var p=budget.pending;if(p==null||!Objects.equals(p.context,c)||p.operation!=op||!Objects.equals(p.requestHash,hash))throw reject("EGRESS_PERMIT_BINDING_INVALID");return p;}
  private void stop(SelfopSyncTask root,SelfopValidationBudget v,String reason,boolean failed,LocalDateTime now){if("DEAD".equals(root.getStatus())||"SAMPLE_DONE".equals(root.getStatus()))return;v.stopReason=reason;root.setValidationBudgetJson(SelfopSupport.json(v));root.setStatus(failed?"DEAD":"SAMPLE_DONE");root.setNextRetryAtUtc(null);mapper.saveRoot(root);mapper.terminateValidationRuns(root.getId(),failed?"FAILED":"SAMPLE_DONE",reason,now);}

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopCommandService.java

@@ -22,7 +22,7 @@ public class SelfopCommandService {
     public SelfopCommandService(SelfopTaskService taskService,SelfopSyncScopeMapper scopes,SelfopOutboundDiscoveryScopeMapper accounts,SelfopAnalyticsProperties config,SelfopSyncTaskMapper tasks){
         this.taskService=taskService;this.scopes=scopes;this.accounts=accounts;this.config=config;this.tasks=tasks;
     }
-    @Transactional
+    @Transactional(noRollbackFor=SelfopTaskService.BusinessSessionStoppedException.class)
     public SelfopSyncTask request(Command c,String operator){
         try { return requestChecked(c,operator); }
         catch(IllegalStateException failure){
@@ -103,6 +103,10 @@ public class SelfopCommandService {
         if(value==null||value.isBlank()||value.length()>max)throw new BizException(label+"不能为空且不能超过"+max+"个字符");
     }
     private static final Map<String,String> REJECTIONS=Map.ofEntries(
+        Map.entry("BUSINESS_SESSION_ROOT_REQUIRED","该测试任务没有有效的会话预算，请新建有界任务"),
+        Map.entry("BUSINESS_SESSION_ROOT_INVALID","测试采集会话关联无效，请核对任务记录"),
+        Map.entry("BUSINESS_SESSION_IDENTITY_INVALID","测试采集会话不能重新指定预算或根任务"),
+        Map.entry("BUSINESS_SESSION_BUDGET_INVALID","测试采集会话预算无效，请核对任务记录"),
         Map.entry("SELFOP_WAREHOUSE_UNCONFIGURED","该仓库尚未配置采集，暂不能提交任务"),
         Map.entry("SELFOP_TASK_SCOPE_MISMATCH","任务不存在或不属于当前账号和仓库"),
         Map.entry("SELFOP_TASK_MISSING","任务不存在，请刷新后重试"),

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java

@@ -1,6 +1,7 @@
 package com.baiyue.lastleg.selfoperated.analytics.sync;
 
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
 import com.baomidou.mybatisplus.core.toolkit.IdWorker;
 import com.baiyue.lastleg.selfoperated.analytics.client.LingxingAnalyticsClient;
 import com.baiyue.lastleg.selfoperated.analytics.client.AnalyticsClientException;
@@ -53,7 +54,7 @@ public class SelfopDiscoveryService {
                 require(page.total()!=null&&page.pages()!=null,"SELFOP_PAGE_METADATA_MISSING");
                 if(manifest.getTotal()==null){manifest.setTotal(page.total());manifest.setPages(page.pages());}
                 require(Objects.equals(manifest.getTotal(),page.total())&&Objects.equals(manifest.getPages(),page.pages()),"SELFOP_PAGE_TOTAL_CHANGED");
-                tx.executeWithoutResult(s->{leases.verify(lease);
+                tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);
                     for(var summary:page.records()) {
                         identity(summary.customerCode(),64);identity(summary.deliveryNo(),128);identity(summary.warehouseCode(),64);
                         String key=hash(List.of(task.getSourceAccount(),summary.customerCode(),summary.deliveryNo()));String content=hash(summary);
@@ -80,11 +81,11 @@ public class SelfopDiscoveryService {
                         if(manifest.getScan()==1){manifest.setScan(2);manifest.setNextPage(1);}
                         else{manifest.getObjects().keySet().removeIf(k->k.startsWith("seen:"));manifest.setDirectoryComplete(true);}
                     }
-                    saveManifest(task,manifest);
+                    saveManifest(checkpoint.task(),checkpoint.run(),manifest);
                 });
             }
         }
-        route(task,manifest,lease,false);
+        route(task,run,manifest,lease,false);
         for(var entry:manifest.getObjects().entrySet()) {
             var ref=entry.getValue();
             if(ref.getRefreshTaskId()!=null){
@@ -97,7 +98,7 @@ public class SelfopDiscoveryService {
                 var refreshed=read(refresh.getRoutingManifestJson(),SelfopRoutingManifest.class).getObjects().get(entry.getKey());
                 require(refreshed!=null&&refreshed.getEffectiveStageId()!=null,"SELFOP_REFRESH_REFERENCE_MISSING");
                 ref.setEffectiveStageId(refreshed.getEffectiveStageId());ref.setWarehouseCode(refreshed.getWarehouseCode());
-                tx.executeWithoutResult(s->{leases.verify(lease);saveManifest(task,manifest);});
+                tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);saveManifest(checkpoint.task(),checkpoint.run(),manifest);});
             }
             var latest=stages.selectOne(new QueryWrapper<SelfopSyncStage>().eq("object_type","OUTBOUND_ORDER")
                 .eq("object_key_hash",entry.getKey()).eq("validation_status","VALID").isNotNull("observation_sequence").orderByDesc("observation_sequence").last("LIMIT 1"));
@@ -128,11 +129,11 @@ public class SelfopDiscoveryService {
                 var locateQuery=new Query(null,null,ref.getDeliveryNo(),ref.getCustomerCode(),task.getId());relocate.setQueryScopeJson(json(locateQuery));relocate.setRequestHash(hash(relocate.getQueryScopeJson()));
                 fetchTask=taskService.create(relocate);fetchRun=taskService.begin(fetchTask.getId(),lease);ref.setRefreshTaskId(fetchTask.getId());
                 fetchTask.setAttemptCount(fetchRun.getAttemptNo());fetchTask.setStatus("RUNNING");
-                tx.executeWithoutResult(s->{leases.verify(lease);saveManifest(task,manifest);});
+                tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);saveManifest(checkpoint.task(),checkpoint.run(),manifest);});
                 var located=new SelfopRoutingManifest();
-                if(!collectLocated(fetchTask,fetchRun,lease,budget,locateQuery,located)){tx.executeWithoutResult(s->{leases.verify(lease);saveManifest(task,manifest);});taskService.yield(task.getId(),run.getId(),lease);return;}
+                if(!collectLocated(fetchTask,fetchRun,lease,budget,locateQuery,located)){tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);saveManifest(checkpoint.task(),checkpoint.run(),manifest);});taskService.yield(task.getId(),run.getId(),lease);return;}
                 ref.setWarehouseCode(located.getObjects().values().iterator().next().getWarehouseCode());
-                tx.executeWithoutResult(s->{leases.verify(lease);saveManifest(task,manifest);});
+                tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);saveManifest(checkpoint.task(),checkpoint.run(),manifest);});
                 if(!budget.take()){taskService.yield(fetchTask.getId(),fetchRun.getId(),lease);taskService.yield(task.getId(),run.getId(),lease);return;}
                 observation=leases.nextObservation(lease);detail=client.detail(SelfopRequestContext.forTask(fetchTask,fetchRun,config),ref.getWarehouseCode(),ref.getCustomerCode(),ref.getDeliveryNo());
             }
@@ -141,7 +142,9 @@ public class SelfopDiscoveryService {
             require(detail.items()!=null&&detail.sourceStatus()!=null,"SELFOP_DETAIL_STRUCTURE_INVALID");
             identity(detail.warehouseCode(),64);
             var finalFetchTask=fetchTask;var finalFetchRun=fetchRun;var acceptedDetail=detail;long acceptedObservation=observation;
-            tx.executeWithoutResult(s->{leases.verify(lease);
+            tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);
+                var fetchCheckpoint=finalFetchTask.getId().equals(task.getId())?checkpoint:taskService.checkpoint(finalFetchTask,finalFetchRun,lease);
+                var currentFetchTask=fetchCheckpoint.task();var currentFetchRun=fetchCheckpoint.run();
                 var existing=stages.selectOne(new QueryWrapper<SelfopSyncStage>().eq("stage_batch_id",finalFetchTask.getId()).eq("object_type","OUTBOUND_ORDER").eq("object_key_hash",entry.getKey()));
                 SelfopSyncStage full;
                 if(existing!=null&&"VALID".equals(existing.getValidationStatus())){
@@ -150,29 +153,30 @@ public class SelfopDiscoveryService {
                     full=stage(finalFetchTask,finalFetchRun,acceptedDetail,existing==null?1:existing.getFirstPageNo(),acceptedObservation,"VALID");
                     if(existing!=null){full.setId(existing.getId());full.setRunId(existing.getRunId());full.setCreatedAt(existing.getCreatedAt());stages.updateById(full);}else stages.insert(full);
                 }
-                ref.setEffectiveStageId(full.getId());ref.setWarehouseCode(acceptedDetail.warehouseCode());saveManifest(task,manifest);
-                if(!finalFetchTask.getId().equals(task.getId())){var refreshManifest=new SelfopRoutingManifest();refreshManifest.setDirectoryComplete(true);refreshManifest.setPages(1);refreshManifest.setTotal(1L);
+                ref.setEffectiveStageId(full.getId());ref.setWarehouseCode(acceptedDetail.warehouseCode());saveManifest(checkpoint.task(),checkpoint.run(),manifest);
+                if(!currentFetchTask.getId().equals(task.getId())){var refreshManifest=new SelfopRoutingManifest();refreshManifest.setDirectoryComplete(true);refreshManifest.setPages(1);refreshManifest.setTotal(1L);
                     var refreshRef=new SelfopRoutingManifest.Entry();refreshRef.setCustomerCode(ref.getCustomerCode());refreshRef.setDeliveryNo(ref.getDeliveryNo());refreshRef.setWarehouseCode(ref.getWarehouseCode());refreshRef.setEffectiveStageId(full.getId());
-                    refreshManifest.getObjects().put(entry.getKey(),refreshRef);finalFetchTask.setRoutingManifestJson(json(refreshManifest));
-                    SelfopProgressProjection.freezeRun(finalFetchTask,finalFetchRun);finalFetchRun.setStatus("ROUTED");finalFetchRun.setFetchFinishedAtUtc(utcNow());finalFetchRun.setFinishedAtUtc(utcNow());finalFetchRun.setUpdatedAt(utcNow());runs.updateById(finalFetchRun);
-                    finalFetchTask.setStatus("SUCCEEDED");finalFetchTask.setUpdatedAt(utcNow());tasks.updateById(finalFetchTask);}
+                    refreshManifest.getObjects().put(entry.getKey(),refreshRef);currentFetchTask.setRoutingManifestJson(json(refreshManifest));
+                    SelfopProgressProjection.freezeRun(currentFetchTask,currentFetchRun);currentFetchRun.setStatus("ROUTED");currentFetchRun.setFetchFinishedAtUtc(utcNow());currentFetchRun.setFinishedAtUtc(utcNow());currentFetchRun.setUpdatedAt(utcNow());runs.updateById(currentFetchRun);
+                    currentFetchTask.setStatus("SUCCEEDED");currentFetchTask.setUpdatedAt(utcNow());tasks.updateById(currentFetchTask);}
             });
-            route(task,manifest,lease,false);
+            route(task,run,manifest,lease,false);
         }
         // A directory warehouse is provisional until every account detail (including resumed
         // revision checks) has resolved. An empty warehouse must not publish a false zero early.
-        route(task,manifest,lease,true);
-        tx.executeWithoutResult(s->{leases.verify(lease);var now=utcNow();
-            SelfopCollectionPolicy.requireBusiness(task);SelfopCollectionPolicy.requireBusiness(run);SelfopProgressProjection.freezeRun(task,run);
-            run.setStatus("ROUTED");run.setCoverageStatus("COMPLETE");run.setPageCount(manifest.getPages()==null?0:manifest.getPages()*2);
-            run.setSourceTotalCount(manifest.getTotal());run.setUniqueObjectCount((long)manifest.getObjects().size());run.setFetchFinishedAtUtc(now);run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run);
-            task.setStatus("SUCCEEDED");task.setUpdatedAt(now);tasks.updateById(task);
-            var account=accounts.selectById(task.getDiscoveryScopeId());
+        route(task,run,manifest,lease,true);
+        tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);var currentTask=checkpoint.task();var currentRun=checkpoint.run();var now=utcNow();
+            SelfopCollectionPolicy.requireBusiness(currentTask);SelfopCollectionPolicy.requireBusiness(currentRun);SelfopProgressProjection.freezeRun(currentTask,currentRun);
+            currentRun.setStatus("ROUTED");currentRun.setCoverageStatus("COMPLETE");currentRun.setPageCount(manifest.getPages()==null?0:manifest.getPages()*2);
+            currentRun.setSourceTotalCount(manifest.getTotal());currentRun.setUniqueObjectCount((long)manifest.getObjects().size());currentRun.setFetchFinishedAtUtc(now);currentRun.setFinishedAtUtc(now);currentRun.setUpdatedAt(now);runs.updateById(currentRun);
+            currentTask.setStatus("SUCCEEDED");currentTask.setUpdatedAt(now);tasks.updateById(currentTask);
+            var account=accounts.selectOne(new QueryWrapper<SelfopOutboundDiscoveryScope>().eq("id",task.getDiscoveryScopeId()).last("FOR UPDATE"));
             if(task.getQueryFromUtc()!=null&&task.getQueryUntilUtc()!=null&&
                 (account.getCatalogUntilUtc()==null||!task.getQueryFromUtc().isAfter(account.getCatalogUntilUtc()))) {
                 account.setCatalogFromUtc(account.getCatalogFromUtc()==null?task.getQueryFromUtc():account.getCatalogFromUtc());
                 account.setCatalogUntilUtc(account.getCatalogUntilUtc()==null||task.getQueryUntilUtc().isAfter(account.getCatalogUntilUtc())?task.getQueryUntilUtc():account.getCatalogUntilUtc());
-                account.setUpdatedAt(now);accounts.updateById(account);}
+                account.setUpdatedAt(now);accounts.update(null,new UpdateWrapper<SelfopOutboundDiscoveryScope>().eq("id",account.getId())
+                    .set("catalog_from_utc",account.getCatalogFromUtc()).set("catalog_until_utc",account.getCatalogUntilUtc()).set("updated_at",now));}
         });
     }
     private boolean collectLocated(SelfopSyncTask task,SelfopSyncRun run,SelfopLeaseService.Lease lease,SelfopBudget budget,Query query,SelfopRoutingManifest manifest) {
@@ -189,7 +193,7 @@ public class SelfopDiscoveryService {
                 var e=new SelfopRoutingManifest.Entry();e.setCustomerCode(summary.customerCode());e.setDeliveryNo(summary.deliveryNo());e.setWarehouseCode(summary.warehouseCode());manifest.getObjects().put(key,e);}
             manifest.setNextPage(manifest.getNextPage()+1);if(page.pages()==0||manifest.getNextPage()>page.pages()){
                 require(manifest.getObjects().size()==1,"SELFOP_LOCATE_UNRESOLVED");manifest.setDirectoryComplete(true);}
-            tx.executeWithoutResult(s->{leases.verify(lease);saveManifest(task,manifest);});
+            tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);saveManifest(checkpoint.task(),checkpoint.run(),manifest);});
         }
         return true;
     }
@@ -199,17 +203,18 @@ public class SelfopDiscoveryService {
         s.setObjectKeyHash(hash(List.of(payload.sourceAccount(),payload.customerCode(),payload.deliveryNo())));s.setFirstPageNo(page);s.setObservedAtUtc(utcNow());
         s.setContentHash(SelfopOutboundFingerprint.order(payload));s.setPayloadJson(json(payload));s.setValidationStatus(status);s.setCreatedAt(utcNow());s.setUpdatedAt(utcNow());return s;
     }
-    private void saveManifest(SelfopSyncTask task,SelfopRoutingManifest manifest){task.setRoutingManifestJson(json(manifest));task.setUpdatedAt(utcNow());
-        var active=runs.selectOne(new QueryWrapper<SelfopSyncRun>().eq("task_id",task.getId()).eq("status","FETCHING").orderByDesc("attempt_no").last("LIMIT 1"));
-        if(active!=null){SelfopProgressProjection.applyCheckpoint(task,active,manifest.isDirectoryComplete()?"DETAIL":"DIRECTORY",manifest.isDirectoryComplete()?"ORDER":"PAGE",
+    private void saveManifest(SelfopSyncTask task,SelfopSyncRun active,SelfopRoutingManifest manifest){
+        task.setRoutingManifestJson(json(manifest));task.setUpdatedAt(utcNow());
+        SelfopProgressProjection.applyCheckpoint(task,active,manifest.isDirectoryComplete()?"DETAIL":"DIRECTORY",manifest.isDirectoryComplete()?"ORDER":"PAGE",
             manifest.isDirectoryComplete()?manifest.getObjects().values().stream().filter(e->e.getEffectiveStageId()!=null).count():Math.max(0,manifest.getNextPage()-1),
-            manifest.isDirectoryComplete()?(long)manifest.getObjects().size():manifest.getPages()==null?null:manifest.getPages().longValue(),manifest.getScan());runs.updateById(active);}
-        tasks.updateById(task);}
-    private void route(SelfopSyncTask parent,SelfopRoutingManifest manifest,SelfopLeaseService.Lease lease,boolean detailsResolved){
+            manifest.isDirectoryComplete()?(long)manifest.getObjects().size():manifest.getPages()==null?null:manifest.getPages().longValue(),manifest.getScan());
+        runs.updateById(active);tasks.updateById(task);
+    }
+    private void route(SelfopSyncTask parent,SelfopSyncRun parentRun,SelfopRoutingManifest manifest,SelfopLeaseService.Lease lease,boolean detailsResolved){
         // Refresh commands only replace a parent manifest reference; never create an independent
         // fact writer that could publish around that parent's half-applied barrier.
         if(read(parent.getQueryScopeJson(),Query.class).parentTaskId()!=null)return;
-        tx.executeWithoutResult(s->{leases.verify(lease);if(!manifest.isDirectoryComplete())return;
+        tx.executeWithoutResult(s->{taskService.checkpoint(parent,parentRun,lease);if(!manifest.isDirectoryComplete())return;
             boolean accountDetailsReady=detailsResolved&&manifest.getObjects().values().stream().allMatch(e->e.getEffectiveStageId()!=null);
             for(var scope:scopes.selectList(new QueryWrapper<SelfopSyncScope>().eq("source_account",parent.getSourceAccount()).eq("data_family","OUTBOUND"))){
                 if(!config.getWarehouses().containsKey(scope.getWarehouseCode()))continue;
@@ -232,7 +237,7 @@ public class SelfopDiscoveryService {
                 child.setRequestKey("route:"+parent.getId()+":"+scope.getWarehouseCode());child.setTaskType(parent.getTaskType());child.setQueryFromUtc(parent.getQueryFromUtc());child.setQueryUntilUtc(parent.getQueryUntilUtc());
                 child.setQueryScopeJson(scope.getScopeJson());child.setScopeHash(scope.getScopeHash());child.setRequestHash(hash(List.of(parent.getId(),scope.getScopeHash())));child.setStatus(ready?"READY":"WAITING_SOURCE");child.setApplyState("NOT_STARTED");
                 var saved=taskService.create(child);
-                if(!"SUCCEEDED".equals(saved.getStatus())&& !Set.of("APPLYING","APPLIED","BLOCKED").contains(saved.getApplyState())){
+                if(!Set.of("SUCCEEDED","DEAD","SAMPLE_DONE").contains(saved.getStatus())&& !Set.of("APPLYING","APPLIED","BLOCKED").contains(saved.getApplyState())){
                     saved.setRoutingManifestJson(json(own));saved.setStatus(ready?"READY":"WAITING_SOURCE");saved.setUpdatedAt(utcNow());tasks.updateById(saved);}
             }
         });

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryCollector.java

@@ -34,7 +34,7 @@ public class SelfopInventoryCollector {
             require(page.total()!=null&&page.pages()!=null,"SELFOP_PAGE_METADATA_MISSING");
             if(manifest.getTotal()==null){manifest.setTotal(page.total());manifest.setPages(page.pages());}
             require(Objects.equals(manifest.getTotal(),page.total())&&Objects.equals(manifest.getPages(),page.pages()),"SELFOP_PAGE_TOTAL_CHANGED");
-            tx.executeWithoutResult(s->{leases.verify(lease);
+            tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);var currentTask=checkpoint.task();var currentRun=checkpoint.run();
                 for(var row:page.records()){
                     require(task.getSourceAccount().equals(row.sourceAccount())&&task.getWarehouseCode().equals(row.warehouseCode()),"SELFOP_STOCK_SCOPE_LEAK");
                     identity(row.customerCode(),64);identity(row.skuCode(),128);require(row.productQuality()!=null,"SELFOP_STOCK_IDENTITY_MISSING");
@@ -49,13 +49,13 @@ public class SelfopInventoryCollector {
                     require(manifest.getFirstScan().size()==manifest.getTotal(),"SELFOP_STOCK_COUNT_MISMATCH");
                     manifest.setDirectoryComplete(true);
                 }
-                task.setRoutingManifestJson(json(manifest));task.setUpdatedAt(utcNow());tasks.updateById(task);
-                SelfopProgressProjection.applyCheckpoint(task,run,"INVENTORY","PAGE",Math.max(0,manifest.getNextPage()-1),manifest.getPages()==null?null:manifest.getPages().longValue(),manifest.getScan());tasks.updateById(task);runs.updateById(run);
+                currentTask.setRoutingManifestJson(json(manifest));currentTask.setUpdatedAt(utcNow());tasks.updateById(currentTask);
+                SelfopProgressProjection.applyCheckpoint(currentTask,currentRun,"INVENTORY","PAGE",Math.max(0,manifest.getNextPage()-1),manifest.getPages()==null?null:manifest.getPages().longValue(),manifest.getScan());tasks.updateById(currentTask);runs.updateById(currentRun);
             });
         }
-        tx.executeWithoutResult(s->{leases.verify(lease);run.setStatus("VALIDATED");run.setLastPhase("VALIDATED");run.setCoverageStatus("COMPLETE");
-            run.setPageCount(Math.max(1,manifest.getPages()));run.setSourceTotalCount(manifest.getTotal());run.setUniqueObjectCount((long)manifest.getFirstScan().size());
-            run.setFetchFinishedAtUtc(utcNow());run.setUpdatedAt(utcNow());runs.updateById(run);
+        tx.executeWithoutResult(s->{var checkpoint=taskService.checkpoint(task,run,lease);var currentTask=checkpoint.task();var currentRun=checkpoint.run();currentRun.setStatus("VALIDATED");currentRun.setLastPhase("VALIDATED");currentRun.setCoverageStatus("COMPLETE");
+            currentRun.setPageCount(Math.max(1,manifest.getPages()));currentRun.setSourceTotalCount(manifest.getTotal());currentRun.setUniqueObjectCount((long)manifest.getFirstScan().size());
+            currentRun.setFetchFinishedAtUtc(utcNow());currentRun.setUpdatedAt(utcNow());runs.updateById(currentRun);
         });return true;
     }
 }

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java

@@ -1,6 +1,7 @@
 package com.baiyue.lastleg.selfoperated.analytics.sync;
 
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget;
 import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
 import com.baomidou.mybatisplus.core.toolkit.IdWorker;
 import com.baiyue.lastleg.selfoperated.analytics.mapper.*;
@@ -88,11 +89,11 @@ public class SelfopTaskService {
         }
         return account;
     }
-    @Transactional
+    @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
     public SelfopSyncTask create(SelfopSyncTask task) {
         identity(task.getRequestKey(),128); var existing=tasks.selectOne(new QueryWrapper<SelfopSyncTask>()
             .eq("source_account",task.getSourceAccount()).eq("request_key",task.getRequestKey()));
-        if(existing!=null){require(existing.getRequestHash().equals(task.getRequestHash()),"SELFOP_IDEMPOTENCY_CONFLICT");return existing;}
+        if(existing!=null){require(existing.getRequestHash().equals(task.getRequestHash()),"SELFOP_IDEMPOTENCY_CONFLICT");businessGuard(existing,!localBusinessWork(existing));return existing;}
         if(task.getExecutionPurpose()==null)task.setExecutionPurpose(configuredPurpose(config));
         requireExecution(config,task.getExecutionPurpose());
         var now=utcNow();task.setId(IdWorker.getId());task.setAttemptCount(0);task.setCreatedAt(now);task.setUpdatedAt(now);
@@ -107,6 +108,20 @@ public class SelfopTaskService {
                     :com.baiyue.lastleg.selfoperated.analytics.request.SelfopValidationBudget.freeze(now));}
             task.setApplyState("NOT_APPLICABLE");
         }
+        if(testBusiness(task)) {
+            require(task.getBusinessRootTaskId()==null&&task.getBusinessBudgetJson()==null,"BUSINESS_SESSION_IDENTITY_INVALID");
+            if(task.getRequestKey().startsWith("manual:")&&task.getUpstreamTaskId()==null) {
+                require(controlledBusinessTask(task,tasks::selectById),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
+                task.setBusinessRootTaskId(task.getId());
+                task.setBusinessBudgetJson(SelfopBusinessBudget.freeze(now,config.getRequestBudget(),config.getTimeBudgetSeconds()));
+            }else {
+                var parent=businessParent(task);
+                require(parent!=null&&Objects.equals(parent.getSourceAccount(),task.getSourceAccount())&&controlledBusinessTask(parent,tasks::selectById),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
+                if("MANUAL_RETRY".equals(task.getTaskType()))require("SUCCEEDED".equals(task.getStatus())&&task.getRequestKey().startsWith("retry:")&&task.getUpstreamTaskId()!=null,"SELFOP_TEST_MANUAL_TASK_REQUIRED");
+                require("MANUAL_RETRY".equals(task.getTaskType())||controlledBusinessTask(task,tasks::selectById),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
+                businessGuard(parent,!"WAREHOUSE_SYNC".equals(task.getTaskRole()));task.setBusinessRootTaskId(parent.getBusinessRootTaskId());
+            }
+        }
         task.setStatus(task.getStatus()==null?"READY":task.getStatus());
         task.setApplyState(task.getApplyState()==null?"NOT_APPLICABLE":task.getApplyState());
         if("ACCOUNT_DISCOVERY".equals(task.getTaskRole())) {
@@ -115,12 +130,13 @@ public class SelfopTaskService {
         }
         try{tasks.insert(task);}catch(DuplicateKeyException ignored){existing=tasks.selectOne(new QueryWrapper<SelfopSyncTask>()
             .eq("source_account",task.getSourceAccount()).eq("request_key",task.getRequestKey()).last("FOR UPDATE"));
-            require(existing!=null&&existing.getRequestHash().equals(task.getRequestHash()),"SELFOP_IDEMPOTENCY_CONFLICT");return existing;}
+            require(existing!=null&&existing.getRequestHash().equals(task.getRequestHash()),"SELFOP_IDEMPOTENCY_CONFLICT");businessGuard(existing,!localBusinessWork(existing));return existing;}
         return task;
     }
-    @Transactional
+    @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
     public SelfopSyncRun begin(long taskId,SelfopLeaseService.Lease lease) {
-        leases.verify(lease);
+        lockBusinessAccount(taskId);leases.verify(lease);
+        if("test".equals(config.getEnvironment())){var preview=tasks.selectById(taskId);businessGuard(preview,!localBusinessWork(preview));}
         var task=locks.lockTask(taskId);require(task!=null&&!"SUCCEEDED".equals(task.getStatus())&&!"DEAD".equals(task.getStatus()),"SELFOP_TASK_NOT_EXECUTABLE");
         checkTaskLease(task,lease);
         requireExecution(config,task.getExecutionPurpose());
@@ -159,15 +175,17 @@ public class SelfopTaskService {
         String initialPhase=lease.account()?"DIRECTORY":("INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())?"INVENTORY":"MERGE");
         SelfopProgressProjection.applyCheckpoint(task,run,initialPhase,"PAGE",0,null,1);tasks.updateById(task);return run;
     }
-    @Transactional
+    @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
     public void yield(long taskId,long runId,SelfopLeaseService.Lease lease) {
-        leases.verify(lease);
+        lockBusinessAccount(taskId);leases.verify(lease);
+        guardBeforeTaskLock(taskId,true);
         var now=utcNow();var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run);SelfopProgressProjection.freezeRun(task,run);run.setStatus("YIELDED");run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run);
         task.setStatus("RETRY_WAIT");task.setNextRetryAtUtc(now.plusSeconds(1));task.setUpdatedAt(now);tasks.updateById(task);
     }
-    @Transactional
+    @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
     public void fail(long taskId,long runId,String code,boolean retryable,SelfopLeaseService.Lease lease) {
-        leases.verify(lease);
+        lockBusinessAccount(taskId);leases.verify(lease);
+        try{guardBeforeTaskLock(taskId,true);}catch(BusinessSessionStoppedException stopped){return;}
         var now=utcNow();var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run);SelfopProgressProjection.freezeRun(task,run);run.setLastPhase(run.getStatus());run.setStatus("FAILED");run.setErrorCode(code);
         run.setErrorMessage("自营同步失败，需核对运行记录");run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run);
         // A yielded execution is not a failed request; finite failures are counted separately.
@@ -184,19 +202,21 @@ public class SelfopTaskService {
         }
         task.setNextRetryAtUtc(now.plusSeconds(Math.min(3600,5L<<Math.min(9,(int)failures))));task.setUpdatedAt(now);tasks.updateById(task);
     }
-    @Transactional
+    @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
     public SelfopSyncTask retry(long id,String requestKey,String operator,String reason) {
         identity(reason,500);
         identity(requestKey,128);var preview=tasks.selectById(id);require(preview!=null,"SELFOP_TASK_MISSING");
+        lockBusinessAccount(id);
         var commandQuery=new QueryWrapper<SelfopSyncTask>().eq("source_account",preview.getSourceAccount()).eq("request_key","retry:"+requestKey);
         var prior=tasks.selectOne(commandQuery);
-        if(prior!=null){require(prior.getRequestHash().equals(hash(List.of(id,requestKey,reason))),"SELFOP_IDEMPOTENCY_CONFLICT");return preview;}
+        if(prior!=null){require(prior.getRequestHash().equals(hash(List.of(id,requestKey,reason))),"SELFOP_IDEMPOTENCY_CONFLICT");businessGuard(preview,!localBusinessWork(preview));return preview;}
         // Match executor lock ordering: account/warehouse scope first, task second.
         String activeOwner;LocalDateTime activeUntil;
         if("ACCOUNT_DISCOVERY".equals(preview.getTaskRole())){var scope=locks.lockDiscovery(preview.getDiscoveryScopeId());require(scope!=null,"SELFOP_SCOPE_MISSING");activeOwner=scope.getLeaseOwner();activeUntil=scope.getLeaseUntilUtc();}
         else{var scope=locks.lockScope(preview.getScopeId());require(scope!=null,"SELFOP_SCOPE_MISSING");activeOwner=scope.getLeaseOwner();activeUntil=scope.getLeaseUntilUtc();}
         prior=tasks.selectOne(new QueryWrapper<SelfopSyncTask>().eq("source_account",preview.getSourceAccount()).eq("request_key","retry:"+requestKey).last("FOR UPDATE"));
         if(prior!=null){require(prior.getRequestHash().equals(hash(List.of(id,requestKey,reason))),"SELFOP_IDEMPOTENCY_CONFLICT");return tasks.selectById(id);}
+        businessGuard(preview,!localBusinessWork(preview));
         require(SelfopLeaseService.available(activeOwner,activeUntil,utcNow()),"SELFOP_RETRY_LEASE_BUSY");
         var original=locks.lockTask(id);require(original!=null,"SELFOP_TASK_MISSING");
         requireBusiness(original);requireExecution(config,original.getExecutionPurpose());
@@ -212,21 +232,104 @@ public class SelfopTaskService {
         return original;
     }
     public List<SelfopSyncTask> due(int limit){return locks.due(utcNow(),limit,config.getCollectionMode());}
-    @Transactional
+    @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
     public void candidateFinished(long taskId,long runId,SelfopLeaseService.Lease lease) {
-        leases.verify(lease);var task=locks.lockTask(taskId);var run=locks.lockRun(runId);
+        lockBusinessAccount(taskId);leases.verify(lease);guardBeforeTaskLock(taskId,false);var task=locks.lockTask(taskId);var run=locks.lockRun(runId);
         checkRunLease(task,run,lease);
         require("READY".equals(run.getStatus()),"SELFOP_CANDIDATE_NOT_READY");
         run.setFinishedAtUtc(utcNow());run.setUpdatedAt(utcNow());runs.updateById(run);
         // Candidate readiness is not business success until a published result is linked.
         task.setStatus("RETRY_WAIT");task.setNextRetryAtUtc(utcNow().plusSeconds(config.getMinimumBuildIntervalSeconds()));task.setUpdatedAt(utcNow());tasks.updateById(task);
     }
-    @Transactional
+    @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
     public void waitForBudget(long taskId,long runId,LocalDateTime until,SelfopLeaseService.Lease lease) {
-        leases.verify(lease);var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run);
+        lockBusinessAccount(taskId);leases.verify(lease);if("test".equals(config.getEnvironment())){var preview=tasks.selectById(taskId);businessGuard(preview,!localBusinessWork(preview));}var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run);
         SelfopProgressProjection.freezeRun(task,run);run.setLastPhase(run.getStatus());run.setStatus("YIELDED");run.setFinishedAtUtc(utcNow());run.setUpdatedAt(utcNow());runs.updateById(run);
         task.setStatus("RETRY_WAIT");task.setNextRetryAtUtc(until);task.setUpdatedAt(utcNow());tasks.updateById(task);
     }
+    public record CollectionCheckpoint(SelfopSyncTask task,SelfopSyncRun run) { }
+    /** Caller holds the checkpoint transaction through all stage/progress writes. Never trust response-era entities. */
+    @Transactional(propagation=org.springframework.transaction.annotation.Propagation.MANDATORY)
+    public CollectionCheckpoint checkpoint(SelfopSyncTask observedTask,SelfopSyncRun observedRun,SelfopLeaseService.Lease lease) {
+        long taskId=observedTask.getId();lockBusinessAccount(taskId);leases.verify(lease);
+        if("test".equals(config.getEnvironment())) {
+            var preview=tasks.selectById(taskId);require(preview!=null,"SELFOP_TASK_MISSING");businessGuard(preview,false,false);
+        }
+        var task=locks.lockTask(taskId);var run=locks.lockRun(observedRun.getId());checkRunLease(task,run,lease);
+        requireBusiness(task);requireBusiness(run);
+        require("RUNNING".equals(task.getStatus())&&"FETCHING".equals(run.getStatus()),"SELFOP_COLLECTION_TERMINATED");
+        return new CollectionCheckpoint(task,run);
+    }
+    /** Only TEST manual collection has a session ceiling; production retains its resumable scheduler. */
+    public static final class BusinessSessionStoppedException extends com.baiyue.common.core.exception.BizException {
+        private final String reason;
+        BusinessSessionStoppedException(String reason){super("本次测试采集会话已结束或预算无效，请核对记录后新建有界任务");this.reason=reason;}
+        public String reason(){return reason;}
+    }
+    private boolean testBusiness(SelfopSyncTask task){return task!=null&&"test".equals(config.getEnvironment())&&"BUSINESS".equals(task.getExecutionPurpose());}
+    private SelfopSyncTask businessParent(SelfopSyncTask task){
+        Long id=task.getUpstreamTaskId();
+        if(id==null&&task.getQueryScopeJson()!=null){var query=read(task.getQueryScopeJson(),Map.class);if(query.get("parentTaskId") instanceof Number n)id=n.longValue();}
+        return id==null?null:tasks.selectById(id);
+    }
+    private void lockBusinessAccount(long id){
+        if(!"test".equals(config.getEnvironment()))return;
+        var task=tasks.selectById(id);if(!testBusiness(task))return;
+        require(task.getBusinessRootTaskId()!=null,"BUSINESS_SESSION_ROOT_REQUIRED");
+        var account=accounts.selectOne(new QueryWrapper<SelfopOutboundDiscoveryScope>().eq("source_account",task.getSourceAccount()));
+        require(account!=null,"SELFOP_SCOPE_MISSING");locks.lockDiscovery(account.getId());
+    }
+    private void guardBeforeTaskLock(long id,boolean checkDeadline){
+        if(!"test".equals(config.getEnvironment()))return;
+        var task=tasks.selectById(id);businessGuard(task,checkDeadline);
+    }
+    private boolean localBusinessWork(SelfopSyncTask task){
+        if(!testBusiness(task)||!"WAREHOUSE_SYNC".equals(task.getTaskRole()))return false;
+        var scope=scopes.selectById(task.getScopeId());
+        if(scope==null||!"OUTBOUND".equals(scope.getDataFamily()))return false;
+        if("REBUILD".equals(task.getTaskType())||"APPLIED".equals(task.getApplyState()))return true;
+        if(task.getRoutingManifestJson()==null)return false;
+        Map<?,?> manifest;
+        try{manifest=read(task.getRoutingManifestJson(),Map.class);}catch(IllegalArgumentException invalid){return false;}
+        if(manifest==null||!Boolean.TRUE.equals(manifest.get("directoryComplete"))||!(manifest.get("objects") instanceof Map<?,?> objects))return false;
+        return objects.values().stream().allMatch(value->{
+            if(!(value instanceof Map<?,?> entry))return false;
+            Object id=entry.get("effectiveStageId");
+            return (id instanceof Long||id instanceof Integer)&&((Number)id).longValue()>0;
+        });
+    }
+    private void businessGuard(SelfopSyncTask task,boolean checkDeadline){businessGuard(task,checkDeadline,true);}
+    private void businessGuard(SelfopSyncTask task,boolean checkDeadline,boolean terminate){
+        if(!testBusiness(task))return;
+        require(controlledBusinessTask(task,tasks::selectById),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
+        require(task.getBusinessRootTaskId()!=null,"BUSINESS_SESSION_ROOT_REQUIRED");
+        var ancestor=task;
+        while(!ancestor.getRequestKey().startsWith("manual:"))ancestor=businessParent(ancestor);
+        require(Objects.equals(ancestor.getId(),task.getBusinessRootTaskId()),"BUSINESS_SESSION_ROOT_INVALID");
+        // Serialize with egress in account -> root -> task order, before taking any task row lock.
+        var account=accounts.selectOne(new QueryWrapper<SelfopOutboundDiscoveryScope>().eq("source_account",task.getSourceAccount()));
+        require(account!=null,"SELFOP_SCOPE_MISSING");locks.lockDiscovery(account.getId());
+        var root=locks.lockTask(task.getBusinessRootTaskId());
+        require(root!=null&&Objects.equals(root.getId(),root.getBusinessRootTaskId())&&Objects.equals(root.getSourceAccount(),task.getSourceAccount())
+            &&root.getRequestKey()!=null&&root.getRequestKey().startsWith("manual:")&&controlledBusinessTask(root,tasks::selectById),"BUSINESS_SESSION_ROOT_INVALID");
+        require(root.getBusinessBudgetJson()!=null,"BUSINESS_SESSION_BUDGET_INVALID");
+        var budget=read(root.getBusinessBudgetJson(),SelfopBusinessBudget.class);budget.validate();
+        // HTTP exhaustion is enforced at egress, allowing the final permitted response to finish local work.
+        String reason=budget.stopReason;
+        if(reason==null&&checkDeadline&&!utcNow().isBefore(budget.deadlineAtUtc))reason="BUSINESS_SESSION_DEADLINE";
+        if(reason==null&&"DEAD".equals(root.getStatus()))reason="BUSINESS_SESSION_STOPPED";
+        if(reason==null)return;
+        if(!terminate)throw new BusinessSessionStoppedException(reason);
+        var now=utcNow();budget.stopReason=reason;
+        tasks.update(null,new UpdateWrapper<SelfopSyncTask>().eq("id",root.getId()).set("business_budget_json",json(budget)).set("status","DEAD").set("next_retry_at_utc",null).set("updated_at",now));
+        var members=tasks.selectList(new QueryWrapper<SelfopSyncTask>().eq("business_root_task_id",root.getId()).eq("source_account",root.getSourceAccount()).eq("execution_purpose","BUSINESS"));
+        var ids=new ArrayList<Long>();for(var member:members)ids.add(member.getId());if(!ids.contains(root.getId()))ids.add(root.getId());
+        runs.update(null,new UpdateWrapper<SelfopSyncRun>().in("task_id",ids).eq("source_account",root.getSourceAccount()).eq("execution_purpose","BUSINESS").in("status",List.of("FETCHING","VALIDATED","YIELDED","READY","BUILDING"))
+            .set("status","FAILED").set("error_code",reason).set("error_message","测试采集会话已结束").set("finished_at_utc",now).set("updated_at",now));
+        tasks.update(null,new UpdateWrapper<SelfopSyncTask>().in("id",ids).eq("source_account",root.getSourceAccount()).eq("execution_purpose","BUSINESS").in("status",List.of("WAITING_SOURCE","READY","RUNNING","RETRY_WAIT"))
+            .set("status","DEAD").set("next_retry_at_utc",null).set("updated_at",now));
+        throw new BusinessSessionStoppedException(reason);
+    }
     private static void checkTaskLease(SelfopSyncTask task,SelfopLeaseService.Lease lease){
         require(task!=null&&(lease.account()?"ACCOUNT_DISCOVERY".equals(task.getTaskRole())&&Objects.equals(task.getDiscoveryScopeId(),lease.scopeId())
             :"WAREHOUSE_SYNC".equals(task.getTaskRole())&&Objects.equals(task.getScopeId(),lease.scopeId())),"SELFOP_TASK_LEASE_SCOPE_MISMATCH");