fix(selfop): persist bounded next-root quantity diagnostics

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticContract.java

@@ -1,22 +1,16 @@
 package com.baiyue.lastleg.selfoperated.analytics.diagnostic;
-
 import java.math.BigDecimal;
 import java.time.Instant;
 import java.util.List;
-
-/** Diagnostic evidence only; never consumed by sales publication. */
 public final class QuantityDiagnosticContract {
-    private QuantityDiagnosticContract() { }
-    public record Order(String diagnosticSessionId, String collectionSessionId, String rootTaskId,
-                        String sourceAccount, String warehouseCode, String ownerCode, String orderNo,
-                        Observation listObservation, Observation detailObservation, List<Sku> skus) { }
-    public record Observation(String status, String outboundTime, Instant observedAt, String attemptId,
-                              Boolean skuSetComplete) { }
-    public record Sku(String sku, BigDecimal listQty, BigDecimal forecastQty, BigDecimal actualQty,
-                      Boolean listPresent, Boolean detailPresent) { }
-    public enum Comparison {
-        SKU_SET_MISMATCH, QUANTITY_UNKNOWN, FORECAST_ACTUAL_DIFFERENCE, ALL_EQUAL, LIST_QUANTITY_DIFFERENCE
-    }
-    public record Receipt(String outcome, Comparison comparison, String stopReason,
-                          int acceptedOrders, int acceptedSkuRows, long acceptedBytes) { }
+ private QuantityDiagnosticContract() {}
+ public record Arm(String diagnosticSessionId,String sourceAccount,List<String> warehouseCodes,String ownerCode,String excludedRootTaskId) {}
+ public record Status(String sourceAccount,String producerId,boolean attach,boolean drained) {}
+ public record Stop(String diagnosticSessionId,String sourceAccount,String producerId,String reason) {}
+ public record Observation(String status,String outboundTime,Instant observedAt,String attemptId,Boolean skuSetComplete) {}
+ public record Sku(String sku,BigDecimal listQty,BigDecimal forecastQty,BigDecimal actualQty) {}
+ public record Event(String diagnosticSessionId,String sourceAccount,String producerId,String rootTaskId,String collectionSessionId,String operation,String warehouseCode,String ownerCode,String orderNo,String side,Observation observation,List<Sku> skus) {}
+ public record Key(String warehouseCode,String ownerCode,String orderNo,boolean paired) {}
+ public record View(String diagnosticSessionId,String sourceAccount,String rootTaskId,String collectionSessionId,String state,String stopReason,Instant startsAt,Instant expiresAt,Instant bindBefore,List<String> warehouseCodes,String ownerCode,String excludedRootTaskId,int candidates,int pairedOrders,int retainedRows,long retainedBytes,boolean deliveryUncertain,List<Key> selected) {}
+ public record Receipt(String outcome,String comparison,View session) {}
 }

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticController.java

@@ -24,20 +24,35 @@ public class QuantityDiagnosticController {
         this.json = json.copy().enable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
             .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);
     }
