feat(selfop): add bounded inventory flow diagnostic through account egress

2026-10-15 11:37:49 +08:00
parent e19d303a76
commit d81d9cbda4
16 changed files with 365 additions and 5 deletions
@@ -78,6 +78,15 @@ public class LingxingInternalController {
        return ApiResponse.success(client.queryDetail(request, context(context)));
    }

+    @Operation(summary = "单SKU最多31日库存流水诊断，仅第一页最多1000条")
+    @PostMapping("/inventory-flow/query")
+    public ApiResponse<QueryPage<InventoryFlow>> inventoryFlow(
+        @RequestHeader(value="X-Internal-Service-Token",required=false) String token,
+        @RequestHeader(value="X-Lingxing-Egress-Context",required=false) String context,
+        @Valid @RequestBody InventoryFlowRequest request) {
+        tokenGuard.verify(token);return ApiResponse.success(client.queryInventoryFlow(request,context(context)));
+    }
+
    @Operation(summary = "读取单仓全部货主、全部品质且保留零库存的产品库存单页")
    @PostMapping("/product-inventory/query")
    public ApiResponse<QueryPage<ProductStock>> inventory(
@@ -6,7 +6,7 @@ import java.time.Instant;
 public final class LingxingEgressDtos {
    private LingxingEgressDtos() { }
    public enum CollectionMode { OFF, VALIDATION, BUSINESS }
-    public enum Operation { WAREHOUSES, OUTBOUND_QUERY, OUTBOUND_LOCATE, OUTBOUND_DETAIL, PRODUCT_INVENTORY, CUSTOMERS, PRODUCTS }
+    public enum Operation { WAREHOUSES, OUTBOUND_QUERY, OUTBOUND_LOCATE, OUTBOUND_DETAIL, PRODUCT_INVENTORY, CUSTOMERS, PRODUCTS, INVENTORY_FLOW }
    public record EgressContext(String sourceAccount, CollectionMode executionPurpose, String rootTaskId,
                                String sessionId, String attemptId, String exclusiveGrantId) {
        public boolean wellFormed() {
@@ -20,6 +20,12 @@ public final class LingxingWmsDtos {

    public record WarehouseQueryRequest(@Min(1) int current, @Min(1) @Max(1000) int size) { }
    public record ProductQueryRequest(@Min(1) int current, @Min(1) @Max(100) int size) { }
+    public record InventoryFlowRequest(@NotBlank @Size(max=64) String warehouseCode,
+        @NotBlank @Size(max=64) String customerCode, @NotBlank @Size(max=128) String sku,
+        @NotNull LocalDateTime startTime, @NotNull LocalDateTime endTime) { }
+    public record InventoryFlow(String id, String customerCode, String whCode, String productSku,
+        String cellNo, String innerBatchNo, String billNo, String billType, Integer productQuality,
+        BigDecimal bookQty, String bookChangQty, BigDecimal closeQty, String createTime) { }
    public record Customer(String customerCode, String customerName, String status) { }
    public record Product(String customerCode, String sku, String productName,
                          BigDecimal length, BigDecimal width, BigDecimal height, String sizeUnit,
@@ -162,6 +162,30 @@ public final class LingxingWmsClient {
            (data, metadata) -> inventory(data, request, metadata));
    }

+    /** Single bounded diagnostic page; never changes the current-inventory collection protocol. */
+    public QueryPage<InventoryFlow> queryInventoryFlow(InventoryFlowRequest request, EgressContext context) {
+        if(request==null)throw badRequest();
+        requireInput(request.warehouseCode(),64);requireInput(request.customerCode(),64);requireInput(request.sku(),128);
+        if(request.startTime()==null||request.endTime()==null||!request.startTime().isBefore(request.endTime())
+            ||request.endTime().isAfter(request.startTime().plusDays(31)))throw badRequest();
+        Map<String,Object> body=pagination(1,1000);
+        body.put("whCode",request.warehouseCode());body.put("customerCode",request.customerCode());
+        body.put("barcodeAccurate",request.sku());body.put("startTime",SOURCE_TIME.format(request.startTime()));
+        body.put("endTime",SOURCE_TIME.format(request.endTime()));body.put("sortName","createTime");body.put("sortOrder","asc");
+        return execute("/openapi/v2/stock/stockCellBook/page",body,context,Operation.INVENTORY_FLOW,
+            (data,metadata)->mappedPage(data,1,1000,"current","size",false,metadata,row->{
+                if(!request.warehouseCode().equals(requiredText(row,"whCode",64))
+                    ||!request.customerCode().equals(requiredText(row,"customerCode",64))
+                    ||!request.sku().equals(requiredText(row,"productSku",128)))throw invalidResponse("FLOW_IDENTITY_MISMATCH",200);
+                JsonNode id=row.get("id");
+                if(id==null||!id.isIntegralNumber())throw invalidResponse("FLOW_ID_MISSING",200);
+                return new InventoryFlow(id.asText(),request.customerCode(),request.warehouseCode(),request.sku(),
+                    optionalText(row,"cellNo"),optionalText(row,"innerBatchNo"),optionalText(row,"billNo"),optionalText(row,"billType"),
+                    requiredInt(row.get("productQuality"),"PRODUCT_QUALITY"),quantity(row.get("bookQty")),
+                    optionalText(row,"bookChangQty"),quantity(row.get("closeQty")),optionalText(row,"createTime"));
+            }));
+    }
+
    private QueryPage<ProductStock> inventory(JsonNode data, InventoryQueryRequest request, AttemptMetadata metadata) {
        Set<List<Object>> keys = new HashSet<>();
        return mappedPage(data, request.current(), request.size(), "current", "size", false, metadata, row -> {
@@ -203,6 +227,9 @@ public final class LingxingWmsClient {
            JsonNode json;
            try { json = envelope(response.body()); }
            catch (LingxingWmsException error) {
+                // This optional endpoint may lack permission while ordinary collection remains authorized.
+                if(operation==Operation.INVENTORY_FLOW&&Integer.valueOf(11008).equals(error.sourceCode()))
+                    throw upstreamFailure("FLOW_PERMISSION_DENIED","当前账号没有库存流水接口权限",false,200,11008);
                if (Integer.valueOf(31001).equals(error.sourceCode()) || Integer.valueOf(429).equals(error.sourceCode())) {
                    handle.rateLimited(retryAfter(response));
                }
@@ -51,6 +51,32 @@ x-lastleg-egress-coordination:
 security:
  - InternalServiceToken: []
 paths:
+  /internal/integrations/lingxing/wms/inventory-flow/query:
+    post:
+      summary: One SKU inventory-flow diagnostic page
+      description: >-
+        Calls stock/stockCellBook/page through the existing INVENTORY_FLOW permit. Exact warehouse/customer/SKU,
+        at most 31 days, current=1 and size=1000 fixed. No automatic next page or retry. Source quality and signed
+        quantities are preserved, not translated to inventory-snapshot quality. Source 11008 only rejects this optional
+        operation; authentication, rate-limit and uncertain-result account protection remain effective.
+      parameters:
+        - {$ref: '#/components/parameters/LingxingEgressContext'}
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [warehouseCode, customerCode, sku, startTime, endTime]
+              properties:
+                warehouseCode: {type: string, maxLength: 64}
+                customerCode: {type: string, maxLength: 64}
+                sku: {type: string, maxLength: 128}
+                startTime: {type: string, format: date-time, description: Beijing local datetime without offset}
+                endTime: {type: string, format: date-time, description: Beijing local datetime without offset}
+      responses:
+        '200':
+          description: Standard envelope with QueryPage or sanitized ProviderFailure. Flow rows retain id as string, customerCode, whCode, productSku, cellNo, innerBatchNo, billNo, billType, productQuality, numeric bookQty/closeQty, string bookChangQty/createTime. Unknown source fields remain null; no calculated inventory is published.
  /internal/integrations/lingxing/wms/customers/query:
    post:
      summary: Query one page of customers using the existing account egress permit
@@ -315,7 +341,7 @@ components:
        {"current":1,"size":2,"whCode":"WH-DEMO","isHideInventory":0}
    EgressOperation:
      type: string
-      enum: [WAREHOUSES, OUTBOUND_QUERY, OUTBOUND_LOCATE, OUTBOUND_DETAIL, PRODUCT_INVENTORY, CUSTOMERS, PRODUCTS]
+      enum: [WAREHOUSES, OUTBOUND_QUERY, OUTBOUND_LOCATE, OUTBOUND_DETAIL, PRODUCT_INVENTORY, CUSTOMERS, PRODUCTS, INVENTORY_FLOW]
    EgressMetadata:
      type: object
      required: [attemptId, actualHttpAttempts, accountPaused]
@@ -311,6 +311,38 @@ class LingxingWmsClientTest {
        response.set(page("current", "size", 1, 1, 1, 1, ORDER).replace("\"code\":200", "\"code\":200,\"success\":" + success));
        assertThrows(LingxingWmsException.class, () -> client().queryPage(query(1, 1), context()));
    }
+    private InventoryFlowRequest flow() {return new InventoryFlowRequest("W-DEMO","C-DEMO","SKU-DEMO",
+        LocalDateTime.of(2026,9,14,8,0),LocalDateTime.of(2026,10,14,8,0));}
+    @Test void flowUsesExactScopeSinglePageAndPreservesSignedChangeAndLargeId() throws Exception {
+        response.set(page("current","size",1,1000,1,1,"""
+            {"id":9223372036854775806,"customerCode":"C-DEMO","whCode":"W-DEMO","productSku":"SKU-DEMO",
+             "productQuality":1,"bookQty":-3,"bookChangQty":"-3","closeQty":17,"createTime":"2026-10-13 12:00:00"}
+            """));
+        var result=client().queryInventoryFlow(flow(),context());
+        assertThat(path.get()).isEqualTo("/openapi/v2/stock/stockCellBook/page");
+        var sent=mapper.readTree(body.get()).path("data");
+        assertThat(sent.path("barcodeAccurate").asText()).isEqualTo("SKU-DEMO");
+        assertThat(sent.path("size").asInt()).isEqualTo(1000);assertThat(sent.has("productQuality")).isFalse();
+        assertThat(result.records().get(0).id()).isEqualTo("9223372036854775806");
+        assertThat(result.records().get(0).bookQty()).isEqualByComparingTo("-3");
+        assertThat(result.records().get(0).bookChangQty()).isEqualTo("-3");assertThat(calls.get()).isEqualTo(1);
+        assertThat(completion.get().operation()).isEqualTo(Operation.INVENTORY_FLOW);
+    }
+    @Test void flowPermissionDenialIsScopedAndDoesNotStopOrdinaryQueries() {
+        var c=client();response.set("{\"code\":11008,\"msg\":\"private\"}");
+        var error=assertThrows(LingxingWmsException.class,()->c.queryInventoryFlow(flow(),context()));
+        assertThat(error.code()).isEqualTo("LINGXING_WMS_FLOW_PERMISSION_DENIED");
+        assertThat(completion.get().accountPaused()).isFalse();
+        response.set(page("current","size",1,1,1,1,ORDER));
+        assertThat(c.queryPage(query(1,1),context()).records()).hasSize(1);assertThat(calls.get()).isEqualTo(2);
+    }
+    @Test void flowRejectsUnboundedWindowBeforeSendingAndWrongSkuAfterResponse() {
+        var q=flow();var c=client();
+        assertThrows(LingxingWmsException.class,()->c.queryInventoryFlow(new InventoryFlowRequest(q.warehouseCode(),q.customerCode(),q.sku(),q.startTime(),q.startTime().plusDays(32)),context()));
+        assertThat(calls.get()).isZero();
+        response.set(page("current","size",1,1000,1,1,"{\"whCode\":\"W-DEMO\",\"customerCode\":\"C-DEMO\",\"productSku\":\"OTHER\"}"));
+        contract(()->c.queryInventoryFlow(q,context()),"FLOW_IDENTITY_MISMATCH");
+    }
    private LingxingWmsClient client() { return client(CollectionMode.BUSINESS, true, true); }
    private LingxingWmsClient client(CollectionMode mode, boolean grant, boolean budget) {
        var clock = new LingxingEgressGateTest.MutableClock(NOW);
@@ -33,6 +33,7 @@ import java.util.function.Function;
 public class RestLingxingAnalyticsClient implements LingxingAnalyticsClient {
    private static final String PREFIX = "/internal/integrations/lingxing/wms";
    private static final Set<String> PROVIDER_ERROR_CODES = Set.of(
+            "LINGXING_WMS_FLOW_PERMISSION_DENIED",
            "LINGXING_WMS_AUTHORIZATION_REJECTED", "LINGXING_WMS_SIGNATURE_REJECTED",
            "LINGXING_WMS_SOURCE_REJECTED", "LINGXING_WMS_INVALID_RESPONSE", "LINGXING_WMS_RATE_LIMITED",
            "LINGXING_WMS_HTTP_RATE_LIMITED", "LINGXING_WMS_SOURCE_RATE_LIMITED",
@@ -122,6 +123,13 @@ public class RestLingxingAnalyticsClient implements LingxingAnalyticsClient {
                quantity(n,"wmsLength"),quantity(n,"wmsWidth"),quantity(n,"wmsHeight"),optional(n,"wmsSizeUnit")));
    }

+    public JsonNode inventoryFlow(SelfopRequestContext context,String warehouse,String customer,String sku,
+        LocalDateTime start,LocalDateTime end) {
+        gate(context);argument(warehouse,64);argument(customer,64);argument(sku,128);
+        return post(context,"/inventory-flow/query",Map.of("warehouseCode",warehouse,"customerCode",customer,"sku",sku,
+            "startTime",DateTimeFormatter.ISO_LOCAL_DATE_TIME.format(start),"endTime",DateTimeFormatter.ISO_LOCAL_DATE_TIME.format(end)));
+    }
+
    private Summary summary(String account, JsonNode n) {
        List<SelfopPayloads.Item> items = null;
        JsonNode products = n.get("productList");
@@ -160,9 +168,10 @@ public class RestLingxingAnalyticsClient implements LingxingAnalyticsClient {
            case "/product-inventory/query" -> Operation.PRODUCT_INVENTORY;
            case "/customers/query" -> Operation.CUSTOMERS;
            case "/products/query" -> Operation.PRODUCTS;
+            case "/inventory-flow/query" -> Operation.INVENTORY_FLOW;
            default -> throw failure("EGRESS_OPERATION_INVALID");
        };
-        int size = fields.containsKey("size") ? ((Number)fields.get("size")).intValue() : 0;
+        int size = op==Operation.INVENTORY_FLOW?1000:fields.containsKey("size") ? ((Number)fields.get("size")).intValue() : 0;
        String warehouse=(String)fields.get("warehouseCode");
        String key=op==Operation.OUTBOUND_DETAIL ? SelfopSupport.json(List.of(fields.get("customerCode"),fields.get("deliveryNo"))) : null;
        var permission = egress.reserve(context,op,size,warehouse,key,SelfopSupport.hash(fields));
@@ -0,0 +1,23 @@
+package com.baiyue.lastleg.selfoperated.analytics.controller;
+import com.baiyue.common.core.api.ApiResponse;
+import com.baiyue.common.security.AuthenticatedUser;
+import com.baiyue.lastleg.selfoperated.analytics.sync.InventoryFlowProbeService;
+import org.springframework.web.bind.annotation.*;
+import org.springframework.security.access.prepost.PreAuthorize;
+import org.springframework.security.core.annotation.AuthenticationPrincipal;
+import org.springframework.web.server.ResponseStatusException;
+import org.springframework.http.HttpStatus;
+@RestController
+@RequestMapping("/api/platform/v1/last-leg/self-operated-analytics/inventory-flow-probe")
+public class PlatformInventoryFlowProbeController {
+    private final InventoryFlowProbeService service;
+    public PlatformInventoryFlowProbeController(InventoryFlowProbeService service){this.service=service;}
+    @PostMapping
+    @PreAuthorize("hasAuthority('platform:self-operated-analytics:sync')")
+    public ApiResponse<InventoryFlowProbeService.Result> probe(@AuthenticationPrincipal AuthenticatedUser user,
+        @RequestBody InventoryFlowProbeService.Request request) {
+        if(user==null||!Long.valueOf(0).equals(user.tenantId())||!"PLATFORM".equals(user.appCode())||user.impersonated()
+            ||!user.permissions().contains("platform:self-operated-analytics:sync"))throw new ResponseStatusException(HttpStatus.FORBIDDEN);
+        return ApiResponse.success(service.probe(request,user.username()));
+    }
+}
@@ -15,7 +15,7 @@ public interface SelfopCoordinationMapper {
    SelfopSyncTask lockTask(long id);
    @Select("SELECT * FROM ll_selfop_sync_run WHERE id=#{id} FOR UPDATE")
    SelfopSyncRun lockRun(long id);
-    @Select("SELECT * FROM ll_selfop_sync_task WHERE execution_purpose=#{purpose} AND status IN ('READY','RETRY_WAIT','RUNNING') AND (execution_purpose='BUSINESS' OR (status='READY' AND validation_root_task_id=id)) AND (next_retry_at_utc IS NULL OR next_retry_at_utc<=#{now}) ORDER BY CASE WHEN task_type IN ('RECHECK','REBUILD') THEN 1 WHEN task_role='WAREHOUSE_SYNC' THEN 0 WHEN task_type='BACKFILL' THEN 3 ELSE 2 END,id LIMIT #{limit}")
+    @Select("SELECT * FROM ll_selfop_sync_task WHERE execution_purpose=#{purpose} AND task_type<>'INVENTORY_FLOW_PROBE' AND status IN ('READY','RETRY_WAIT','RUNNING') AND (execution_purpose='BUSINESS' OR (status='READY' AND validation_root_task_id=id)) AND (next_retry_at_utc IS NULL OR next_retry_at_utc<=#{now}) ORDER BY CASE WHEN task_type IN ('RECHECK','REBUILD') THEN 1 WHEN task_role='WAREHOUSE_SYNC' THEN 0 WHEN task_type='BACKFILL' THEN 3 ELSE 2 END,id LIMIT #{limit}")
    List<SelfopSyncTask> due(@Param("now") LocalDateTime now, @Param("limit") int limit, @Param("purpose") String purpose);
    // Started inventory work may resume across midnight; unstarted historical slots remain excluded.
    // Publication independently requires the planned day and observation day to match for daily samples.
@@ -7,7 +7,7 @@ public final class SelfopEgressDtos {
  boolean continuousEnabled,String sourceAccountFingerprint,boolean grantPresent,String grantFingerprint,
  long intervalMs,int permitMaxSeconds,boolean executionEnabled,int pendingMaxSeconds,
  boolean exclusiveHandoffVerified,boolean priorEgressClosed,boolean inflightDrained,boolean transferredBudgetVerified){}
- public enum Operation {WAREHOUSES,OUTBOUND_QUERY,OUTBOUND_LOCATE,OUTBOUND_DETAIL,PRODUCT_INVENTORY,CUSTOMERS,PRODUCTS}
+ public enum Operation {WAREHOUSES,OUTBOUND_QUERY,OUTBOUND_LOCATE,OUTBOUND_DETAIL,PRODUCT_INVENTORY,CUSTOMERS,PRODUCTS,INVENTORY_FLOW}
 public record EgressContext(String sourceAccount,String executionPurpose,String rootTaskId,String sessionId,String attemptId,String exclusiveGrantId){}
 @com.fasterxml.jackson.annotation.JsonInclude(com.fasterxml.jackson.annotation.JsonInclude.Include.ALWAYS)
 public record AuthorizationDenial(String errorCode,boolean sampleDone,Instant waitUntil){}
@@ -39,6 +39,11 @@ public class SelfopEgressService {
  SelfopBusinessBudget business=hasBusinessSession(root)?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);
+  boolean flowOperation=operation==Operation.INVENTORY_FLOW;
+  if(flowOperation!="INVENTORY_FLOW_PROBE".equals(task.getTaskType())
+      ||flowOperation&&(!"test".equals(config.getEnvironment())||!"BUSINESS".equals(task.getExecutionPurpose())
+          ||pageSize!=1000||!config.getWarehouses().containsKey(warehouseCode)
+          ||business==null||business.http!=0||!Objects.equals(task.getRequestHash(),requestHash)))throw reject("EGRESS_FLOW_PROBE_INVALID");
  boolean masterOperation=operation==Operation.CUSTOMERS||operation==Operation.PRODUCTS;
  if(masterOperation!=("MASTER_DATA".equals(task.getTaskType()))
      ||masterOperation&&(!"BUSINESS".equals(task.getExecutionPurpose())||pageSize<1||pageSize>(operation==Operation.CUSTOMERS?1000:100)))throw reject("EGRESS_OPERATION_TASK_MISMATCH");
@@ -0,0 +1,103 @@
+package com.baiyue.lastleg.selfoperated.analytics.sync;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.toolkit.IdWorker;
+import com.baiyue.lastleg.selfoperated.analytics.client.RestLingxingAnalyticsClient;
+import com.baiyue.lastleg.selfoperated.analytics.mapper.*;
+import com.baiyue.lastleg.selfoperated.analytics.model.*;
+import com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.support.TransactionTemplate;
+import java.time.LocalDateTime;
+import java.util.*;
+import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*;
+
+/** TEST-only, one audited request. No schedule, retry, publication or historical inventory mutation. */
+@Service
+public class InventoryFlowProbeService {
+    private final SelfopAnalyticsProperties config;
+    private final SelfopSyncTaskMapper tasks;
+    private final SelfopSyncRunMapper runs;
+    private final SelfopOutboundDiscoveryScopeMapper accounts;
+    private final SelfopLeaseService leases;
+    private final RestLingxingAnalyticsClient client;
+    private final TransactionTemplate tx;
+    public InventoryFlowProbeService(SelfopAnalyticsProperties config,SelfopSyncTaskMapper tasks,SelfopSyncRunMapper runs,
+        SelfopOutboundDiscoveryScopeMapper accounts,SelfopLeaseService leases,RestLingxingAnalyticsClient client,TransactionTemplate tx) {
+        this.config=config;this.tasks=tasks;this.runs=runs;this.accounts=accounts;this.leases=leases;this.client=client;this.tx=tx;
+    }
+    public record Request(String requestKey,String warehouseCode,String customerCode,String sku,LocalDateTime startTime,LocalDateTime endTime) {
+        public Map<String,Object> providerRequest(){return Map.of("warehouseCode",warehouseCode,"customerCode",customerCode,"sku",sku,
+            "startTime",java.time.format.DateTimeFormatter.ISO_LOCAL_DATE_TIME.format(startTime),
+            "endTime",java.time.format.DateTimeFormatter.ISO_LOCAL_DATE_TIME.format(endTime));}
+    }
+    public record Result(String taskNo,boolean complete,JsonNode page) {}
+    static void validate(Request q,LocalDateTime now) {
+        require(q!=null,"SELFOP_FLOW_REQUEST_REQUIRED");identity(q.requestKey(),80);identity(q.warehouseCode(),64);
+        identity(q.customerCode(),64);identity(q.sku(),128);
+        require(q.startTime()!=null&&q.endTime()!=null&&q.startTime().isBefore(q.endTime())
+            &&!q.endTime().isAfter(q.startTime().plusDays(31))&&!q.endTime().isAfter(now),"SELFOP_FLOW_WINDOW_INVALID");
+    }
+    public Result probe(Request q,String actor) {
+        validate(q,utcNow().plusHours(8));
+        require("test".equals(config.getEnvironment())&&"BUSINESS".equals(config.getCollectionMode())
+            &&config.isExecutionEnabled()&&config.getWarehouses().containsKey(q.warehouseCode()),"SELFOP_FLOW_PROBE_DISABLED");
+        String key="flow-probe:"+q.requestKey(),digest=hash(q.providerRequest());
+        var account=accounts.selectOne(new QueryWrapper<SelfopOutboundDiscoveryScope>().eq("source_account",config.getSourceAccount()));
+        require(account!=null,"SELFOP_SCOPE_MISSING");
+        var lease=leases.acquireDiscovery(account.getId(),"flow-probe-"+UUID.randomUUID());
+        try {
+            var existing=tasks.selectOne(new QueryWrapper<SelfopSyncTask>().eq("source_account",config.getSourceAccount()).eq("request_key",key));
+            if(existing!=null) {
+                require(digest.equals(existing.getRequestHash()),"SELFOP_IDEMPOTENCY_CONFLICT");
+                require("SAMPLE_DONE".equals(existing.getStatus())&&existing.getRoutingManifestJson()!=null,"SELFOP_FLOW_PREVIOUS_ATTEMPT_REQUIRES_REVIEW");
+                return read(existing.getRoutingManifestJson(),Result.class);
+            }
+            var now=utcNow();var task=new SelfopSyncTask();var run=new SelfopSyncRun();
+            task.setId(IdWorker.getId());task.setTaskNo("LXFLOW-"+UUID.randomUUID().toString().replace("-","").toUpperCase(Locale.ROOT));
+            task.setSourceAccount(config.getSourceAccount());task.setExecutionPurpose("BUSINESS");task.setTaskRole("ACCOUNT_DISCOVERY");
+            task.setDiscoveryScopeId(account.getId());task.setTaskType("INVENTORY_FLOW_PROBE");task.setStatus("RUNNING");
+            task.setApplyState("NOT_APPLICABLE");task.setAttemptCount(1);task.setRequestKey(key);task.setRequestHash(digest);
+            task.setScopeHash(digest);task.setQueryScopeJson(json(q.providerRequest()));task.setBusinessRootTaskId(task.getId());
+            task.setBusinessBudgetJson(SelfopBusinessBudget.freeze(now));task.setRequestedBy(actor);task.setCreatedAt(now);task.setUpdatedAt(now);
+            run.setId(IdWorker.getId());run.setTaskId(task.getId());run.setSourceAccount(config.getSourceAccount());run.setExecutionPurpose("BUSINESS");
+            run.setRunRole("ACCOUNT_DISCOVERY");run.setAttemptNo(1);run.setScopeHash(digest);run.setScopeSnapshotJson(task.getQueryScopeJson());
+            run.setFenceToken(lease.fence());run.setStatus("FETCHING");run.setLastPhase("FETCHING");run.setFetchStartedAtUtc(now);
+            run.setCoverageStatus("UNKNOWN");run.setPageCount(0);run.setUniqueObjectCount(0L);run.setCreatedAt(now);run.setUpdatedAt(now);
+            tx.executeWithoutResult(s->{leases.verify(lease);tasks.insert(task);runs.insert(run);});
+            try {
+                JsonNode page=client.inventoryFlow(SelfopRequestContext.forTask(task,run,config),q.warehouseCode(),q.customerCode(),q.sku(),q.startTime(),q.endTime());
+                boolean complete=validatePage(page);
+                var result=new Result(task.getTaskNo(),complete,page);
+                tx.executeWithoutResult(s->{
+                    leases.verify(lease);var done=utcNow();
+                    task.setStatus("SAMPLE_DONE");task.setRoutingManifestJson(json(result));task.setUpdatedAt(done);
+                    run.setStatus("SAMPLE_DONE");run.setLastPhase("INVENTORY_FLOW_PROBE");run.setPageCount(1);
+                    run.setSourceTotalCount(page.path("total").asLong());run.setUniqueObjectCount((long)page.path("records").size());
+                    run.setCoverageStatus(complete?"COMPLETE":"PARTIAL");run.setFetchFinishedAtUtc(done);run.setFinishedAtUtc(done);run.setUpdatedAt(done);
+                    tasks.updateById(task);runs.updateById(run);
+                });
+                return result;
+            } catch(RuntimeException failure) {
+                String error=failure instanceof com.baiyue.lastleg.selfoperated.analytics.client.AnalyticsClientException c?c.code():
+                    failure instanceof com.baiyue.lastleg.selfoperated.analytics.request.EgressRejectedException e?e.code():"FLOW_PROBE_FAILED";
+                tx.executeWithoutResult(s->{leases.verify(lease);task.setStatus("DEAD");task.setUpdatedAt(utcNow());
+                    run.setStatus("FAILED");run.setErrorCode(error);run.setErrorMessage("库存流水诊断失败；不自动重试");run.setFinishedAtUtc(utcNow());run.setUpdatedAt(utcNow());
+                    tasks.updateById(task);runs.updateById(run);});
+                throw failure;
+            }
+        } finally {leases.release(lease);}
+    }
+    static boolean validatePage(JsonNode page) {
+        require(page!=null&&page.path("records").isArray()&&page.path("current").asInt()==1&&page.path("size").asInt()==1000
+            &&page.path("total").asLong(-1)>=0,"SELFOP_FLOW_PAGE_INVALID");
+        long total=page.path("total").asLong();
+        require(page.path("pages").asLong(-1)==(total==0?0:(total-1)/1000+1)
+            &&page.path("records").size()==Math.min(1000,total),"SELFOP_FLOW_PAGE_INVALID");
+        var ids=new HashSet<String>();
+        for(var row:page.path("records"))require(row.path("id").isTextual()&&!row.path("id").asText().isBlank()
+            &&ids.add(row.path("id").asText()),"SELFOP_FLOW_DUPLICATE_ID");
+        return total<=1000;
+    }
+}
@@ -21,6 +21,34 @@ info:
    重试沿既有POST sync-tasks，action=RETRY/requestKey/warehouseCode/taskNo/reason，使用独立sync权限并最终复核资格。
 security: [{bearerAuth: []}]
 paths:
+  /api/platform/v1/last-leg/self-operated-analytics/inventory-flow-probe:
+    post:
+      summary: TEST单SKU库存流水有界诊断
+      description: >-
+        仅TEST、BUSINESS、执行开启及配置仓库；PLATFORM、tenantId=0、非代登录、独立platform:self-operated-analytics:sync权限。
+        每个requestKey最多一次供应商请求、第一页最多1000条、时间范围最多31天且不得未来。
+        复用现有账号lease/fence、1200ms间隔、permit和消费记录。占用时拒绝，由操作者另行发起，不自动重试。
+        同键成功重放读取保存结果；同键异参拒绝；失败或结果未知不得重发。任务类型INVENTORY_FLOW_PROBE，成功SAMPLE_DONE，失败DEAD。
+        complete仅证明本次分页记录已全部返回，不证明历史流水无缺失或库存可还原。只保存诊断结果，不发布销量、库存或日样本。
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [requestKey, warehouseCode, customerCode, sku, startTime, endTime]
+              properties:
+                requestKey: {type: string, minLength: 1, maxLength: 80}
+                warehouseCode: {type: string, minLength: 1, maxLength: 64}
+                customerCode: {type: string, minLength: 1, maxLength: 64}
+                sku: {type: string, minLength: 1, maxLength: 128}
+                startTime: {type: string, description: 北京时间本地日期时间}
+                endTime: {type: string, description: 北京时间本地日期时间}
+      responses:
+        '200':
+          description: ApiResponse；成功data包含taskNo、complete与page（records、total、current、size、pages、observedAt、egress）。流水字段定义见Integration库存流水合同。业务失败保留错误码，不返回伪成功。
+        '403':
+          description: 非平台或缺少采集权限
  /api/platform/v1/last-leg/self-operated-analytics/owner-directory:
    get:
      summary: 完整资料批次中的客户名称与编码
@@ -0,0 +1,22 @@
+package com.baiyue.lastleg.selfoperated.analytics.controller;
+import com.baiyue.common.security.AuthenticatedUser;
+import com.baiyue.lastleg.selfoperated.analytics.sync.InventoryFlowProbeService;
+import org.junit.jupiter.api.Test;
+import java.util.Set;
+import org.springframework.web.server.ResponseStatusException;
+import org.springframework.security.access.prepost.PreAuthorize;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+class PlatformInventoryFlowProbeControllerTest {
+ @Test void probeRequiresRealPlatformSyncPermission() throws Exception {
+  var service=mock(InventoryFlowProbeService.class);var c=new PlatformInventoryFlowProbeController(service);
+  var reader=new AuthenticatedUser(1L,0L,0L,1L,"reader","PLATFORM","WEB",Set.of("platform:self-operated-analytics:list"));
+  var customer=new AuthenticatedUser(1L,1L,1L,1L,"customer","CLIENT","WEB",Set.of("platform:self-operated-analytics:sync"));
+  assertThrows(ResponseStatusException.class,()->c.probe(null,null));assertThrows(ResponseStatusException.class,()->c.probe(reader,null));
+  assertThrows(ResponseStatusException.class,()->c.probe(customer,null));verifyNoInteractions(service);
+  var writer=new AuthenticatedUser(1L,0L,0L,1L,"operator","PLATFORM","WEB",Set.of("platform:self-operated-analytics:sync"));
+  c.probe(writer,null);verify(service).probe(null,"operator");
+  assertEquals("hasAuthority('platform:self-operated-analytics:sync')",PlatformInventoryFlowProbeController.class
+   .getMethod("probe",AuthenticatedUser.class,InventoryFlowProbeService.Request.class).getAnnotation(PreAuthorize.class).value());
+ }
+}
@@ -44,6 +44,21 @@ class SelfopEgressServiceTest {
  success(c,Operation.CUSTOMERS);success(reserve(Operation.PRODUCTS,100,null,null),Operation.PRODUCTS);
  assertThat(budget().confirmedHttp).isEqualTo(2);
 }
+ @Test void flowProbeIsTestOnlyBoundToFrozenHashAndExactlyOneAttempt(){
+  testBusiness(5,300);task.setTaskType("INVENTORY_FLOW_PROBE");task.setRequestHash(INTERNAL);
+  assertThatThrownBy(()->reserve(Operation.OUTBOUND_QUERY,100,null,null)).hasMessage("EGRESS_FLOW_PROBE_INVALID");
+  assertThatThrownBy(()->reserve(Operation.INVENTORY_FLOW,1000,"OTHER",null)).hasMessage("EGRESS_FLOW_PROBE_INVALID");
+  assertThatThrownBy(()->service.reserve(context(),Operation.INVENTORY_FLOW,1000,"A",null,PROVIDER)).hasMessage("EGRESS_FLOW_PROBE_INVALID");
+  var c=reserve(Operation.INVENTORY_FLOW,1000,"A",null);success(c,Operation.INVENTORY_FLOW);
+  assertThatThrownBy(()->reserve(Operation.INVENTORY_FLOW,1000,"A",null)).hasMessage("EGRESS_FLOW_PROBE_INVALID");
+  assertThat(businessBudget().http).isEqualTo(1);assertThat(budget().confirmedHttp).isEqualTo(1);
+ }
+ @Test void flowPermissionFailureDoesNotPauseExistingSalesAccount(){
+  testBusiness(5,300);task.setTaskType("INVENTORY_FLOW_PROBE");task.setRequestHash(INTERNAL);
+  var c=reserve(Operation.INVENTORY_FLOW,1000,"A",null);service.authorize(new AuthorizationRequest(c,Operation.INVENTORY_FLOW,PROVIDER));
+  service.complete(new CompletionRequest(c,Operation.INVENTORY_FLOW,PROVIDER,1,"LINGXING_WMS_FLOW_PERMISSION_DENIED",200,11008,false,null));
+  assertThat(account.getApiPauseReason()).isNull();assertThat(budget().confirmedHttp).isEqualTo(1);
+ }
 @Test void ordinaryTasksCannotUseMasterDataOperations(){
  business();assertThatThrownBy(()->reserve(Operation.CUSTOMERS,10,null,null)).hasMessage("EGRESS_OPERATION_TASK_MISMATCH");
  assertThatThrownBy(()->reserve(Operation.PRODUCTS,10,null,null)).hasMessage("EGRESS_OPERATION_TASK_MISMATCH");
@@ -0,0 +1,55 @@
+package com.baiyue.lastleg.selfoperated.analytics.sync;
+import org.junit.jupiter.api.Test;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.time.LocalDateTime;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+import static org.mockito.ArgumentMatchers.*;
+import com.baiyue.lastleg.selfoperated.analytics.model.*;
+import com.baiyue.lastleg.selfoperated.analytics.mapper.*;
+import com.baiyue.lastleg.selfoperated.analytics.client.RestLingxingAnalyticsClient;
+import org.springframework.transaction.support.TransactionTemplate;
+class InventoryFlowProbeServiceTest {
+ @Test void successfulProbeIsAuditedReplayedAndNeverPublished() throws Exception {
+  var cfg=new SelfopAnalyticsProperties();cfg.setEnvironment("test");cfg.setCollectionMode("BUSINESS");cfg.setSourceAccount("demo");
+  cfg.getWarehouses().put("W",new SelfopAnalyticsProperties.Warehouse());
+  var tasks=mock(SelfopSyncTaskMapper.class);var runs=mock(SelfopSyncRunMapper.class);var accounts=mock(SelfopOutboundDiscoveryScopeMapper.class);
+  var leases=mock(SelfopLeaseService.class);var client=mock(RestLingxingAnalyticsClient.class);var tx=mock(TransactionTemplate.class);
+  doAnswer(i->{((java.util.function.Consumer<org.springframework.transaction.TransactionStatus>)i.getArgument(0)).accept(null);return null;}).when(tx).executeWithoutResult(any());
+  var a=new SelfopOutboundDiscoveryScope();a.setId(1L);when(accounts.selectOne(any())).thenReturn(a);
+  var lease=new SelfopLeaseService.Lease(1L,"probe",2L,true);when(leases.acquireDiscovery(eq(1L),anyString())).thenReturn(lease);
+  var saved=new java.util.concurrent.atomic.AtomicReference<SelfopSyncTask>();
+  when(tasks.selectOne(any())).thenAnswer(i->saved.get());when(tasks.insert(any(SelfopSyncTask.class))).thenAnswer(i->{saved.set(i.getArgument(0));return 1;});
+  var page=new ObjectMapper().readTree("{\"current\":1,\"size\":1000,\"pages\":1,\"total\":1,\"records\":[{\"id\":\"1\"}]}");
+  when(client.inventoryFlow(any(),anyString(),anyString(),anyString(),any(),any())).thenReturn(page);
+  var s=new InventoryFlowProbeService(cfg,tasks,runs,accounts,leases,client,tx);
+  var end=LocalDateTime.of(2026,1,1,0,0);var q=new InventoryFlowProbeService.Request("one","W","C","S",end.minusDays(30),end);
+  var first=s.probe(q,"operator");assertTrue(first.complete());assertEquals(first,s.probe(q,"operator"));
+  assertEquals("SAMPLE_DONE",saved.get().getStatus());assertNull(saved.get().getResultPublishedRunId());
+  var capture=org.mockito.ArgumentCaptor.forClass(SelfopSyncRun.class);verify(runs).insert(capture.capture());
+  assertEquals("SAMPLE_DONE",capture.getValue().getStatus());assertNull(capture.getValue().getPublishedAtUtc());assertNull(capture.getValue().getInventoryDailySampleDate());
+  verify(client,times(1)).inventoryFlow(any(),anyString(),anyString(),anyString(),any(),any());verify(leases,times(2)).release(lease);
+  assertThrows(IllegalStateException.class,()->s.probe(new InventoryFlowProbeService.Request("one","W","C","OTHER",q.startTime(),q.endTime()),"operator"));
+  verify(client,times(1)).inventoryFlow(any(),anyString(),anyString(),anyString(),any(),any());
+  saved.set(null);when(client.inventoryFlow(any(),anyString(),anyString(),anyString(),any(),any())).thenThrow(new IllegalStateException("PROVIDER_FAILURE"));
+  assertThrows(IllegalStateException.class,()->s.probe(q,"operator"));assertEquals("DEAD",saved.get().getStatus());
+  assertThrows(IllegalStateException.class,()->s.probe(q,"operator"));
+  verify(client,times(2)).inventoryFlow(any(),anyString(),anyString(),anyString(),any(),any());
+ }
+ @Test void refusesFutureAndOversizedWindow() {
+  var now=LocalDateTime.of(2026,10,15,12,0);
+  InventoryFlowProbeService.validate(new InventoryFlowProbeService.Request("one","W","C","S",now.minusDays(30),now),now);
+  assertThrows(IllegalStateException.class,()->InventoryFlowProbeService.validate(new InventoryFlowProbeService.Request("one","W","C","S",now.minusDays(32),now),now));
+  assertThrows(IllegalStateException.class,()->InventoryFlowProbeService.validate(new InventoryFlowProbeService.Request("one","W","C","S",now,now.plusSeconds(1)),now));
+ }
+ @Test void refusesPartialPageAndDuplicateEvidence() throws Exception {
+  var mapper=new ObjectMapper();
+  assertTrue(InventoryFlowProbeService.validatePage(mapper.readTree("{\"current\":1,\"size\":1000,\"pages\":1,\"total\":1,\"records\":[{\"id\":\"1\"}]}")));
+  assertThrows(IllegalStateException.class,()->InventoryFlowProbeService.validatePage(mapper.readTree("{\"current\":1,\"size\":1000,\"pages\":1,\"total\":2,\"records\":[{\"id\":\"1\"}]}")));
+  assertThrows(IllegalStateException.class,()->InventoryFlowProbeService.validatePage(mapper.readTree("{\"current\":1,\"size\":1000,\"pages\":1,\"total\":2,\"records\":[{\"id\":\"1\"},{\"id\":\"1\"}]}")));
+  var root=mapper.createObjectNode().put("current",1).put("size",1000).put("pages",2).put("total",1001);
+  var rows=root.putArray("records");for(int i=0;i<1000;i++)rows.addObject().put("id",String.valueOf(i));
+  assertFalse(InventoryFlowProbeService.validatePage(root));
+ }
+}
+