diff --git a/deploy/sql/README.md b/deploy/sql/README.md
index da5775a5d..818b00712 100644
--- a/deploy/sql/README.md
+++ b/deploy/sql/README.md
@@ -22,7 +22,7 @@
内部拼车需求池的当前结构在 Lastleg 源基线和海外统一空库基线同步维护,独立五项权限在 IAM 源种子和国内统一基线维护。已存在基础表的指定环境结构补齐使用 `migrations/overseas/baiyue_lastleg/V10__fleet_capacity_pool.sql` 与 `migrations/cn/baiyue_iam/V4__fleet_capacity_offer_permissions.sql`;执行前确认三张新表、六个新增字段均未存在,并单独核对目标环境。随后 `V11__align_capacity_pool_collation.sql` 将预订与商业桥的客户关联字段显式统一为 `utf8mb4_unicode_ci`,与原订单域一致,避免依赖MySQL实例默认排序规则导致JOIN失败。它们不回填、删除或重置业务数据,不能重放整个空库基线。候选文件存在不表示已经执行。
-平台自营领星统计十张 `ll_selfop_*` 表直接同步 Lastleg 源基线及海外统一空库基线;时间使用 UTC 存储及已确认北京时间业务日,数量使用 DECIMAL 保留 PCS 精度,维持独立发布版本。包含账号共享预算/冷却、验证根用途与硬预算、稳定任务号、阶段进度/心跳和旧版替代时间;库存 run 增加每日代表日期、选择规则及账号/仓库/日期唯一约束,代表库存与 profile 不适用销量副本淘汰。IAM 权限源为 `iam/src/main/resources/db/selfoperated-analytics-permissions.sql`,自营业务下经营总览、销量分析、库存分析、数据采集的菜单资源与独立读写权限同步国内统一基线,仅平台管理员默认授权。本批只交付文件与离线验证,不执行 SQL、不部署;真实采集模式默认 OFF。
+平台自营领星统计十张 `ll_selfop_*` 表直接同步 Lastleg 源基线及海外统一空库基线;时间使用 UTC 存储及已确认北京时间业务日,数量使用 DECIMAL 保留 PCS 精度,维持独立发布版本。包含账号共享预算/冷却、验证根用途与硬预算、TEST BUSINESS 根会话持久预算、稳定任务号、阶段进度/心跳和旧版替代时间;任务的 `business_root_task_id` 标记共享根,只有根保存 `business_budget_json`,生产任务不使用该测试预算。库存 run 包含每日代表日期、选择规则及账号/仓库/日期唯一约束,代表库存与 profile 不适用销量副本淘汰。IAM 权限源为 `iam/src/main/resources/db/selfoperated-analytics-permissions.sql`,自营业务下经营总览、销量分析、库存分析、数据采集的菜单资源与独立读写权限同步国内统一基线,仅平台管理员默认授权。执行和部署状态只维护于工作区测试环境主契约;真实采集模式默认 OFF。
## 唯一入口与区域边界
diff --git a/deploy/sql/海外区域统一增量-MySQL8.sql b/deploy/sql/海外区域统一增量-MySQL8.sql
index 4a505e045..9fe44e578 100644
--- a/deploy/sql/海外区域统一增量-MySQL8.sql
+++ b/deploy/sql/海外区域统一增量-MySQL8.sql
@@ -3990,6 +3990,8 @@ CREATE TABLE ll_selfop_sync_task (
execution_purpose VARCHAR(16) NOT NULL COMMENT 'BUSINESS或VALIDATION;由服务端冻结用途,子任务继承',
validation_root_task_id BIGINT NULL COMMENT '验证会话唯一预算根任务,根引用自身,正式任务为空',
validation_budget_json JSON NULL COMMENT '验证根冻结硬预算、累计消耗、库存分仓页数、截止时刻及停止原因',
+ business_root_task_id BIGINT NULL COMMENT '测试人工正式采集共享预算根任务;生产任务为空',
+ business_budget_json JSON NULL COMMENT '仅测试正式采集根保存冻结请求上限、截止时间、消耗及停止原因',
source_account VARCHAR(64) NOT NULL COMMENT '受控来源账号别名,不保存密钥',
task_role VARCHAR(24) NOT NULL COMMENT 'ACCOUNT_DISCOVERY 或 WAREHOUSE_SYNC',
scope_id BIGINT NULL COMMENT '分仓任务必须引用分仓作用域;仅账号共享发现为空',
@@ -4026,6 +4028,8 @@ CREATE TABLE ll_selfop_sync_task (
KEY idx_selfop_task_apply (apply_state,id),
KEY idx_selfop_task_scope_list (scope_id,created_at,id),
KEY idx_selfop_task_account_list (source_account,task_role,created_at,id),
+ KEY idx_selfop_task_business_root (business_root_task_id,status,id),
+ CONSTRAINT chk_selfop_business_budget CHECK ((business_root_task_id IS NULL AND business_budget_json IS NULL) OR (execution_purpose='BUSINESS' AND business_root_task_id IS NOT NULL AND ((business_root_task_id=id AND business_budget_json IS NOT NULL) OR (business_root_task_id<>id AND business_budget_json IS NULL)))),
CONSTRAINT chk_selfop_sync_task CHECK (attempt_count >= 0 AND execution_purpose IN ('BUSINESS','VALIDATION') AND ((execution_purpose = 'BUSINESS' AND validation_root_task_id IS NULL AND validation_budget_json IS NULL) OR (execution_purpose = 'VALIDATION' AND validation_root_task_id IS NOT NULL AND apply_state = 'NOT_APPLICABLE' AND result_published_run_id IS NULL AND status <> 'SUCCEEDED')) AND ((task_role = 'ACCOUNT_DISCOVERY' AND scope_id IS NULL AND warehouse_code IS NULL AND discovery_scope_id IS NOT NULL) OR (task_role = 'WAREHOUSE_SYNC' AND scope_id IS NOT NULL AND warehouse_code IS NOT NULL)))
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='一次计划或人工请求';
diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRequestContext.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRequestContext.java
index 50b42c69d..c8fb94183 100644
--- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRequestContext.java
+++ b/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());
}
diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopSyncTask.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopSyncTask.java
index 9bf50a686..bf537782c 100644
--- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopSyncTask.java
+++ b/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;
diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java
index 40ecaf3d6..d6f39b2ee 100644
--- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java
+++ b/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 "采集尝试失败,请根据错误码查看受控诊断记录";
diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java
index 916790867..e882c4872 100644
--- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java
+++ b/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("")
List tasksByIds(@Param("account") String account,@Param("ids") List ids);
@Select("