-    @PostMapping(value = "/orders", consumes = "application/json")
-    public ApiResponse<QuantityDiagnosticContract.Receipt> accept(
-        @RequestHeader(value = "X-Internal-Service-Token", required = false) String token, @RequestBody byte[] body) {
+    @PostMapping(value="/arm",consumes="application/json")
+    public ApiResponse<QuantityDiagnosticContract.View> arm(@RequestHeader(value="X-Internal-Service-Token",required=false) String token,jakarta.servlet.http.HttpServletRequest request){
+        return run(()->service.arm(read(token,request,QuantityDiagnosticContract.Arm.class)));
+    }
+    @PostMapping(value="/status",consumes="application/json")
+    public ApiResponse<QuantityDiagnosticContract.View> status(@RequestHeader(value="X-Internal-Service-Token",required=false) String token,jakarta.servlet.http.HttpServletRequest request){
+        return run(()->service.status(read(token,request,QuantityDiagnosticContract.Status.class)));
+    }
+    @PostMapping(value="/observe",consumes="application/json")
+    public ApiResponse<QuantityDiagnosticContract.Receipt> observe(@RequestHeader(value="X-Internal-Service-Token",required=false) String token,jakarta.servlet.http.HttpServletRequest request){
+        return run(()->service.observe(read(token,request,QuantityDiagnosticContract.Event.class)));
+    }
+    @PostMapping(value="/stop",consumes="application/json")
+    public ApiResponse<QuantityDiagnosticContract.View> stop(@RequestHeader(value="X-Internal-Service-Token",required=false) String token,jakarta.servlet.http.HttpServletRequest request){
+        return run(()->service.stop(read(token,request,QuantityDiagnosticContract.Stop.class)));
+    }
+    private <T>T read(String token,jakarta.servlet.http.HttpServletRequest request,Class<T> type){
         guard.verify(token);
-        if (body.length > QuantityDiagnosticPolicy.MAX_REQUEST_BYTES)
-            throw QuantityDiagnosticPolicy.invalid("DIAGNOSTIC_REQUEST_TOO_LARGE");
-        QuantityDiagnosticContract.Order order;
-        try { order = json.readValue(body, QuantityDiagnosticContract.Order.class); }
-        catch (IOException e) { throw QuantityDiagnosticPolicy.invalid("DIAGNOSTIC_JSON_INVALID"); }
-        try { return ApiResponse.success(service.accept(order)); }
-        catch (DataAccessException | TransactionException e) {
-            // No SQL, request payload or credentials are returned to the caller.
-            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "DIAGNOSTIC_STORAGE_UNAVAILABLE");
-        }
+        int limit=QuantityDiagnosticPolicy.MAX_REQUEST_BYTES;
+        if(request.getContentLengthLong()>limit)throw QuantityDiagnosticPolicy.invalid("DIAGNOSTIC_REQUEST_TOO_LARGE");
+        try {
+            byte[] body=request.getInputStream().readNBytes(limit+1);
+            if(body.length>limit)throw QuantityDiagnosticPolicy.invalid("DIAGNOSTIC_REQUEST_TOO_LARGE");
+            return json.readValue(body,type);
+        }catch(IOException e){throw QuantityDiagnosticPolicy.invalid("DIAGNOSTIC_JSON_INVALID");}
+    }
+    private <T>ApiResponse<T> run(java.util.function.Supplier<T> action){
+        try{return ApiResponse.success(action.get());}
+        catch(DataAccessException|TransactionException e){throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,"DIAGNOSTIC_STORAGE_UNAVAILABLE");}
     }
     @ExceptionHandler(ResponseStatusException.class)
     public ResponseEntity<ApiResponse<Void>> rejected(ResponseStatusException error) {

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticPolicy.java

@@ -1,89 +1,33 @@
 package com.baiyue.lastleg.selfoperated.analytics.diagnostic;
-
 import static com.baiyue.lastleg.selfoperated.analytics.diagnostic.QuantityDiagnosticContract.*;
 import java.math.BigDecimal;
-import java.time.Duration;
 import java.time.Instant;
-import java.util.Comparator;
-import java.util.HashSet;
-import java.util.List;
+import java.util.*;
 import org.springframework.http.HttpStatus;
 import org.springframework.web.server.ResponseStatusException;
-
 final class QuantityDiagnosticPolicy {
-    static final int MAX_ORDERS = 100, MAX_ROWS = 2000, MAX_ORDER_ROWS = 200;
-    static final int MAX_BYTES = 2 * 1024 * 1024, MAX_REQUEST_BYTES = 256 * 1024;
-    private QuantityDiagnosticPolicy() { }
-    static ResponseStatusException invalid(String code) {
-        return new ResponseStatusException(HttpStatus.BAD_REQUEST, code);
-    }
-    static void text(String value, int maximum) {
-        if (value == null || value.isBlank() || value.length() > maximum
-            || value.codePoints().anyMatch(Character::isISOControl)) throw invalid("DIAGNOSTIC_FIELD_INVALID");
-    }
-    static void settings(QuantityDiagnosticProperties p) {
-        text(p.diagnosticSessionId(), 128); text(p.collectionSessionId(), 128);
-        text(p.rootTaskId(), 128); text(p.sourceAccount(), 64);
-        if (p.startsAt() == null || p.expiresAt() == null || !p.expiresAt().isAfter(p.startsAt())
-            || Duration.between(p.startsAt(), p.expiresAt()).compareTo(Duration.ofMinutes(30)) > 0
-            || p.warehouseCodes() == null || p.warehouseCodes().isEmpty() || p.warehouseCodes().size() > 20)
-            throw invalid("DIAGNOSTIC_CONFIGURATION_INVALID");
-        p.warehouseCodes().forEach(w -> text(w, 64));
-    }
-    static Order normalize(Order r, QuantityDiagnosticProperties p, Instant now) {
-        if (r == null) throw invalid("DIAGNOSTIC_ORDER_REQUIRED");
-        if (!p.diagnosticSessionId().equals(r.diagnosticSessionId())
-            || !p.collectionSessionId().equals(r.collectionSessionId()) || !p.rootTaskId().equals(r.rootTaskId())
-            || !p.sourceAccount().equals(r.sourceAccount()) || !p.warehouseCodes().contains(r.warehouseCode()))
-            throw invalid("DIAGNOSTIC_SCOPE_MISMATCH");
-        text(r.ownerCode(), 128); text(r.orderNo(), 128);
-        observation(r.listObservation(), p, now); observation(r.detailObservation(), p, now);
-        if (r.skus() == null || r.skus().isEmpty() || r.skus().size() > MAX_ORDER_ROWS)
-            throw invalid("DIAGNOSTIC_SKU_LIMIT");
-        var seen = new HashSet<String>();
-        for (Sku s : r.skus()) {
-            if (s == null) throw invalid("DIAGNOSTIC_SKU_REQUIRED");
-            text(s.sku(), 128);
-            if (!seen.add(s.sku())) throw invalid("DIAGNOSTIC_DUPLICATE_SKU");
-            if (s.listPresent() == null || s.detailPresent() == null || (!s.listPresent() && !s.detailPresent()))
-                throw invalid("DIAGNOSTIC_PRESENCE_INVALID");
-            if ((!s.listPresent() && s.listQty() != null)
-                || (!s.detailPresent() && (s.forecastQty() != null || s.actualQty() != null)))
-                throw invalid("DIAGNOSTIC_ABSENT_QUANTITY");
-            quantity(s.listQty()); quantity(s.forecastQty()); quantity(s.actualQty());
-        }
-        List<Sku> skus = r.skus().stream().sorted(Comparator.comparing(Sku::sku))
-            .map(s -> new Sku(s.sku(), decimal(s.listQty()), decimal(s.forecastQty()), decimal(s.actualQty()),
-                s.listPresent(), s.detailPresent())).toList();
-        return new Order(r.diagnosticSessionId(), r.collectionSessionId(), r.rootTaskId(), r.sourceAccount(),
-            r.warehouseCode(), r.ownerCode(), r.orderNo(), r.listObservation(), r.detailObservation(), skus);
-    }
-    private static void observation(Observation o, QuantityDiagnosticProperties p, Instant now) {
-        if (o == null || o.observedAt() == null || o.skuSetComplete() == null)
-            throw invalid("DIAGNOSTIC_OBSERVATION_REQUIRED");
-        text(o.attemptId(), 128);
-        if (o.status() != null) text(o.status(), 32);
-        if (o.outboundTime() != null) text(o.outboundTime(), 32);
-        if (o.observedAt().isBefore(p.startsAt()) || o.observedAt().isAfter(p.expiresAt())
-            || o.observedAt().isAfter(now)) throw invalid("DIAGNOSTIC_OBSERVATION_OUTSIDE_SESSION");
-    }
-    private static BigDecimal decimal(BigDecimal v) { return v == null ? null : v.stripTrailingZeros(); }
-    private static void quantity(BigDecimal v) {
-        if (v != null && (v.signum() < 0 || v.scale() > 6 || v.precision() - v.scale() > 32))
-            throw invalid("DIAGNOSTIC_QUANTITY_INVALID");
-    }
-    static boolean forecastDifference(Order r) {
-        return r.skus().stream().anyMatch(s -> s.forecastQty() != null && s.actualQty() != null
-            && s.forecastQty().compareTo(s.actualQty()) != 0);
-    }
-    static Comparison compare(Order r) {
-        if (!r.listObservation().skuSetComplete() || !r.detailObservation().skuSetComplete()
-            || r.skus().stream().anyMatch(s -> !s.listPresent() || !s.detailPresent()))
-            return Comparison.SKU_SET_MISMATCH;
-        if (r.skus().stream().anyMatch(s -> s.listQty() == null || s.forecastQty() == null || s.actualQty() == null))
-            return Comparison.QUANTITY_UNKNOWN;
-        if (forecastDifference(r)) return Comparison.FORECAST_ACTUAL_DIFFERENCE;
-        return r.skus().stream().allMatch(s -> s.listQty().compareTo(s.actualQty()) == 0)
-            ? Comparison.ALL_EQUAL : Comparison.LIST_QUANTITY_DIFFERENCE;
-    }
+ static final int MAX_ORDERS=100,MAX_ROWS=2000,MAX_BYTES=2097152,MAX_REQUEST_BYTES=262144;
+ private QuantityDiagnosticPolicy() {}
+ static ResponseStatusException invalid(String code){return new ResponseStatusException(HttpStatus.BAD_REQUEST,code);}
+ static void require(boolean value,String code){if(!value)throw invalid(code);}
+ static void text(String s,int max){require(s!=null&&!s.isBlank()&&s.length()<=max&&s.codePoints().noneMatch(Character::isISOControl),"DIAGNOSTIC_FIELD_INVALID");}
+ static Event normalize(Event e,Instant now){
+  require(e!=null,"DIAGNOSTIC_EVENT_REQUIRED");text(e.diagnosticSessionId(),128);text(e.sourceAccount(),64);text(e.rootTaskId(),128);text(e.collectionSessionId(),128);text(e.warehouseCode(),64);text(e.ownerCode(),128);text(e.orderNo(),128);
+  text(e.producerId(),128);require("LIST".equals(e.side())||"DETAIL".equals(e.side()),"DIAGNOSTIC_SIDE_INVALID");require(("LIST".equals(e.side())?"OUTBOUND_QUERY":"OUTBOUND_DETAIL").equals(e.operation()),"DIAGNOSTIC_OPERATION_INVALID");require(e.observation()!=null&&e.observation().observedAt()!=null&&e.observation().skuSetComplete()!=null,"DIAGNOSTIC_OBSERVATION_REQUIRED");text(e.observation().attemptId(),128);
+  if(e.observation().status()!=null)text(e.observation().status(),32);if(e.observation().outboundTime()!=null)text(e.observation().outboundTime(),64);
+  require(!e.observation().observedAt().isAfter(now),"DIAGNOSTIC_FUTURE_OBSERVATION");require(e.skus()!=null&&(!e.skus().isEmpty()||!e.observation().skuSetComplete())&&e.skus().size()<=200,"DIAGNOSTIC_SKU_LIMIT");
+  var seen=new HashSet<String>();var skus=new ArrayList<Sku>();for(var s:e.skus()){require(s!=null,"DIAGNOSTIC_SKU_REQUIRED");text(s.sku(),128);require(seen.add(s.sku()),"DIAGNOSTIC_DUPLICATE_SKU");
+   for(var n:Arrays.asList(s.listQty(),s.forecastQty(),s.actualQty()))if(n!=null)require(n.signum()>=0&&n.scale()<=6&&n.precision()-n.scale()<=32,"DIAGNOSTIC_QUANTITY_INVALID");
+   require("LIST".equals(e.side())?s.forecastQty()==null&&s.actualQty()==null:s.listQty()==null,"DIAGNOSTIC_SIDE_FIELDS");skus.add(new Sku(s.sku(),decimal(s.listQty()),decimal(s.forecastQty()),decimal(s.actualQty())));
+  }skus.sort(Comparator.comparing(Sku::sku));return new Event(e.diagnosticSessionId(),e.sourceAccount(),e.producerId(),e.rootTaskId(),e.collectionSessionId(),e.operation(),e.warehouseCode(),e.ownerCode(),e.orderNo(),e.side(),e.observation(),List.copyOf(skus));
+ }
+ static BigDecimal decimal(BigDecimal n){return n==null?null:n.stripTrailingZeros();}
+ static boolean different(BigDecimal a,BigDecimal b){return a!=null&&b!=null&&a.compareTo(b)!=0;}
+ static String compare(Event list,Event detail){
+  var l=new TreeMap<String,Sku>();var d=new TreeMap<String,Sku>();list.skus().forEach(s->l.put(s.sku(),s));detail.skus().forEach(s->d.put(s.sku(),s));
+  boolean incomplete=!Boolean.TRUE.equals(list.observation().skuSetComplete())||!Boolean.TRUE.equals(detail.observation().skuSetComplete())||!l.keySet().equals(d.keySet());
+  var keys=new HashSet<String>(l.keySet());keys.addAll(d.keySet());boolean difference=false;
+  for(String key:keys){var a=l.get(key);var b=d.get(key);BigDecimal x=a==null?null:a.listQty(),y=b==null?null:b.forecastQty(),z=b==null?null:b.actualQty();difference|=different(x,y)||different(x,z)||different(y,z);incomplete|=x==null||y==null||z==null;}
+  return incomplete?"INCONCLUSIVE":difference?"QUANTITY_DIFFERENCE":"ALL_EQUAL";
+ }
 }

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticProperties.java

@@ -1,11 +1,6 @@
 package com.baiyue.lastleg.selfoperated.analytics.diagnostic;
-
-import java.time.Instant;
 import java.util.List;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.bind.DefaultValue;
-
 @ConfigurationProperties("baiyue.selfop.quantity-diagnostics")
-public record QuantityDiagnosticProperties(@DefaultValue("false") boolean enabled,
-    String diagnosticSessionId, String collectionSessionId, String rootTaskId, String sourceAccount,
-    List<String> warehouseCodes, Instant startsAt, Instant expiresAt) { }
+public record QuantityDiagnosticProperties(@DefaultValue("false") boolean enabled, String sourceAccount, List<String> warehouseCodes) {}

lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/diagnostic/QuantityDiagnosticService.java

@@ -2,124 +2,218 @@ package com.baiyue.lastleg.selfoperated.analytics.diagnostic;
 
 import static com.baiyue.lastleg.selfoperated.analytics.diagnostic.QuantityDiagnosticContract.*;
 import static com.baiyue.lastleg.selfoperated.analytics.diagnostic.QuantityDiagnosticPolicy.*;
-import com.fasterxml.jackson.core.JsonProcessingException;
+import com.baiyue.lastleg.selfoperated.analytics.model.SelfopSyncTask;
+import com.baiyue.lastleg.selfoperated.analytics.model.SelfopCollectionPolicy;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.zaxxer.hikari.HikariDataSource;
+import jakarta.annotation.PreDestroy;
+import javax.sql.DataSource;
 import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
-import java.time.Clock;
-import java.time.Instant;
-import java.util.HexFormat;
-import java.util.List;
+import java.time.*;
+import java.util.*;
+import java.util.concurrent.Semaphore;
+import java.util.function.Supplier;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.BeanPropertyRowMapper;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.web.server.ResponseStatusException;
 
-/** One short transaction per paired order; only the two diagnostic tables are written. */
+/** Serializes diagnostic state only; never locks or updates collection/business rows. */
 @Service
 public class QuantityDiagnosticService {
     private final JdbcTemplate jdbc;
     private final TransactionTemplate tx;
     private final ObjectMapper json;
-    private final QuantityDiagnosticProperties properties;
+    private final QuantityDiagnosticProperties config;
     private final Clock clock;
+    private final Semaphore admission = new Semaphore(1);
+    private HikariDataSource pool;
 
     @Autowired
-    public QuantityDiagnosticService(JdbcTemplate jdbc, PlatformTransactionManager manager,
-                                     ObjectMapper json, QuantityDiagnosticProperties properties) {
-        this(jdbc, manager, json, properties, Clock.systemUTC());
-    }
-    QuantityDiagnosticService(JdbcTemplate jdbc, PlatformTransactionManager manager,
-                              ObjectMapper json, QuantityDiagnosticProperties properties, Clock clock) {
-        this.jdbc = jdbc; this.json = json.copy(); this.properties = properties; this.clock = clock;
-        this.tx = new TransactionTemplate(manager); this.tx.setTimeout(3);
-    }
-    public Receipt accept(Order input) {
-        if (!properties.enabled()) return new Receipt("STOPPED", null, "DISABLED", 0, 0, 0);
-        settings(properties);
-        Order order = normalize(input, properties, clock.instant());
-        String payload = encode(order);
-        int bytes = payload.getBytes(StandardCharsets.UTF_8).length;
-        if (bytes > MAX_REQUEST_BYTES) throw invalid("DIAGNOSTIC_REQUEST_TOO_LARGE");
-        String key = hash(encode(List.of(order.sourceAccount(), order.warehouseCode(), order.ownerCode(), order.orderNo())));
-        String payloadHash = hash(payload);
-        String configuration = hash(encode(List.of("v1", properties.diagnosticSessionId(), properties.collectionSessionId(),
-            properties.rootTaskId(), properties.sourceAccount(), properties.warehouseCodes().stream().sorted().distinct().toList(),
-            properties.startsAt(), properties.expiresAt(), MAX_ORDERS, MAX_ROWS, MAX_BYTES)));
-        return tx.execute(status -> persist(order, key, payload, payloadHash, bytes, configuration));
-    }
-    private Receipt persist(Order order, String key, String payload, String payloadHash, int bytes, String configuration) {
-        String id = order.diagnosticSessionId();
-        // Duplicate insert acquires the same durable session lock, including first-request races.
-        jdbc.update("""
-            INSERT INTO ll_selfop_quantity_diagnostic_session
-            (diagnostic_session_id, configuration_hash, source_account, collection_session_id, root_task_id,
-             starts_at, expires_at, accepted_orders, accepted_sku_rows, accepted_bytes)
-            VALUES (?, ?, ?, ?, ?, ?, ?, 0, 0, 0)
-            ON DUPLICATE KEY UPDATE diagnostic_session_id = diagnostic_session_id
-            """, id, configuration, order.sourceAccount(), order.collectionSessionId(), order.rootTaskId(),
-            properties.startsAt().toString(), properties.expiresAt().toString());
-        Session session = jdbc.queryForObject("""
-            SELECT configuration_hash, accepted_orders, accepted_sku_rows, accepted_bytes, stop_reason
-            FROM ll_selfop_quantity_diagnostic_session WHERE diagnostic_session_id = ? FOR UPDATE
-            """, (rs, n) -> new Session(rs.getString(1), rs.getInt(2), rs.getInt(3), rs.getLong(4), rs.getString(5)), id);
-        if (session == null || !configuration.equals(session.configurationHash()))
-            throw new ResponseStatusException(HttpStatus.CONFLICT, "DIAGNOSTIC_SESSION_CONFIGURATION_CHANGED");
-        var existing = jdbc.query("""
-            SELECT payload_hash, comparison FROM ll_selfop_quantity_diagnostic_order
-            WHERE diagnostic_session_id = ? AND order_key_hash = ?
-            """, (rs, n) -> new Existing(rs.getString(1), Comparison.valueOf(rs.getString(2))), id, key);
-        if (!existing.isEmpty()) {
-            if (!payloadHash.equals(existing.get(0).payloadHash()))
-                throw new ResponseStatusException(HttpStatus.CONFLICT, "DIAGNOSTIC_ORDER_CONFLICT");
-            return session.receipt("DUPLICATE", existing.get(0).comparison(), session.stopReason());
+    public QuantityDiagnosticService(DataSource source, ObjectMapper json, QuantityDiagnosticProperties config) {
+        this.json=json.copy().disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+        this.config=config; this.clock=Clock.systemUTC();
+        // Separate, bounded connection acquisition. Diagnostics cannot exhaust the collection pool.
+        if(config.enabled()) {
+            try {
+                HikariDataSource base=source.unwrap(HikariDataSource.class);
+                pool=new HikariDataSource(); pool.setJdbcUrl(base.getJdbcUrl()); pool.setUsername(base.getUsername());
+                pool.setPassword(base.getPassword()); pool.setDataSourceProperties(base.getDataSourceProperties());
+                pool.setMaximumPoolSize(1); pool.setMinimumIdle(0); pool.setConnectionTimeout(250);
+                pool.setValidationTimeout(250); pool.setInitializationFailTimeout(-1); pool.setPoolName("quantity-diagnostic");
+            } catch(Exception unavailable) { if(pool!=null)pool.close(); pool=null; }
         }
-        if (session.stopReason() != null) return session.receipt("STOPPED", null, session.stopReason());
-        Instant now = clock.instant();
-        if (now.isBefore(properties.startsAt())) return session.receipt("STOPPED", null, "NOT_STARTED");
-        if (!now.isBefore(properties.expiresAt())) return stop(id, session, "EXPIRED");
-        if (session.orders() >= MAX_ORDERS || session.rows() + order.skus().size() > MAX_ROWS
-            || session.bytes() + bytes > MAX_BYTES) return stop(id, session, "LIMIT_REACHED");
-        Comparison comparison = compare(order);
-        jdbc.update("""
-            INSERT INTO ll_selfop_quantity_diagnostic_order
-            (diagnostic_session_id, order_key_hash, source_account, warehouse_code, owner_code, order_no,
-             payload_hash, comparison, sku_rows, payload_bytes, evidence_json, received_at)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-            """, id, key, order.sourceAccount(), order.warehouseCode(), order.ownerCode(), order.orderNo(),
-            payloadHash, comparison.name(), order.skus().size(), bytes, payload, now.toString());
-        int orders = session.orders() + 1, rows = session.rows() + order.skus().size();
-        long totalBytes = session.bytes() + bytes;
-        String reason = forecastDifference(order) ? "FORECAST_ACTUAL_DIFFERENCE"
-            : orders == MAX_ORDERS || rows == MAX_ROWS || totalBytes == MAX_BYTES ? "LIMIT_REACHED" : null;
-        int updated = jdbc.update("""
-            UPDATE ll_selfop_quantity_diagnostic_session
-            SET accepted_orders = ?, accepted_sku_rows = ?, accepted_bytes = ?, stop_reason = ?
-            WHERE diagnostic_session_id = ?
-            """, orders, rows, totalBytes, reason, id);
-        if (updated != 1) throw new IllegalStateException("DIAGNOSTIC_SESSION_UPDATE_FAILED");
-        return new Receipt("ACCEPTED", comparison, reason, orders, rows, totalBytes);
+        jdbc=pool==null?null:new JdbcTemplate(pool);
+        tx=pool==null?null:new TransactionTemplate(new DataSourceTransactionManager(pool));
+        if(tx!=null) { tx.setTimeout(3); jdbc.setQueryTimeout(2); }
     }
-    private Receipt stop(String id, Session session, String reason) {
-        jdbc.update("UPDATE ll_selfop_quantity_diagnostic_session SET stop_reason = ? WHERE diagnostic_session_id = ?", reason, id);
-        return session.receipt("STOPPED", null, reason);
+    QuantityDiagnosticService(JdbcTemplate jdbc,PlatformTransactionManager manager,ObjectMapper json,
+                              QuantityDiagnosticProperties config,Clock clock) {
+        this.jdbc=jdbc;this.json=json.copy().disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+        this.config=config;this.clock=clock;tx=new TransactionTemplate(manager);tx.setTimeout(3);
     }
-    private String encode(Object value) {
-        try { return json.writeValueAsString(value); }
-        catch (JsonProcessingException e) { throw invalid("DIAGNOSTIC_ENCODING_FAILED"); }
+    @PreDestroy public void close(){if(pool!=null)pool.close();}
+    private <T>T guarded(Supplier<T> action){
+        require(config.enabled(),"DIAGNOSTIC_DISABLED");
+        if(tx==null)throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,"DIAGNOSTIC_DATASOURCE_UNAVAILABLE");
+        if(!admission.tryAcquire())throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,"DIAGNOSTIC_BUSY");
+        try{return tx.execute(s->action.get());}finally{admission.release();}
     }
-    private static String hash(String value) {
-        try { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8))); }
-        catch (NoSuchAlgorithmException e) { throw new IllegalStateException(e); }
+    private void account(String account){text(account,64);require(account.equals(config.sourceAccount()),"DIAGNOSTIC_ACCOUNT_REJECTED");}
+    public View arm(Arm request){
+        require(request!=null,"DIAGNOSTIC_ARM_REQUIRED");text(request.diagnosticSessionId(),128);account(request.sourceAccount());
+        text(request.ownerCode(),128);if(request.excludedRootTaskId()!=null)text(request.excludedRootTaskId(),128);
+        require(request.warehouseCodes()!=null&&!request.warehouseCodes().isEmpty()&&request.warehouseCodes().size()<=8,"DIAGNOSTIC_SCOPE_REJECTED");
+        request.warehouseCodes().forEach(w->text(w,64));
+        require(config.warehouseCodes()!=null&&config.warehouseCodes().containsAll(request.warehouseCodes()),"DIAGNOSTIC_SCOPE_REJECTED");
+        Arm scope=new Arm(request.diagnosticSessionId(),request.sourceAccount(),request.warehouseCodes().stream().sorted().distinct().toList(),request.ownerCode(),request.excludedRootTaskId());
+        return guarded(()->{
+            Session existing=load(scope.diagnosticSessionId());
+            if(existing!=null){require(existing.scope.equals(scope),"DIAGNOSTIC_ARM_CONFLICT");advance(existing);save(existing);return view(existing);}
+            // Read-only arm snapshot: unfinished automatic roots that have actually started.
+            var current=jdbc.queryForList("SELECT id FROM ll_selfop_sync_task WHERE source_account=? AND task_role='ACCOUNT_DISCOVERY' AND task_type='BACKFILL' AND execution_purpose='BUSINESS' AND requested_by='SELFOP_CONTINUOUS_PLANNER' AND (status IN ('RUNNING','RETRY_WAIT') OR (status='READY' AND attempt_count>0)) ORDER BY id LIMIT 2",String.class,scope.sourceAccount());
+            require(current.size()<=1,"DIAGNOSTIC_CURRENT_ROOT_AMBIGUOUS");
+            String excluded=current.isEmpty()?null:current.get(0);
+            require(Objects.equals(excluded,scope.excludedRootTaskId()),"DIAGNOSTIC_CURRENT_ROOT_CHANGED");
+            require(excluded==null||eligible(root(excluded),scope.sourceAccount()),"DIAGNOSTIC_EXCLUDED_ROOT_INVALID");
+            var active=jdbc.queryForList("SELECT diagnostic_session_id FROM ll_selfop_quantity_diagnostic_session WHERE source_account=? AND active_slot=1",String.class,scope.sourceAccount());
+            if(!active.isEmpty()){Session s=load(active.get(0));advance(s);save(s);require(terminal(s),"DIAGNOSTIC_ALREADY_ARMED");}
+            Session s=new Session();s.scope=scope;s.armedAt=clock.instant();s.bindBefore=s.armedAt.plusSeconds(21600);s.expiresAt=s.armedAt.plusSeconds(36000);s.state="ARMED";
+            // Account+active_slot unique constraint is the cross-instance arm arbitration point.
+            jdbc.update("INSERT INTO ll_selfop_quantity_diagnostic_session(diagnostic_session_id,configuration_hash,source_account,starts_at,expires_at,active_slot,state_json) VALUES(?,?,?,?,?,1,?)",scope.diagnosticSessionId(),hash(encode(scope)),scope.sourceAccount(),s.armedAt.toString(),s.expiresAt.toString(),encode(s));
+            return view(s);
+        });
     }
-    private record Existing(String payloadHash, Comparison comparison) { }
-    private record Session(String configurationHash, int orders, int rows, long bytes, String stopReason) {
-        Receipt receipt(String outcome, Comparison comparison, String reason) {
-            return new Receipt(outcome, comparison, reason, orders, rows, bytes);
-        }
+    public View status(Status request){
+        require(request!=null,"DIAGNOSTIC_STATUS_REQUIRED");account(request.sourceAccount());text(request.producerId(),128);
+        return guarded(()->{
+            var ids=jdbc.queryForList("SELECT diagnostic_session_id FROM ll_selfop_quantity_diagnostic_session WHERE source_account=? AND state_json IS NOT NULL ORDER BY active_slot DESC, starts_at DESC LIMIT 1",String.class,request.sourceAccount());
+            if(ids.isEmpty())return null;
+            Session s=load(ids.get(0));advance(s);
+            if(!terminal(s)){
+                if(s.producerId==null)s.producerId=request.producerId();
+                else if(!s.producerId.equals(request.producerId())){
+                    require(request.attach(),"DIAGNOSTIC_PRODUCER_FENCED");
+                    // Recovery retains persisted candidates, but cannot claim the lost in-memory tail was complete.
+                    s.deliveryUncertain=true;s.producerId=request.producerId();
+                }
+                if("DRAINING".equals(s.state)&&request.drained())finish(s);
+            }
+            save(s);return view(s);
+        });
+    }
+    public View stop(Stop request){
+        require(request!=null,"DIAGNOSTIC_STOP_REQUIRED");account(request.sourceAccount());text(request.diagnosticSessionId(),128);text(request.producerId(),128);
+        require(Set.of("INCOMPLETE","QUEUE_FULL","OBSERVER_FAILED","SHUTDOWN").contains(request.reason()),"DIAGNOSTIC_STOP_REASON");
+        return guarded(()->{Session s=load(request.diagnosticSessionId());require(s!=null&&s.scope.sourceAccount().equals(request.sourceAccount()),"DIAGNOSTIC_SESSION_UNKNOWN");
+            require(Objects.equals(s.producerId,request.producerId()),"DIAGNOSTIC_PRODUCER_FENCED");
+            if(!terminal(s)){s.deliveryUncertain=true;end(s,"INCOMPLETE");}save(s);return view(s);});
+    }
+    public Receipt observe(Event input){
+        Event e=normalize(input,clock.instant());account(e.sourceAccount());String payload=encode(e);int bytes=size(payload);
+        require(bytes<=MAX_REQUEST_BYTES,"DIAGNOSTIC_REQUEST_TOO_LARGE");
+        return guarded(()->{
+            Session s=load(e.diagnosticSessionId());require(s!=null&&s.scope.sourceAccount().equals(e.sourceAccount()),"DIAGNOSTIC_SESSION_UNKNOWN");
+            String key=hash(encode(List.of(e.sourceAccount(),e.warehouseCode(),e.ownerCode(),e.orderNo())));
+            var completed=jdbc.query("SELECT evidence_json,comparison FROM ll_selfop_quantity_diagnostic_order WHERE diagnostic_session_id=? AND order_key_hash=?",(r,n)->new String[]{r.getString(1),r.getString(2)},e.diagnosticSessionId(),key);
+            if(!completed.isEmpty()){
+                Pair pair=decode(completed.get(0)[0],Pair.class);Event original="LIST".equals(e.side())?pair.list:pair.detail;
+                require(encode(original).equals(payload),"DIAGNOSTIC_OBSERVATION_CONFLICT");
+                return new Receipt("DUPLICATE",completed.get(0)[1],view(s));
+            }
+            var pending=jdbc.query("SELECT evidence_json FROM ll_selfop_quantity_diagnostic_pending WHERE diagnostic_session_id=? AND order_key_hash=?",(r,n)->decode(r.getString(1),Event.class),e.diagnosticSessionId(),key);
+            if("LIST".equals(e.side())&&!pending.isEmpty()){
+                require(encode(pending.get(0)).equals(payload),"DIAGNOSTIC_OBSERVATION_CONFLICT");return new Receipt("DUPLICATE",null,view(s));
+            }
+            // Session lock is the receiver acceptance boundary. Integration enqueue is not acceptance.
+            advance(s);
+            if(terminal(s)){save(s);return new Receipt("STOPPED",null,view(s));}
+            require(Objects.equals(s.producerId,e.producerId()),"DIAGNOSTIC_PRODUCER_FENCED");
+            require(!e.observation().observedAt().isBefore(s.armedAt),"DIAGNOSTIC_OLD_OBSERVATION");
+            require(s.scope.warehouseCodes().contains(e.warehouseCode())&&("*".equals(s.scope.ownerCode())||s.scope.ownerCode().equals(e.ownerCode())),"DIAGNOSTIC_SCOPE_REJECTED");
+            require(("selfop-"+e.rootTaskId()).equals(e.collectionSessionId()),"DIAGNOSTIC_COLLECTION_SESSION_REJECTED");
+            SelfopSyncTask task=root(e.rootTaskId());require(eligible(task,e.sourceAccount()),"DIAGNOSTIC_ROOT_REJECTED");
+            if(s.rootTaskId==null){
+                if(!"LIST".equals(e.side())||e.rootTaskId().equals(s.scope.excludedRootTaskId())||!clock.instant().isBefore(s.bindBefore)||!e.observation().observedAt().isBefore(s.bindBefore))return new Receipt("IGNORED",null,view(s));
+                s.rootTaskId=e.rootTaskId();s.collectionSessionId=e.collectionSessionId();s.boundAt=e.observation().observedAt();
+                s.expiresAt=s.boundAt.plusSeconds(14400).isBefore(s.expiresAt)?s.boundAt.plusSeconds(14400):s.expiresAt;s.state="BOUND";
+                advance(s);
+                if(terminal(s)){save(s);return new Receipt("STOPPED",null,view(s));}
+            }
+            require(s.rootTaskId.equals(e.rootTaskId())&&s.collectionSessionId.equals(e.collectionSessionId()),"DIAGNOSTIC_ROOT_FROZEN");
+            if(ended(task)&&!"DRAINING".equals(s.state)){s.state="DRAINING";s.rootEndedAt=clock.instant();}
+            if("LIST".equals(e.side())&&("DRAINING".equals(s.state)||s.candidates>=MAX_ORDERS)){save(s);return new Receipt("IGNORED",null,view(s));}
+            if("DETAIL".equals(e.side())&&pending.isEmpty()){save(s);return new Receipt("IGNORED",null,view(s));}
+            if(s.rootEndedAt!=null&&e.observation().observedAt().isAfter(s.rootEndedAt)){save(s);return new Receipt("IGNORED",null,view(s));}
+            // Retained bytes/rows are monotonic and count both observations; promotion never refunds quota.
+            int overhead="LIST".equals(e.side())?20:0;
+            if(s.rows+e.skus().size()>MAX_ROWS||s.bytes+bytes+overhead>MAX_BYTES){end(s,"INCOMPLETE");save(s);return new Receipt("STOPPED",null,view(s));}
+            s.rows+=e.skus().size();s.bytes+=bytes+overhead;
+            String comparison=null;
+            if("LIST".equals(e.side())){
+                s.candidates++;
+                jdbc.update("INSERT INTO ll_selfop_quantity_diagnostic_pending(diagnostic_session_id,order_key_hash,warehouse_code,owner_code,order_no,evidence_json) VALUES(?,?,?,?,?,?)",e.diagnosticSessionId(),key,e.warehouseCode(),e.ownerCode(),e.orderNo(),payload);
+                if(!Boolean.TRUE.equals(e.observation().skuSetComplete())||e.skus().stream().anyMatch(v->v.listQty()==null)){comparison="INCONCLUSIVE";end(s,comparison);}
+            }else{
+                Event list=pending.get(0);comparison=compare(list,e);String evidence=encode(new Pair(list,e));s.paired++;
+                require(size(evidence)<=MAX_REQUEST_BYTES,"DIAGNOSTIC_PAIR_TOO_LARGE");
+                jdbc.update("INSERT INTO ll_selfop_quantity_diagnostic_order(diagnostic_session_id,order_key_hash,source_account,warehouse_code,owner_code,order_no,payload_hash,comparison,sku_rows,payload_bytes,evidence_json,received_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",e.diagnosticSessionId(),key,e.sourceAccount(),e.warehouseCode(),e.ownerCode(),e.orderNo(),hash(evidence),comparison,Math.max(list.skus().size(),e.skus().size()),size(evidence),evidence,clock.instant().toString());
+                jdbc.update("DELETE FROM ll_selfop_quantity_diagnostic_pending WHERE diagnostic_session_id=? AND order_key_hash=?",e.diagnosticSessionId(),key);
+                if(!"ALL_EQUAL".equals(comparison))end(s,comparison);
+                else if(s.paired==MAX_ORDERS)end(s,s.deliveryUncertain?"INCOMPLETE":"COMPLETED");
+            }
+            save(s);return new Receipt("ACCEPTED",comparison,view(s));
+        });
+    }
+    private void advance(Session s){
+        if(terminal(s))return;
+        Instant now=clock.instant();
+        if(!now.isBefore(s.expiresAt)){end(s,"INCOMPLETE");return;}
+        if(s.rootTaskId==null){if(!now.isBefore(s.bindBefore))end(s,"BIND_TIMEOUT");return;}
+        if(ended(root(s.rootTaskId))&&!"DRAINING".equals(s.state)){s.state="DRAINING";s.rootEndedAt=now;}
+    }
+    private void finish(Session s){end(s,s.deliveryUncertain||s.candidates!=s.paired?"INCOMPLETE":s.paired<MAX_ORDERS?"ROOT_ENDED_INSUFFICIENT":"COMPLETED");}
+    private static boolean terminal(Session s){return "TERMINAL".equals(s.state);}
+    private static void end(Session s,String reason){s.state="TERMINAL";s.reason=reason;}
+    private static boolean ended(SelfopSyncTask t){return t==null||Set.of("SUCCEEDED","DEAD","FAILED","ABORTED","CANCELLED").contains(t.getStatus());}
+    private static boolean eligible(SelfopSyncTask t,String account){return t!=null&&account.equals(t.getSourceAccount())&&"ACCOUNT_DISCOVERY".equals(t.getTaskRole())&&SelfopCollectionPolicy.continuousBusinessRoot(t);}
+    private SelfopSyncTask root(String id){
+        require(id!=null&&id.matches("[0-9]{1,19}"),"DIAGNOSTIC_ROOT_ID_INVALID");
+        var rows=jdbc.query("SELECT id,source_account,execution_purpose,upstream_task_id,validation_root_task_id,requested_by,scope_hash,request_key,task_role,task_type,discovery_scope_id,query_from_utc,query_until_utc,query_scope_json,status FROM ll_selfop_sync_task WHERE id=?",new BeanPropertyRowMapper<>(SelfopSyncTask.class),id);
+        return rows.isEmpty()?null:rows.get(0);
+    }
+    private Session load(String id){
+        var sessions=jdbc.query("SELECT state_json,root_task_id,collection_session_id,starts_at,expires_at,accepted_orders,accepted_sku_rows,accepted_bytes,stop_reason FROM ll_selfop_quantity_diagnostic_session WHERE diagnostic_session_id=? FOR UPDATE",(r,n)->{
+            String metadata=r.getString(1);require(metadata!=null,"DIAGNOSTIC_HISTORICAL_ID_READ_ONLY");Session s=decode(metadata,Session.class);
+            // Existing columns remain the sole authority for counters, root, deadline and stop reason.
+            s.rootTaskId=r.getString(2);s.collectionSessionId=r.getString(3);s.boundAt=s.rootTaskId==null?null:Instant.parse(r.getString(4));s.expiresAt=Instant.parse(r.getString(5));
+            s.paired=r.getInt(6);s.rows=r.getInt(7);s.bytes=r.getLong(8);s.reason=r.getString(9);
+            s.state=s.reason!=null?"TERMINAL":s.rootTaskId==null?"ARMED":s.rootEndedAt!=null?"DRAINING":"BOUND";return s;
+        },id);return sessions.isEmpty()?null:sessions.get(0);
+    }
+    private void save(Session s){jdbc.update("UPDATE ll_selfop_quantity_diagnostic_session SET state_json=?,active_slot=?,collection_session_id=?,root_task_id=?,starts_at=?,expires_at=?,accepted_orders=?,accepted_sku_rows=?,accepted_bytes=?,stop_reason=? WHERE diagnostic_session_id=?",encode(s),terminal(s)?null:1,s.collectionSessionId,s.rootTaskId,(s.boundAt==null?s.armedAt:s.boundAt).toString(),s.expiresAt.toString(),s.paired,s.rows,s.bytes,s.reason,s.scope.diagnosticSessionId());}
+    private View view(Session s){
+        var selected=jdbc.query("SELECT warehouse_code,owner_code,order_no,0 AS paired FROM ll_selfop_quantity_diagnostic_pending WHERE diagnostic_session_id=? UNION ALL SELECT warehouse_code,owner_code,order_no,1 AS paired FROM ll_selfop_quantity_diagnostic_order WHERE diagnostic_session_id=?",(r,n)->new Key(r.getString(1),r.getString(2),r.getString(3),r.getBoolean(4)),s.scope.diagnosticSessionId(),s.scope.diagnosticSessionId());
+        return new View(s.scope.diagnosticSessionId(),s.scope.sourceAccount(),s.rootTaskId,s.collectionSessionId,s.state,s.reason,s.boundAt==null?s.armedAt:s.boundAt,s.expiresAt,s.bindBefore,s.scope.warehouseCodes(),s.scope.ownerCode(),s.scope.excludedRootTaskId(),s.candidates,s.paired,s.rows,s.bytes,s.deliveryUncertain,selected);
+    }
+    private String encode(Object value){try{return json.writeValueAsString(value);}catch(Exception e){throw invalid("DIAGNOSTIC_ENCODING_FAILED");}}
+    private <T>T decode(String value,Class<T> type){try{return json.readValue(value,type);}catch(Exception e){throw new IllegalStateException("DIAGNOSTIC_STORAGE_INVALID");}}
+    private static int size(String s){return s.getBytes(StandardCharsets.UTF_8).length;}
+    private static String hash(String s){try{return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(s.getBytes(StandardCharsets.UTF_8)));}catch(Exception e){throw new IllegalStateException(e);}}
+    public record Pair(Event list,Event detail){}
+    public static class Session {
+        public Arm scope;public String producerId;
+        public Instant armedAt,bindBefore,rootEndedAt;
+        public int candidates;public boolean deliveryUncertain;
+        @com.fasterxml.jackson.annotation.JsonIgnore public String state,reason,rootTaskId,collectionSessionId;
+        @com.fasterxml.jackson.annotation.JsonIgnore public Instant boundAt,expiresAt;
+        @com.fasterxml.jackson.annotation.JsonIgnore public int paired,rows;
+        @com.fasterxml.jackson.annotation.JsonIgnore public long bytes;
     }
 }

lastleg-service/src/main/resources/db/selfop-quantity-diagnostics.sql

@@ -2,16 +2,20 @@
 -- Run only after explicit TEST SQL approval. Not executed by service startup.
 CREATE TABLE ll_selfop_quantity_diagnostic_session (
   diagnostic_session_id VARCHAR(128) NOT NULL COMMENT '固定有界数量诊断会话标识',
-  configuration_hash CHAR(64) NOT NULL COMMENT '冻结范围、UTC起止时间及限额的SHA256摘要',
+  configuration_hash CHAR(64) NOT NULL COMMENT '冻结授权范围的SHA256摘要',
   source_account VARCHAR(64) NOT NULL COMMENT '来源账号别名，不保存凭据',
-  collection_session_id VARCHAR(128) NOT NULL COMMENT '被观察的来源采集会话标识',
-  root_task_id VARCHAR(128) NOT NULL COMMENT '被观察的账号采集根任务标识',
+  collection_session_id VARCHAR(128) NULL COMMENT '被观察的来源采集会话标识',
+  root_task_id VARCHAR(128) NULL COMMENT '被观察的账号采集根任务标识',
   starts_at VARCHAR(40) NOT NULL COMMENT '诊断开始时刻，ISO8601 UTC字符串',
-  expires_at VARCHAR(40) NOT NULL COMMENT '诊断截止时刻，距开始最多30分钟，ISO8601 UTC字符串',
+  expires_at VARCHAR(40) NOT NULL COMMENT '诊断截止时刻，首个LIST观测后最多4小时且授权后最多10小时，ISO8601 UTC字符串',
   accepted_orders INT NOT NULL DEFAULT 0 COMMENT '已接收完整配对订单数，跨重启累计',
-  accepted_sku_rows INT NOT NULL DEFAULT 0 COMMENT '已接收归并SKU行数，跨重启累计',
+  accepted_sku_rows INT NOT NULL DEFAULT 0 COMMENT '两侧已保留SKU行数，跨重启累计',
   accepted_bytes BIGINT NOT NULL DEFAULT 0 COMMENT '已持久化白名单规范JSON的UTF8字节数',
   stop_reason VARCHAR(64) NULL COMMENT '诊断终止原因，空值表示尚未终止',
+  active_slot INT NULL COMMENT '当前授权占位，终态释放；旧证据不自动授权',
+  state_json LONGTEXT NULL COMMENT '新授权的持久生命周期；NULL仅表示未参与此协议的历史证据',
+  UNIQUE KEY uq_qdiag_active (source_account, active_slot),
+  CONSTRAINT chk_qdiag_active CHECK (active_slot IS NULL OR active_slot = 1),
   PRIMARY KEY (diagnostic_session_id),
   CONSTRAINT chk_selfop_qdiag_limits CHECK (accepted_orders BETWEEN 0 AND 100
     AND accepted_sku_rows BETWEEN 0 AND 2000 AND accepted_bytes BETWEEN 0 AND 2097152)
@@ -26,7 +30,7 @@ CREATE TABLE ll_selfop_quantity_diagnostic_order (
   order_no VARCHAR(128) NOT NULL COMMENT '来源出库单号',
   payload_hash CHAR(64) NOT NULL COMMENT '白名单规范载荷SHA256摘要，用于幂等校验',
   comparison VARCHAR(40) NOT NULL COMMENT 'SKU集合与三类数量比较结果',
-  sku_rows INT NOT NULL COMMENT '本单归并后的SKU并集行数',
+  sku_rows INT NOT NULL COMMENT '本单两侧较大的SKU行数；两侧累计额度由会话约束',
   payload_bytes INT NOT NULL COMMENT '本单白名单规范JSON的UTF8字节数',
   evidence_json JSON NOT NULL COMMENT '类型化列表与详情观测及SKU三量证据，不含原始响应或收件信息',
   received_at VARCHAR(40) NOT NULL COMMENT '接收时刻，ISO8601 UTC字符串',
@@ -36,3 +40,14 @@ CREATE TABLE ll_selfop_quantity_diagnostic_order (
   CONSTRAINT chk_selfop_qdiag_order_limits CHECK (sku_rows BETWEEN 1 AND 200
     AND payload_bytes BETWEEN 1 AND 262144)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='独立配对订单数量诊断证据，不参与销量发布';
+
+CREATE TABLE ll_selfop_quantity_diagnostic_pending (
+ diagnostic_session_id VARCHAR(128) NOT NULL,
+ order_key_hash CHAR(64) NOT NULL,
+ warehouse_code VARCHAR(64) NOT NULL,
+ owner_code VARCHAR(128) NOT NULL,
+ order_no VARCHAR(128) NOT NULL,
+ evidence_json LONGTEXT NOT NULL,
+ PRIMARY KEY(diagnostic_session_id,order_key_hash),
+ FOREIGN KEY(diagnostic_session_id) REFERENCES ll_selfop_quantity_diagnostic_session(diagnostic_session_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='首次LIST白名单证据；配对后同事务移至最终证据表';

lastleg-service/src/main/resources/openapi/selfop-quantity-diagnostics-internal-v1.yaml

@@ -1,40 +1,168 @@
 openapi: 3.0.3
 info:
-  title: Internal bounded self-operated quantity diagnostics
-  version: 1.0.0
+  title: Internal next-root bounded quantity diagnostics
+  version: 2.0.0
+  description: Explicit arm only; no automatic rearm. Bind next normal automatic historical OUTBOUND_QUERY root
+    within 6h. Sampling expires 4h after first LIST observedAt, never later than armedAt+10h. Lastleg session lock
+    is receiver acceptance. Source root creation may predate arm. LOCATE, manual/validation roots, current excluded
+    root and different bound root are rejected/ignored. At most 100 distinct LIST candidates total, 2000 retained
+    SKU rows across both sides, 2MiB canonical evidence plus promotion overhead. Pending-to-final never refunds
+    quota. First unknown or SKU mismatch persists INCONCLUSIVE and stops; any known inequality among list/forecast/actual
+    persists QUANTITY_DIFFERENCE and stops. Original side timestamps/attempts are immutable; exact replay returns
+    stored receipt even after termination, changed same-side evidence is rejected. Root termination drains only
+    accepted candidates via an explicit Integration FIFO boundary without extending expiry. Fewer than 100 complete
+    pairs is ROOT_ENDED_INSUFFICIENT; unpaired/lost tail is INCOMPLETE. Producer restart retains persisted candidates
+    but marks delivery uncertain. Every request/response is stream-bounded to 262144 bytes. Never exposed through
+    Gateway, never creates supplier requests or changes business facts.
 paths:
-  /internal/lastleg/self-operated-analytics/quantity-diagnostics/orders:
+  /internal/lastleg/self-operated-analytics/quantity-diagnostics/arm:
     post:
-      operationId: receiveSelfopQuantityDiagnosticOrder
-      description: >-
-        Internal service-token only; never routed through Gateway. One complete paired order per request.
-        Quantities are evidence only. Session scope and UTC time interval are frozen by server configuration.
-        Maximum 100 orders, 2000 SKU rows and 2097152 canonical UTF8 bytes per session;
-        200 SKU rows and 262144 wire/canonical bytes per request; at most 30 minutes.
-        First known forecast/actual difference closes capture regardless of comparison classification.
-        Same identity and canonical evidence is idempotent, including after closure; changed evidence is 409.
+      operationId: quantityDiagnosticArm
       security:
-        - internalServiceToken: []
+      - internalServiceToken: []
       requestBody:
         required: true
         content:
           application/json:
             schema:
-              $ref: '#/components/schemas/PairedOrder'
+              $ref: '#/components/schemas/Arm'
       responses:
         '200':
-          description: Diagnostic receipt in the standard ApiResponse wrapper; code is string 200.
+          description: ApiResponse code 200. status data is null if no grant exists.
           content:
             application/json:
               schema:
                 type: object
+                additionalProperties: false
                 properties:
-                  code: { type: string, example: '200' }
-                  data: { $ref: '#/components/schemas/Receipt' }
-        '400': { description: 'Invalid JSON, unknown fields, invalid scope/evidence, duplicate SKU or request limit.' }
-        '401': { description: Invalid internal service token. }
-        '409': { description: Session configuration changed or order identity reused with different evidence. }
-        '503': { description: Internal service token unconfigured or diagnostic storage unavailable. }
+                  code:
+                    type: string
+                  message:
+                    type: string
+                    nullable: true
+                  data:
+                    $ref: '#/components/schemas/View'
+                    nullable: true
+                required:
+                - code
+                - data
+        '400':
+          description: Invalid scope, changed frozen evidence, invalid observation or disabled feature.
+        '401':
+          description: Missing or invalid internal service token.
+        '503':
+          description: Diagnostic admission, dedicated connection or storage unavailable; collection is unaffected.
+  /internal/lastleg/self-operated-analytics/quantity-diagnostics/status:
+    post:
+      operationId: quantityDiagnosticStatus
+      security:
+      - internalServiceToken: []
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/Status'
+      responses:
+        '200':
+          description: ApiResponse code 200. status data is null if no grant exists.
+          content:
+            application/json:
+              schema:
+                type: object
+                additionalProperties: false
+                properties:
+                  code:
+                    type: string
+                  message:
+                    type: string
+                    nullable: true
+                  data:
+                    $ref: '#/components/schemas/View'
+                    nullable: true
+                required:
+                - code
+                - data
+        '400':
+          description: Invalid scope, changed frozen evidence, invalid observation or disabled feature.
+        '401':
+          description: Missing or invalid internal service token.
+        '503':
+          description: Diagnostic admission, dedicated connection or storage unavailable; collection is unaffected.
+  /internal/lastleg/self-operated-analytics/quantity-diagnostics/observe:
+    post:
+      operationId: quantityDiagnosticObserve
+      security:
+      - internalServiceToken: []
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/Event'
+      responses:
+        '200':
+          description: ApiResponse code 200. status data is null if no grant exists.
+          content:
+            application/json:
+              schema:
+                type: object
+                additionalProperties: false
+                properties:
+                  code:
+                    type: string
+                  message:
+                    type: string
+                    nullable: true
+                  data:
+                    $ref: '#/components/schemas/Receipt'
+                    nullable: true
+                required:
+                - code
+                - data
+        '400':
+          description: Invalid scope, changed frozen evidence, invalid observation or disabled feature.
+        '401':
+          description: Missing or invalid internal service token.
+        '503':
+          description: Diagnostic admission, dedicated connection or storage unavailable; collection is unaffected.
+  /internal/lastleg/self-operated-analytics/quantity-diagnostics/stop:
+    post:
+      operationId: quantityDiagnosticStop
+      security:
+      - internalServiceToken: []
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/Stop'
+      responses:
+        '200':
+          description: ApiResponse code 200. status data is null if no grant exists.
+          content:
+            application/json:
+              schema:
+                type: object
+                additionalProperties: false
+                properties:
+                  code:
+                    type: string
+                  message:
+                    type: string
+                    nullable: true
+                  data:
+                    $ref: '#/components/schemas/View'
+                    nullable: true
+                required:
+                - code
+                - data
+        '400':
+          description: Invalid scope, changed frozen evidence, invalid observation or disabled feature.
+        '401':
+          description: Missing or invalid internal service token.
+        '503':
+          description: Diagnostic admission, dedicated connection or storage unavailable; collection is unaffected.
 components:
   securitySchemes:
     internalServiceToken:
@@ -42,64 +170,365 @@ components:
       in: header
       name: X-Internal-Service-Token
   schemas:
-    PairedOrder:
+    Arm:
       type: object
       additionalProperties: false
-      required: [diagnosticSessionId, collectionSessionId, rootTaskId, sourceAccount, warehouseCode, ownerCode, orderNo, listObservation, detailObservation, skus]
       properties:
-        diagnosticSessionId: { type: string, maxLength: 128 }
-        collectionSessionId: { type: string, maxLength: 128 }
-        rootTaskId: { type: string, maxLength: 128 }
-        sourceAccount: { type: string, maxLength: 64 }
-        warehouseCode: { type: string, maxLength: 64 }
-        ownerCode: { type: string, maxLength: 128 }
-        orderNo: { type: string, maxLength: 128 }
-        listObservation: { $ref: '#/components/schemas/Observation' }
-        detailObservation: { $ref: '#/components/schemas/Observation' }
-        skus:
+        diagnosticSessionId:
+          type: string
+          maxLength: 128
+          minLength: 1
+        sourceAccount:
+          type: string
+          maxLength: 64
+          minLength: 1
+        warehouseCodes:
           type: array
           minItems: 1
-          maxItems: 200
-          description: Union of list/detail SKUs; duplicate source lines aggregated, unknown never zero-filled.
-          items: { $ref: '#/components/schemas/Sku' }
+          maxItems: 8
+          uniqueItems: true
+          items:
+            type: string
+            maxLength: 64
+            minLength: 1
+        ownerCode:
+          type: string
+          maxLength: 128
+          minLength: 1
+          description: Explicit owner code, or * for all allowed owners.
+        excludedRootTaskId:
+          type: string
+          maxLength: 128
+          minLength: 1
+          nullable: true
+          description: Must match the actual started, unfinished automatic root at arm; null only when none exists.
+      required:
+      - diagnosticSessionId
+      - sourceAccount
+      - warehouseCodes
+      - ownerCode
+    Status:
+      type: object
+      additionalProperties: false
+      properties:
+        sourceAccount:
+          type: string
+          maxLength: 64
+          minLength: 1
+        producerId:
+          type: string
+          maxLength: 128
+          minLength: 1
+        attach:
+          type: boolean
+          description: True only on Integration process attachment; replacement marks delivery uncertain and fences
+            prior producer.
+        drained:
+          type: boolean
+          description: Only after observing DRAINING, closing intake and draining previously queued callbacks.
+      required:
+      - sourceAccount
+      - producerId
+      - attach
+      - drained
+    Stop:
+      type: object
+      additionalProperties: false
+      properties:
+        diagnosticSessionId:
+          type: string
+          maxLength: 128
+          minLength: 1
+        sourceAccount:
+          type: string
+          maxLength: 64
+          minLength: 1
+        producerId:
+          type: string
+          maxLength: 128
+          minLength: 1
+        reason:
+          type: string
+          enum:
+          - INCOMPLETE
+          - QUEUE_FULL
+          - OBSERVER_FAILED
+          - SHUTDOWN
+      required:
+      - diagnosticSessionId
+      - sourceAccount
+      - producerId
+      - reason
     Observation:
       type: object
       additionalProperties: false
-      required: [observedAt, attemptId, skuSetComplete]
       properties:
-        status: { type: string, nullable: true, maxLength: 32 }
+        status:
+          type: string
+          maxLength: 32
+          minLength: 1
+          nullable: true
         outboundTime:
           type: string
+          maxLength: 64
+          minLength: 1
           nullable: true
-          maxLength: 32
-          description: Original Asia/Shanghai source text, null if absent; not reinterpreted as another timezone.
-        observedAt: { type: string, format: date-time }
-        attemptId: { type: string, maxLength: 128 }
-        skuSetComplete: { type: boolean }
+        observedAt:
+          type: string
+          format: date-time
+        attemptId:
+          type: string
+          maxLength: 128
+          minLength: 1
+        skuSetComplete:
+          type: boolean
+      required:
+      - status
+      - outboundTime
+      - observedAt
+      - attemptId
+      - skuSetComplete
     Sku:
       type: object
       additionalProperties: false
-      required: [sku, listPresent, detailPresent]
       properties:
-        sku: { type: string, maxLength: 128 }
-        listQty: { type: number, nullable: true, minimum: 0, multipleOf: 0.000001 }
-        forecastQty: { type: number, nullable: true, minimum: 0, multipleOf: 0.000001 }
-        actualQty: { type: number, nullable: true, minimum: 0, multipleOf: 0.000001 }
-        listPresent: { type: boolean }
-        detailPresent: { type: boolean }
-    Receipt:
-      type: object
-      required: [outcome, acceptedOrders, acceptedSkuRows, acceptedBytes]
-      properties:
-        outcome: { type: string, enum: [ACCEPTED, DUPLICATE, STOPPED] }
-        comparison:
+        sku:
           type: string
+          maxLength: 128
+          minLength: 1
+        listQty:
+          type: number
           nullable: true
-          enum: [SKU_SET_MISMATCH, QUANTITY_UNKNOWN, FORECAST_ACTUAL_DIFFERENCE, ALL_EQUAL, LIST_QUANTITY_DIFFERENCE]
+          minimum: 0
+          description: Up to 32 integer digits, at most 6 decimal places; unknown stays null.
+        forecastQty:
+          type: number
+          nullable: true
+          minimum: 0
+          description: Up to 32 integer digits, at most 6 decimal places; unknown stays null.
+        actualQty:
+          type: number
+          nullable: true
+          minimum: 0
+          description: Up to 32 integer digits, at most 6 decimal places; unknown stays null.
+      required:
+      - sku
+      - listQty
+      - forecastQty
+      - actualQty
+    Key:
+      type: object
+      additionalProperties: false
+      properties:
+        warehouseCode:
+          type: string
+          maxLength: 64
+          minLength: 1
+        ownerCode:
+          type: string
+          maxLength: 128
+          minLength: 1
+        orderNo:
+          type: string
+          maxLength: 128
+          minLength: 1
+        paired:
+          type: boolean
+      required:
+      - warehouseCode
+      - ownerCode
+      - orderNo
+      - paired
+    Event:
+      type: object
+      additionalProperties: false
+      properties:
+        diagnosticSessionId:
+          type: string
+          maxLength: 128
+          minLength: 1
+        sourceAccount:
+          type: string
+          maxLength: 64
+          minLength: 1
+        producerId:
+          type: string
+          maxLength: 128
+          minLength: 1
+        rootTaskId:
+          type: string
+          maxLength: 128
+          minLength: 1
+        collectionSessionId:
+          type: string
+          maxLength: 128
+          minLength: 1
+        operation:
+          type: string
+          enum:
+          - OUTBOUND_QUERY
+          - OUTBOUND_DETAIL
+        warehouseCode:
+          type: string
+          maxLength: 64
+          minLength: 1
+        ownerCode:
+          type: string
+          maxLength: 128
+          minLength: 1
+        orderNo:
+          type: string
+          maxLength: 128
+          minLength: 1
+        side:
+          type: string
+          enum:
+          - LIST
+          - DETAIL
+        observation:
+          $ref: '#/components/schemas/Observation'
+        skus:
+          type: array
+          maxItems: 200
+          items:
+            $ref: '#/components/schemas/Sku'
+          description: Empty allowed only when skuSetComplete=false; persists inconclusive evidence.
+      required:
+      - diagnosticSessionId
+      - sourceAccount
+      - producerId
+      - rootTaskId
+      - collectionSessionId
+      - operation
+      - warehouseCode
+      - ownerCode
+      - orderNo
+      - side
+      - observation
+      - skus
+    View:
+      type: object
+      additionalProperties: false
+      properties:
+        diagnosticSessionId:
+          type: string
+          maxLength: 128
+          minLength: 1
+        sourceAccount:
+          type: string
+          maxLength: 64
+          minLength: 1
+        rootTaskId:
+          type: string
+          maxLength: 128
+          minLength: 1
+          nullable: true
+        collectionSessionId:
+          type: string
+          maxLength: 128
+          minLength: 1
+          nullable: true
+        state:
+          type: string
+          enum:
+          - ARMED
+          - BOUND
+          - DRAINING
+          - TERMINAL
         stopReason:
           type: string
           nullable: true
-          enum: [DISABLED, NOT_STARTED, EXPIRED, LIMIT_REACHED, FORECAST_ACTUAL_DIFFERENCE]
-        acceptedOrders: { type: integer }
-        acceptedSkuRows: { type: integer }
-        acceptedBytes: { type: integer, format: int64, description: Canonical whitelist JSON UTF8 bytes persisted in this session. }
+        startsAt:
+          type: string
+          format: date-time
+          nullable: true
+        expiresAt:
+          type: string
+          format: date-time
+          nullable: true
+        bindBefore:
+          type: string
+          format: date-time
+          nullable: true
+        warehouseCodes:
+          type: array
+          maxItems: 8
+          items:
+            type: string
+            maxLength: 64
+            minLength: 1
+        ownerCode:
+          type: string
+          maxLength: 128
+          minLength: 1
+        excludedRootTaskId:
+          type: string
+          maxLength: 128
+          minLength: 1
+          nullable: true
+        candidates:
+          type: integer
+          minimum: 0
+          maximum: 100
+        pairedOrders:
+          type: integer
+          minimum: 0
+          maximum: 100
+        retainedRows:
+          type: integer
+          minimum: 0
+          maximum: 2000
+        retainedBytes:
+          type: integer
+          minimum: 0
+          maximum: 2097152
+        deliveryUncertain:
+          type: boolean
+        selected:
+          type: array
+          maxItems: 100
+          items:
+            $ref: '#/components/schemas/Key'
+      required:
+      - diagnosticSessionId
+      - sourceAccount
+      - rootTaskId
+      - collectionSessionId
+      - state
+      - stopReason
+      - startsAt
+      - expiresAt
+      - bindBefore
+      - warehouseCodes
+      - ownerCode
+      - excludedRootTaskId
+      - candidates
+      - pairedOrders
+      - retainedRows
+      - retainedBytes
+      - deliveryUncertain
+      - selected
+    Receipt:
+      type: object
+      additionalProperties: false
+      properties:
+        outcome:
+          type: string
+          enum:
+          - ACCEPTED
+          - IGNORED
+          - DUPLICATE
+          - STOPPED
+        comparison:
+          type: string
+          nullable: true
+          enum:
+          - ALL_EQUAL
+          - QUANTITY_DIFFERENCE
+          - INCONCLUSIVE
+        session:
+          $ref: '#/components/schemas/View'
+      required:
+      - outcome
+      - comparison
+      - session