From a84a9b730e6b836e40f97b3a43442892a007d0dc Mon Sep 17 00:00:00 2001
From: ouyangxq <775153912@qq.com>
Date: Tue, 13 Oct 2026 00:30:48 +0800
Subject: [PATCH] fix(selfop): publish proven manual sales windows for
 configured warehouses

---
 .../outbound/OutboundAggregateMapper.java     | 13 +++++++++--
 .../OutboundCandidateTransaction.java         |  8 +++++--
 .../sync/SelfopDiscoveryService.java          |  1 +
 .../outbound/OutboundAggregateMapperTest.java | 11 ++++++++-
 .../OutboundCandidateBuilderTest.java         | 23 +++++++++++++++++++
 .../sync/SelfopDiscoveryServiceTest.java      | 12 ++++++++++
 6 files changed, 63 insertions(+), 5 deletions(-)

diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapper.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapper.java
index 7469d3b6d..7594c32a6 100644
--- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapper.java
+++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapper.java
@@ -115,7 +115,7 @@ public interface OutboundAggregateMapper {
     @Select("""
         <script>
         SELECT id,query_from_utc,query_until_utc FROM ll_selfop_sync_task WHERE scope_id=#{scopeId} AND scope_hash=#{scopeHash}
-          AND execution_purpose='BUSINESS' AND apply_state='APPLIED'
+          AND execution_purpose='BUSINESS' AND validation_root_task_id IS NULL AND task_role='WAREHOUSE_SYNC' AND apply_state='APPLIED'
           AND query_from_utc IS NOT NULL AND query_until_utc IS NOT NULL
           AND JSON_UNQUOTE(JSON_EXTRACT(routing_manifest_json,'$.directoryComplete'))='true'
           <if test='id != null'>AND (query_from_utc > #{from} OR (query_from_utc=#{from} AND id > #{id}))</if>
@@ -125,6 +125,15 @@ public interface OutboundAggregateMapper {
     List<SelfopSyncTask> coverageBatch(@Param("scopeId") Long scopeId,@Param("scopeHash") String scopeHash,
                                       @Param("from") LocalDateTime from,@Param("id") Long id);
 
+    @Select("""
+        SELECT MIN(query_from_utc) FROM ll_selfop_sync_task
+        WHERE scope_id=#{scopeId} AND scope_hash=#{scopeHash}
+          AND execution_purpose='BUSINESS' AND validation_root_task_id IS NULL AND task_role='WAREHOUSE_SYNC' AND apply_state='APPLIED'
+          AND query_from_utc IS NOT NULL AND query_until_utc > query_from_utc
+          AND JSON_UNQUOTE(JSON_EXTRACT(routing_manifest_json,'$.directoryComplete'))='true'
+        """)
+    LocalDateTime firstProvenCoverageFrom(@Param("scopeId") Long scopeId,@Param("scopeHash") String scopeHash);
+
     @Select("""
         <script>
         SELECT CAST(#{initial} AS DATETIME) AS covered_from_utc,MAX(until_at) AS covered_until_utc
@@ -136,7 +145,7 @@ public interface OutboundAggregateMapper {
               OVER (ORDER BY from_at,until_at ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS previous_until
             FROM (
               SELECT query_from_utc AS from_at,query_until_utc AS until_at FROM ll_selfop_sync_task
-              WHERE scope_id=#{scopeId} AND scope_hash=#{scopeHash} AND execution_purpose='BUSINESS' AND apply_state='APPLIED'
+              WHERE scope_id=#{scopeId} AND scope_hash=#{scopeHash} AND execution_purpose='BUSINESS' AND validation_root_task_id IS NULL AND task_role='WAREHOUSE_SYNC' AND apply_state='APPLIED'
                 AND query_from_utc IS NOT NULL AND query_until_utc > query_from_utc AND query_until_utc >= #{initial}
                 AND JSON_UNQUOTE(JSON_EXTRACT(routing_manifest_json,'$.directoryComplete'))='true'
               UNION ALL SELECT CAST(#{initial} AS DATETIME),CAST(#{initial} AS DATETIME)
diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateTransaction.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateTransaction.java
index 7214b3d35..27883254e 100644
--- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateTransaction.java
+++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateTransaction.java
@@ -136,8 +136,12 @@ public class OutboundCandidateTransaction {
         SelfopSupport.require(aggregates.invalidFacts(scope.getSourceAccount(),scope.getWarehouseCode(),scope.getSourceTimezone(),scope.getBusinessTimezone(),unit)==0,"FACT_TIME_UNIT_OR_EVIDENCE_INVALID");
     }
     private OutboundCandidateBuilder.Coverage coverage(SelfopSyncScope scope) {
-        if(scope.getInitialFromUtc()==null)return new OutboundCandidateBuilder.Coverage(null,null);
-        SelfopSyncRun result=aggregates.contiguousCoverage(scope.getId(),scope.getScopeHash(),scope.getInitialFromUtc(),scope.getDiscoveryFromUtc(),scope.getDiscoveryUntilUtc());
+        // A manual first slice has no automatic backfill anchor. Only accepted, fully applied
+        // coverage may establish its start; an existing planned anchor must never move past gaps.
+        LocalDateTime initial=scope.getInitialFromUtc();
+        if(initial==null)initial=aggregates.firstProvenCoverageFrom(scope.getId(),scope.getScopeHash());
+        if(initial==null)return new OutboundCandidateBuilder.Coverage(null,null);
+        SelfopSyncRun result=aggregates.contiguousCoverage(scope.getId(),scope.getScopeHash(),initial,scope.getDiscoveryFromUtc(),scope.getDiscoveryUntilUtc());
         return result==null?new OutboundCandidateBuilder.Coverage(null,null):new OutboundCandidateBuilder.Coverage(result.getCoveredFromUtc(),result.getCoveredUntilUtc());
     }
     private Checkpoint checkpoint(SelfopSyncTask task) {
diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java
index fa0c1b1b8..6c835a3ae 100644
--- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java
+++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java
@@ -208,6 +208,7 @@ public class SelfopDiscoveryService {
         if(read(parent.getQueryScopeJson(),Query.class).parentTaskId()!=null)return;
         tx.executeWithoutResult(s->{leases.verify(lease);if(!manifest.isDirectoryComplete())return;
             for(var scope:scopes.selectList(new QueryWrapper<SelfopSyncScope>().eq("source_account",parent.getSourceAccount()).eq("data_family","OUTBOUND"))){
+                if(!config.getWarehouses().containsKey(scope.getWarehouseCode()))continue;
                 var own=new SelfopRoutingManifest();own.setDirectoryComplete(true);own.setTotal(0L);own.setPages(manifest.getPages());
                 for(var e:manifest.getObjects().entrySet()){
                     var ref=e.getValue();var previous=orders.selectOne(new QueryWrapper<SelfopOutboundOrder>().eq("source_account",parent.getSourceAccount()).eq("customer_code",ref.getCustomerCode()).eq("delivery_no",ref.getDeliveryNo()));
diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapperTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapperTest.java
index 5fc54a263..aade9be52 100644
--- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapperTest.java
+++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapperTest.java
@@ -20,7 +20,7 @@ public class OutboundAggregateMapperTest {
         sql("CREATE TABLE ll_selfop_outbound_item (id BIGINT PRIMARY KEY,order_id BIGINT,sku_code VARCHAR(128),product_name VARCHAR(255),quantity_unit VARCHAR(32),actual_quantity DECIMAL(20,6))");
         sql("CREATE TABLE ll_selfop_sync_stage (id BIGINT PRIMARY KEY,run_id BIGINT,warehouse_code VARCHAR(64),object_type VARCHAR(32),validation_status VARCHAR(24),observation_sequence BIGINT,observed_at_utc TIMESTAMP)");
         sql("CREATE TABLE ll_selfop_sync_run (id BIGINT PRIMARY KEY,source_account VARCHAR(64),task_id BIGINT,execution_purpose VARCHAR(24))");
-        sql("CREATE TABLE ll_selfop_sync_task (id BIGINT PRIMARY KEY,execution_purpose VARCHAR(24),validation_root_task_id BIGINT,scope_id BIGINT,scope_hash VARCHAR(64),apply_state VARCHAR(24),query_from_utc TIMESTAMP,query_until_utc TIMESTAMP,routing_manifest_json VARCHAR)");
+        sql("CREATE TABLE ll_selfop_sync_task (id BIGINT PRIMARY KEY,execution_purpose VARCHAR(24),validation_root_task_id BIGINT,task_role VARCHAR(32) DEFAULT 'WAREHOUSE_SYNC',scope_id BIGINT,scope_hash VARCHAR(64),apply_state VARCHAR(24),query_from_utc TIMESTAMP,query_until_utc TIMESTAMP,routing_manifest_json VARCHAR)");
         sql("CREATE ALIAS JSON_EXTRACT FOR 'com.baiyue.lastleg.selfoperated.analytics.outbound.OutboundAggregateMapperTest.jsonExtract'");
         sql("CREATE ALIAS JSON_UNQUOTE FOR 'com.baiyue.lastleg.selfoperated.analytics.outbound.OutboundAggregateMapperTest.jsonUnquote'");
         sql("CREATE TABLE ll_selfop_sku_daily_sales (id BIGINT PRIMARY KEY,created_at TIMESTAMP,updated_at TIMESTAMP,sales_run_id BIGINT,source_account VARCHAR(64),customer_code VARCHAR(64),warehouse_code VARCHAR(64),sku_code VARCHAR(128),business_date DATE,outbound_quantity DECIMAL(20,6),outbound_order_count BIGINT,quantity_unit VARCHAR(32))");
@@ -80,6 +80,15 @@ public class OutboundAggregateMapperTest {
         return new com.fasterxml.jackson.databind.ObjectMapper().readTree(value).path("directoryComplete").asText();
     }
     public static String jsonUnquote(String value) {return value;}
+    @Test void firstManualCoverageRequiresMatchingScopeAppliedCompleteBusinessEvidence() throws Exception {
+        sql("INSERT INTO ll_selfop_sync_task (id,execution_purpose,scope_id,scope_hash,apply_state,query_from_utc,query_until_utc,routing_manifest_json) VALUES (1,'VALIDATION',1,'h','APPLIED','2026-10-01 00:00:00','2026-10-01 00:05:00','{\"directoryComplete\":true}'),(2,'BUSINESS',1,'h','NOT_STARTED','2026-10-02 00:00:00','2026-10-02 00:05:00','{\"directoryComplete\":true}'),(3,'BUSINESS',1,'h','APPLIED','2026-10-03 00:00:00','2026-10-03 00:05:00','{\"directoryComplete\":false}'),(4,'BUSINESS',2,'h','APPLIED','2026-10-04 00:00:00','2026-10-04 00:05:00','{\"directoryComplete\":true}'),(5,'BUSINESS',1,'old','APPLIED','2026-10-05 00:00:00','2026-10-05 00:05:00','{\"directoryComplete\":true}')");
+        sql("INSERT INTO ll_selfop_sync_task (id,execution_purpose,scope_id,scope_hash,apply_state,query_from_utc,query_until_utc,routing_manifest_json,task_role,validation_root_task_id) VALUES (7,'BUSINESS',1,'h','APPLIED','2026-10-01 00:00:00','2026-10-01 00:05:00','{\"directoryComplete\":true}','ACCOUNT_DISCOVERY',NULL),(8,'BUSINESS',1,'h','APPLIED','2026-10-01 00:00:00','2026-10-01 00:05:00','{\"directoryComplete\":true}','WAREHOUSE_SYNC',99)");
+        assertNull(mapper.firstProvenCoverageFrom(1L,"h"));
+        sql("INSERT INTO ll_selfop_sync_task (id,execution_purpose,scope_id,scope_hash,apply_state,query_from_utc,query_until_utc,routing_manifest_json) VALUES (6,'BUSINESS',1,'h','APPLIED','2026-10-10 01:00:00','2026-10-10 01:05:00','{\"directoryComplete\":true}')");
+        var start=mapper.firstProvenCoverageFrom(1L,"h");assertEquals(LocalDateTime.of(2026,10,10,1,0),start);
+        assertEquals(start.plusMinutes(5),mapper.contiguousCoverage(1L,"h",start,null,null).getCoveredUntilUtc());
+    }
+
     @Test void databaseCoverageJoinsOverlapsAndStopsAtFirstUnprovedGap() throws Exception {
         sql("INSERT INTO ll_selfop_sync_task (id,execution_purpose,scope_id,scope_hash,apply_state,query_from_utc,query_until_utc,routing_manifest_json) VALUES (1,'BUSINESS',1,'h','APPLIED','2026-10-01 00:00:00','2026-10-03 00:00:00','{\"directoryComplete\":true}'),(2,'BUSINESS',1,'h','APPLIED','2026-10-02 00:00:00','2026-10-04 00:00:00','{\"directoryComplete\":true}'),(3,'BUSINESS',1,'h','APPLIED','2026-10-05 00:00:00','2026-10-06 00:00:00','{\"directoryComplete\":true}'),(4,'VALIDATION',1,'h','APPLIED','2026-10-04 00:00:00','2026-10-05 00:00:00','{\"directoryComplete\":true}')");
         var coverage=mapper.contiguousCoverage(1L,"h",LocalDateTime.of(2026,10,1,0,0),null,null);
diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilderTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilderTest.java
index 4a8a2c886..6e5f35a57 100644
--- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilderTest.java
+++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuilderTest.java
@@ -38,6 +38,29 @@ class OutboundCandidateBuilderTest {
         SelfopSyncRun coverage=new SelfopSyncRun();coverage.setCoveredFromUtc(task.getQueryFromUtc());coverage.setCoveredUntilUtc(task.getQueryUntilUtc());
         when(aggregates.contiguousCoverage(1L,"scope",scope.getInitialFromUtc(),null,null)).thenReturn(coverage);
     }
+    @Test void manualFirstSliceUsesOnlyProvenIntervalWithoutInventingFullDays() {
+        scope.setInitialFromUtc(null);
+        var start=LocalDateTime.of(2026,10,9,17,0); var end=start.plusMinutes(5);
+        when(aggregates.firstProvenCoverageFrom(1L,"scope")).thenReturn(start);
+        var coverage=new SelfopSyncRun();coverage.setCoveredFromUtc(start);coverage.setCoveredUntilUtc(end);
+        when(aggregates.contiguousCoverage(1L,"scope",start,null,null)).thenReturn(coverage);
+        builder.build(1L,10L,"worker",2);
+        assertEquals(start,run.getCoveredFromUtc());assertEquals(end,run.getCoveredUntilUtc());
+        assertEquals("COMPLETE",run.getCoverageStatus());assertNull(scope.getInitialFromUtc());
+        assertTrue(run.getCompleteDayFrom()==null || !run.getCompleteDayFrom().isBefore(run.getCompleteDayUntil()));
+    }
+    @Test void missingProvenSliceCannotInventCoverage() {
+        scope.setInitialFromUtc(null);builder.build(1L,10L,"worker",2);
+        assertEquals("UNKNOWN",run.getCoverageStatus());assertNull(run.getCoveredFromUtc());assertNull(run.getCoveredUntilUtc());
+        verify(aggregates,never()).contiguousCoverage(any(),any(),any(),any(),any());
+    }
+    @Test void existingPlannedAnchorCannotMoveForwardPastMissingHistory() {
+        when(aggregates.contiguousCoverage(1L,"scope",scope.getInitialFromUtc(),null,null)).thenReturn(null);
+        builder.build(1L,10L,"worker",2);
+        assertEquals("UNKNOWN",run.getCoverageStatus());assertNull(run.getCoveredUntilUtc());
+        verify(aggregates,never()).firstProvenCoverageFrom(any(),any());
+    }
+
     @Test void fullRebuildWithNoCurrentFactsRetractsAllRemovedEvidence() {
         builder.build(1L,10L,"worker",2);
         verify(profiles).delete(any()); verify(aggregates,never()).insertProfiles(any());
diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java
index deb6fab33..6b4b22b04 100644
--- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java
+++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java
@@ -54,6 +54,7 @@ class SelfopDiscoveryServiceTest {
         scope = new SelfopSyncScope(); scope.setId(10L); scope.setSourceAccount("demo"); scope.setWarehouseCode("A");
         scope.setDataFamily("OUTBOUND"); scope.setScopeHash("warehouse-hash"); scope.setScopeJson("{}"); scope.setSourceTimezone("UTC");
         config.setCollectionMode("BUSINESS"); config.setEnvironment("production"); config.setPageSize(2); config.setSourceQueryTimezone("UTC"); config.setQuantityUnit("PCS");
+        config.getWarehouses().put("A", new SelfopAnalyticsProperties.Warehouse());
         when(scopes.selectList(any())).thenReturn(List.of(scope));
         when(leases.nextObservation(lease)).thenReturn(101L, 102L, 103L);
         when(taskService.create(any(SelfopSyncTask.class))).thenAnswer(inv -> {
@@ -89,6 +90,17 @@ class SelfopDiscoveryServiceTest {
             summary("one", "2026-10-11 12:00:00"), summary("two", "2026-10-11 12:00:00")));
     }
 
+    @Test void removedWarehouseScopeDoesNotReceiveRoutedTask() {
+        var removed = new SelfopSyncScope(); removed.setId(11L); removed.setSourceAccount("demo");
+        removed.setWarehouseCode("REMOVED"); removed.setDataFamily("OUTBOUND");
+        removed.setScopeHash("old-hash"); removed.setScopeJson("{}"); removed.setSourceTimezone("UTC");
+        when(scopes.selectList(any())).thenReturn(List.of(scope, removed));
+        directoryTwoOrders(); service.collect(task, run, lease);
+        assertEquals(1, children.size());
+        assertTrue(children.values().stream().allMatch(child -> "A".equals(child.getWarehouseCode())));
+        assertEquals("ROUTED", run.getStatus());
+    }
+
     @Test void paginatedLocateYieldsDurableCursorThenResumesWithoutFiniteFailure() {
         config.setPageSize(1); account.setRequestBudget(1);
         task.setQueryScopeJson(json(new SelfopDiscoveryService.Query(null, null, "target", "customer", null)));