feat(self-operated): persist continuous collection and frozen coverage

2026-10-13 21:13:31 +08:00
parent c2bb769676
commit 242a76221e
45 changed files with 1021 additions and 118 deletions
@@ -17,6 +17,8 @@ public interface SelfopCoordinationMapper {
    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}")
    List<SelfopSyncTask> due(@Param("now") LocalDateTime now, @Param("limit") int limit, @Param("purpose") String purpose);
+    @Select("SELECT t.* FROM ll_selfop_sync_task t JOIN ll_selfop_sync_task r ON r.id=t.business_root_task_id AND r.business_root_task_id=r.id AND r.source_account=t.source_account AND r.execution_purpose='BUSINESS' AND r.validation_root_task_id IS NULL AND r.status<>'DEAD' WHERE t.source_account=#{account} AND t.execution_purpose='BUSINESS' AND t.validation_root_task_id IS NULL AND (r.request_key LIKE 'manual:%' OR (r.requested_by='SELFOP_CONTINUOUS_PLANNER' AND r.request_key LIKE 'continuous:%')) AND t.status IN ('READY','RETRY_WAIT','RUNNING') AND (t.next_retry_at_utc IS NULL OR t.next_retry_at_utc<=#{now}) AND (r.request_key NOT LIKE 'continuous:inventory:%' OR DATE(DATE_ADD(r.planned_at_utc,INTERVAL 8 HOUR))=DATE(DATE_ADD(#{now},INTERVAL 8 HOUR))) ORDER BY CASE WHEN r.task_role='ACCOUNT_DISCOVERY' AND r.query_until_utc>=DATE_SUB(DATE(DATE_ADD(#{now},INTERVAL 8 HOUR)),INTERVAL 8 HOUR) THEN 0 WHEN r.request_key LIKE 'continuous:inventory:%' THEN 1 WHEN t.task_role='WAREHOUSE_SYNC' THEN 2 WHEN t.task_type IN ('RECHECK','REBUILD') THEN 3 ELSE 4 END,t.query_until_utc DESC,t.id LIMIT #{limit}")
+    List<SelfopSyncTask> dueContinuous(@Param("now") LocalDateTime now,@Param("limit") int limit,@Param("account") String account);
    @Select("SELECT COUNT(*) FROM ll_selfop_sync_task WHERE source_account=#{account} AND apply_state IN ('APPLYING','BLOCKED') AND JSON_CONTAINS(affected_warehouse_codes_json,JSON_QUOTE(#{warehouse}))")
    long barriers(@Param("account") String account, @Param("warehouse") String warehouse);
    @Update("UPDATE ll_selfop_sync_task SET status='SUCCEEDED',result_published_run_id=#{published},next_retry_at_utc=NULL,updated_at=#{now} WHERE scope_id=#{scope} AND scope_hash=#{hash} AND execution_purpose='BUSINESS' AND apply_state='APPLIED' AND (query_from_utc IS NULL OR (query_from_utc>=#{from} AND query_until_utc<=#{until})) AND status IN ('READY','RUNNING','RETRY_WAIT','DEAD')")
@@ -13,6 +13,8 @@ import java.time.LocalDateTime;
@ConfigurationProperties(prefix = "baiyue.self-operated-analytics")
 public class SelfopAnalyticsProperties {
    private String collectionMode = "OFF";
+    /** Explicit account-owner continuous BUSINESS planner/egress mode; off by default. */
+    private boolean continuousBusinessCollectionEnabled = false;
    private String environment = "test";
    private String exclusiveGrantId;
    private LocalDateTime exclusiveGrantExpiresAtUtc;
@@ -28,14 +28,46 @@ public final class SelfopCollectionPolicy {
        requireExecution(config,"BUSINESS");
        require("production".equals(config.getEnvironment()),"SELFOP_AUTOMATIC_PLANNING_ENVIRONMENT_REJECTED");
    }
+    public static final String CONTINUOUS_PLANNER="SELFOP_CONTINUOUS_PLANNER";
+    public static boolean continuousBusiness(SelfopAnalyticsProperties config){
+        return config.isContinuousBusinessCollectionEnabled()&&"BUSINESS".equals(config.getCollectionMode())&&businessEnvironment(config);
+    }
+    /** Strict persisted server identity; enabling continuous mode never admits old cron slots. */
+    public static boolean continuousBusinessRoot(SelfopSyncTask task){
+        if(task==null||!"BUSINESS".equals(task.getExecutionPurpose())||task.getUpstreamTaskId()!=null
+            ||task.getValidationRootTaskId()!=null||!CONTINUOUS_PLANNER.equals(task.getRequestedBy())||task.getScopeHash()==null||!task.getScopeHash().matches("[a-f0-9]{64}"))return false;
+        String key=task.getRequestKey();if(key==null)return false;
+        if("ACCOUNT_DISCOVERY".equals(task.getTaskRole())&&"BACKFILL".equals(task.getTaskType())&&task.getDiscoveryScopeId()!=null
+            &&task.getQueryFromUtc()!=null&&task.getQueryUntilUtc()!=null&&task.getQueryUntilUtc().equals(task.getQueryFromUtc().plusDays(1))){
+            var local=task.getQueryFromUtc().plusHours(8);
+            if(!local.toLocalTime().equals(java.time.LocalTime.MIDNIGHT)||!key.equals("continuous:outbound:"+local.toLocalDate()+":"+task.getScopeHash().substring(0,12)))return false;
+            try{Map<?,?> q=read(task.getQueryScopeJson(),Map.class);
+                return q!=null&&q.get("parentTaskId")==null&&q.get("deliveryNo")==null&&q.get("customerCode")==null
+                    &&local.toString().equals(q.get("startTime"))&&task.getQueryUntilUtc().plusHours(8).toString().equals(q.get("endTime"));
+            }catch(RuntimeException invalid){return false;}
+        }
+        if(!"WAREHOUSE_SYNC".equals(task.getTaskRole())||!"SCHEDULED".equals(task.getTaskType())||task.getScopeId()==null
+            ||task.getDiscoveryScopeId()!=null||task.getWarehouseCode()==null||task.getPlannedAtUtc()==null
+            ||task.getQueryFromUtc()!=null||task.getQueryUntilUtc()!=null
+            ||!key.equals("continuous:inventory:"+task.getWarehouseCode()+":"+task.getPlannedAtUtc()+":"+task.getScopeHash().substring(0,12)))return false;
+        try{Map<?,?> q=read(task.getQueryScopeJson(),Map.class);
+            return q!=null&&"INVENTORY".equals(q.get("dataFamily"))&&"Asia/Shanghai".equals(q.get("businessTimezone"))
+                &&java.util.Objects.equals(task.getSourceAccount(),q.get("sourceAccount"))&&java.util.Objects.equals(task.getWarehouseCode(),q.get("warehouseCode"));
+        }catch(RuntimeException invalid){return false;}
+    }
    /** TEST may poll only persisted manual roots and their existing routing/refresh descendants. */
    public static boolean controlledBusinessTask(SelfopSyncTask task,java.util.function.Function<Long,SelfopSyncTask> lookup) {
+        return controlledBusinessTask(task,lookup,false);
+    }
+    public static boolean controlledBusinessTask(SelfopSyncTask task,java.util.function.Function<Long,SelfopSyncTask> lookup,boolean allowContinuous) {
        if(task==null)return false;
        String account=task.getSourceAccount();var seen=new java.util.HashSet<Long>();
        for(int depth=0;task!=null&&depth<64;depth++) {
            if(!"BUSINESS".equals(task.getExecutionPurpose())||task.getValidationRootTaskId()!=null
-                ||!java.util.Objects.equals(account,task.getSourceAccount())||"SCHEDULED".equals(task.getTaskType()))return false;
+                ||!java.util.Objects.equals(account,task.getSourceAccount()))return false;
            if(task.getId()!=null&&!seen.add(task.getId()))return false;
+            if(allowContinuous&&continuousBusinessRoot(task))return true;
+            if("SCHEDULED".equals(task.getTaskType())&&(!allowContinuous||task.getUpstreamTaskId()==null))return false;
            String key=task.getRequestKey();
            if(key!=null&&key.startsWith("manual:")&&task.getRequestedBy()!=null&&!task.getRequestedBy().isBlank())return true;
            Long parent=task.getUpstreamTaskId();
@@ -0,0 +1,46 @@
+package com.baiyue.lastleg.selfoperated.analytics.model;
+
+import java.time.LocalDateTime;
+import java.util.*;
+import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*;
+
+/** Immutable union captured under the warehouse scope lock in the candidate's first transaction. */
+public final class SelfopSalesCoverage {
+    public static final String KEY="salesCoverageV1";
+    private SelfopSalesCoverage() { }
+    public record Interval(LocalDateTime from,LocalDateTime until) { }
+    public record Evidence(Long runId,Long scopeId,String scopeHash,List<Interval> intervals) { }
+    public static Map<String,Object> settings(String snapshot){
+        Map<String,Object> result=new LinkedHashMap<>();
+        if(snapshot!=null)read(snapshot,Map.class).forEach((k,v)->result.put(String.valueOf(k),v));
+        result.remove(KEY);return result;
+    }
+    public static void freeze(SelfopSyncRun run,List<Interval> intervals){
+        validate(intervals);var snapshot=settings(run.getScopeSnapshotJson());
+        snapshot.put(KEY,new Evidence(run.getId(),run.getScopeId(),run.getScopeHash(),List.copyOf(intervals)));
+        run.setScopeSnapshotJson(json(snapshot));
+    }
+    public static Evidence evidence(SelfopSyncRun run){
+        if(run==null||run.getScopeSnapshotJson()==null)return null;
+        Object value=read(run.getScopeSnapshotJson(),Map.class).get(KEY);if(value==null)return null;
+        var evidence=read(json(value),Evidence.class);
+        require(Objects.equals(run.getId(),evidence.runId())&&Objects.equals(run.getScopeId(),evidence.scopeId())
+            &&Objects.equals(run.getScopeHash(),evidence.scopeHash()),"SELFOP_COVERAGE_VERSION_INVALID");
+        validate(evidence.intervals());return evidence;
+    }
+    public static boolean contains(Evidence evidence,LocalDateTime from,LocalDateTime until){
+        return evidence!=null&&from!=null&&until!=null&&from.isBefore(until)&&evidence.intervals().stream()
+            .anyMatch(i->!i.from().isAfter(from)&&!i.until().isBefore(until));
+    }
+    public static void requireTask(SelfopSyncRun run,SelfopSyncTask task){
+        var proof=evidence(run);require(proof!=null&&!proof.intervals().isEmpty(),"SELFOP_COVERAGE_UNPROVEN");
+        if(task.getQueryFromUtc()!=null||task.getQueryUntilUtc()!=null)
+            require(contains(proof,task.getQueryFromUtc(),task.getQueryUntilUtc()),"SELFOP_TASK_COVERAGE_UNPROVEN");
+    }
+    private static void validate(List<Interval> intervals){
+        require(intervals!=null&&intervals.size()<=500,"SELFOP_COVERAGE_INTERVAL_LIMIT");
+        LocalDateTime last=null;
+        for(var i:intervals){require(i!=null&&i.from()!=null&&i.until()!=null&&i.from().isBefore(i.until())
+            &&(last==null||last.isBefore(i.from())),"SELFOP_COVERAGE_INTERVAL_INVALID");last=i.until();}
+    }
+}
@@ -11,7 +11,7 @@ import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*;
 public final class SelfopStageEstimate {
    private SelfopStageEstimate() { }
    public static final int WINDOW_SECONDS=180, MIN_SECONDS=60, MIN_ADVANCES=3,
-        BUCKET_SECONDS=10, MAX_POINTS=20, STALE_SECONDS=60, VALID_SECONDS=10, MAX_BYTES=1024;
+        BUCKET_SECONDS=10, MAX_POINTS=20, STALE_SECONDS=60, VALID_SECONDS=30, MAX_BYTES=1024;
    public record Estimate(String status,String progressIdentity,Long completed,Long total,
        Instant sampleFrom,Instant sampleUntil,Long completedDelta,BigDecimal ratePerSecond,
        Long remainingSeconds,Instant estimatedAt,Instant validUntil,String unavailableReason) { }
@@ -23,7 +23,7 @@ public final class SelfopStageEstimate {
        return hash(values);
    }

-    /** A reuse/uncertain checkpoint resets the baseline, never contributes throughput. */
+    /** Reused count increases reset the baseline; unchanged bookkeeping is not new work. */
    public static void checkpoint(Map<String,Object> p,Instant now,boolean realWork) {
        try {
            Long completed=number(p.get("completed")),total=number(p.get("total"));
@@ -32,7 +32,8 @@ public final class SelfopStageEstimate {
            }
            String id=identity(p);long second=now.getEpochSecond();
            var old=map(p.get("etaSamples"));var points=points(old.get("p"));
-            if(!id.equals(old.get("i"))||!realWork||points.isEmpty()
+            if(!id.equals(old.get("i"))||points.isEmpty()
+                ||!realWork&&!completed.equals(points.get(points.size()-1).get(1))
                ||second<points.get(points.size()-1).get(0)
                ||completed<points.get(points.size()-1).get(1)
                ||second-numberOr(old.get("a"),second)>=STALE_SECONDS)points.clear();
@@ -66,7 +67,9 @@ public final class SelfopStageEstimate {
            if(reason==null&&(!id.equals(sample.get("i"))||points.isEmpty()))reason="SAMPLING";
            if(reason==null) {
                long second=now.getEpochSecond(),lastAdvance=numberOr(sample.get("a"),0);
-                points.removeIf(point->second-point.get(0)>WINDOW_SECONDS);
+                // Select an anchor that stays inside the window through the read validity
+                // period, instead of expiring each time the oldest 10-second bucket rolls out.
+                points.removeIf(point->second-point.get(0)>WINDOW_SECONDS-VALID_SECONDS);
                if(points.isEmpty()||second-lastAdvance>=STALE_SECONDS)reason="NO_RECENT_PROGRESS";
                else {
                    var first=points.get(0);var last=points.get(points.size()-1);
@@ -9,6 +9,22 @@ import java.util.List;
 /** Database aggregation, with bounded result and insert batches. No raw historical rows enter Java. */
@Mapper
 public interface OutboundAggregateMapper {
+    @Select("""
+        SELECT MIN(from_at) AS covered_from_utc,MAX(until_at) AS covered_until_utc FROM (
+          SELECT from_at,until_at,SUM(CASE WHEN previous_until IS NULL OR from_at>previous_until THEN 1 ELSE 0 END)
+            OVER (ORDER BY from_at,until_at ROWS UNBOUNDED PRECEDING) AS island FROM (
+            SELECT from_at,until_at,MAX(until_at) OVER
+              (ORDER BY from_at,until_at ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS previous_until FROM (
+              SELECT query_from_utc AS from_at,query_until_utc AS until_at FROM ll_selfop_sync_task
+              WHERE scope_id=#{scopeId} AND scope_hash=#{scopeHash} AND execution_purpose='BUSINESS'
+                AND validation_root_task_id IS NULL AND task_role='WAREHOUSE_SYNC' AND apply_state='APPLIED'
+                AND apply_completed_at_utc IS NOT NULL AND query_from_utc IS NOT NULL AND query_until_utc>query_from_utc
+                AND JSON_UNQUOTE(JSON_EXTRACT(routing_manifest_json,'$.directoryComplete'))='true'
+            ) accepted
+          ) ordered_intervals
+        ) islands GROUP BY island ORDER BY MIN(from_at) LIMIT 501
+        """)
+    List<SelfopSyncRun> provenCoverageIslands(@Param("scopeId") Long scopeId,@Param("scopeHash") String scopeHash);
    @Select("""
        SELECT COUNT(*) FROM ll_selfop_outbound_order o
        LEFT JOIN ll_selfop_outbound_item i ON i.order_id=o.id
@@ -53,6 +53,7 @@ public class OutboundCandidateTransaction {
        OutboundPolicy.DayCoverage days=OutboundPolicy.completeDays(coverage.from(),coverage.until(),scope.getBusinessTimezone());
        run.setCompleteDayFrom(days.from());run.setCompleteDayUntil(days.until());run.setCoverageStatus(coverage.until()==null?"UNKNOWN":"COMPLETE");
        run.setCalculatedFactRevision(scope.getFactRevision());run.setBuildStartedAtUtc(SelfopSupport.utcNow());run.setStatus("BUILDING");
+        SelfopSalesCoverage.freeze(run,provenIslands(scope));
        Checkpoint result=new Checkpoint(runId,"DAILY",null,null,null,0,0,0);save(held,result);return result;
    }

@@ -106,13 +107,20 @@ public class OutboundCandidateTransaction {
        if(published==null || !Boolean.TRUE.equals(published.getWarehouseDailyReady()) || !"PUBLISHED".equals(published.getStatus()) || !"BUSINESS".equals(published.getExecutionPurpose())
                || published.getBuildStartedAtUtc()==null || !Objects.equals(scopeId,published.getScopeId()) || !Objects.equals(scope.getScopeHash(),published.getScopeHash())
                || !Objects.equals(scope.getFactRevision(),published.getCalculatedFactRevision()) || !Objects.equals(scope.getFactRevision(),scope.getPublishedFactRevision())
-                || !Objects.equals(SelfopSupport.read(held.run().getScopeSnapshotJson(),Map.class),SelfopSupport.read(published.getScopeSnapshotJson(),Map.class)))return null;
+                || !Objects.equals(SelfopSalesCoverage.settings(held.run().getScopeSnapshotJson()),SelfopSalesCoverage.settings(published.getScopeSnapshotJson())))return null;
+        var proof=SelfopSalesCoverage.evidence(published);
+        if(proof==null||!Objects.equals(proof.intervals(),provenIslands(scope)))return null;
        OutboundCandidateBuilder.Coverage next=coverage(scope);
        if(!Objects.equals(next.from(),published.getCoveredFromUtc()) || !Objects.equals(next.until(),published.getCoveredUntilUtc()))return null;
        // Content-identical accepted details still change receipt/run evidence; conservatively rebuild them.
        return aggregates.observationsChangedSince(scope.getSourceAccount(),scope.getWarehouseCode(),published.getBuildStartedAtUtc())==0?published.getId():null;
    }

+    private List<SelfopSalesCoverage.Interval> provenIslands(SelfopSyncScope scope){
+        var rows=aggregates.provenCoverageIslands(scope.getId(),scope.getScopeHash());
+        SelfopSupport.require(rows.size()<=500,"SELFOP_COVERAGE_INTERVAL_LIMIT");
+        return rows.stream().map(r->new SelfopSalesCoverage.Interval(r.getCoveredFromUtc(),r.getCoveredUntilUtc())).toList();
+    }
    private Held held(Long scopeId,Long runId,String owner,long fence,boolean building) {
        SelfopSyncScope scope=locks.scope(scopeId);LocalDateTime now=SelfopSupport.utcNow();
        SelfopSupport.require(scope!=null && "OUTBOUND".equals(scope.getDataFamily()) && Objects.equals(scope.getLeaseOwner(),owner) && Objects.equals(scope.getFenceToken(),fence)
@@ -48,11 +48,14 @@ public class AnalyticsQueryService {
            if(!Objects.equals(unit,row.quantityUnit()))valid=false;
        }
        Instant evidence=earliest(sp,ip);
+        var coverage=SalesQueryCoverage.intervals(s.sales);
        List<SalesBucketPolicy.Point> evaluated=new ArrayList<>(); List<AnalyticsResponses.Point> points=new ArrayList<>();
        for(LocalDate day=s.range.startDate();day.isBefore(s.range.endDateExclusive());day=day.plusDays(1)){
            Daily row=daily.get(day);
-            var point=SalesBucketPolicy.evaluate(new SalesBucketPolicy.Input(day,s.range.anchorDate(),BUSINESS_ZONE,s.fullFrom,s.fullUntil,
-                s.context.salesCoveredFrom(),s.context.salesCoveredUntil(),evidence,row==null?null:row.outboundQuantity(),
+            var covered=SalesQueryCoverage.at(row!=null||evidence==null?coverage:
+                coverage.stream().filter(i->i.until().toInstant(ZoneOffset.UTC).isAfter(evidence)).toList(),day);
+            var point=SalesBucketPolicy.evaluate(new SalesBucketPolicy.Input(day,s.range.anchorDate(),BUSINESS_ZONE,SalesQueryCoverage.fullFrom(covered),SalesQueryCoverage.fullUntil(covered),
+                covered==null?null:utc(covered.from()),covered==null?null:utc(covered.until()),evidence,row==null?null:row.outboundQuantity(),
                row==null?null:row.outboundOrderCount(),valid,s.sourceKnown,s.context.rebuildRequired(),"STALE".equals(s.context.salesStatus())));
            evaluated.add(point);
            points.add(new AnalyticsResponses.Point(day,point.outboundQuantity(),point.observedQuantity(),point.dataStatus(),
@@ -113,7 +116,8 @@ public class AnalyticsQueryService {
            null,r.defectivePhysicalQuantity(),r.outboundQuantity30d(),r.lastOutbound(),historyStart(s),List.copyOf(attention),Map.copyOf(statuses));
    }
    private static boolean positive(BigDecimal n){return n!=null&&n.signum()>0;}
-    private static LocalDate historyStart(Snapshot s){return s.context.salesCoveredFrom()==null?null:s.context.salesCoveredFrom().atZone(BUSINESS_ZONE).toLocalDate();}
+    private static LocalDate historyStart(Snapshot s){return SalesQueryCoverage.intervals(s.sales).stream()
+        .map(i->utc(i.from()).atZone(BUSINESS_ZONE).toLocalDate()).min(LocalDate::compareTo).orElse(null);}
    private static BigDecimal divide30(BigDecimal n){return n==null?null:n.divide(BigDecimal.valueOf(30),12,java.math.RoundingMode.HALF_UP);}
    private static BigDecimal decimal(Map<String,Object> row,String key){Object n=row.get(key);return n==null?null:n instanceof BigDecimal d?d:new BigDecimal(n.toString());}
    private static long number(Map<String,Object> row,String key){BigDecimal n=decimal(row,key);return n==null?0:n.longValueExact();}
@@ -162,11 +166,12 @@ public class AnalyticsQueryService {
        if(!"VALID".equals(result.status()))reasons.add(result.status());
        if(r.inventoryReason()!=null)reasons.add(r.inventoryReason());
        BigDecimal today=r.todayQuantity();
-        if(!"HOT".equals(kind)&&today==null&&s.sourceKnown&&Boolean.TRUE.equals(r.unitValid())&&evidence!=null&&s.context.salesCoveredUntil()!=null){
+        if(!"HOT".equals(kind)&&today==null&&s.sourceKnown&&Boolean.TRUE.equals(r.unitValid())&&evidence!=null){
            Instant start=s.range.anchorDate().atStartOfDay(BUSINESS_ZONE).toInstant();
+            Instant end=s.range.anchorDate().plusDays(1).atStartOfDay(BUSINESS_ZONE).toInstant();
            Instant lower=evidence.isAfter(start)?evidence:start;
-            if(s.context.salesCoveredFrom()!=null&&s.context.salesCoveredFrom().isAfter(lower))lower=s.context.salesCoveredFrom();
-            if(lower.isBefore(s.context.salesCoveredUntil()))today=BigDecimal.ZERO;
+            if(lower.isBefore(end)&&SalesQueryCoverage.intervals(s.sales).stream()
+                .anyMatch(i->utc(i.from()).isBefore(end)&&utc(i.until()).isAfter(lower)))today=BigDecimal.ZERO;
        }
        return new AnalyticsResponses.Metric(s.q.sourceAccount(),r.customerCode(),s.q.warehouseCode(),r.skuCode(),r.productName(),r.quantityUnit(),
            r.windowQuantity(),r.windowSkuOrderCount(),today,stockValid?r.physicalQuantity():null,stockValid?r.availableQuantity():null,
@@ -207,7 +212,8 @@ public class AnalyticsQueryService {
        boolean rebuild=salesScope!=null&&(Boolean.TRUE.equals(salesScope.rebuildRequired())
            ||!Objects.equals(salesScope.factRevision(),salesScope.publishedFactRevision())||mapper.applyBarriers(q.sourceAccount(),q.warehouseCode())>0);
        boolean known=sales!=null&&BUSINESS_ZONE.getId().equals(sc.path("sourceTimezone").asText());
-        String salesState=sales==null?"UNKNOWN":!known?"SOURCE_TIMEZONE_UNKNOWN":stale(sales.coveredUntilUtc(),threshold(sc))?"STALE":"VALID";
+        LocalDateTime latestCoverage=SalesQueryCoverage.intervals(sales).stream().map(i->i.until()).max(LocalDateTime::compareTo).orElse(null);
+        String salesState=sales==null?"UNKNOWN":!known?"SOURCE_TIMEZONE_UNKNOWN":stale(latestCoverage,threshold(sc))?"STALE":"VALID";
        if(rebuild)salesState="REBUILD_REQUIRED";
        String inventoryState=inventory==null?"UNKNOWN":stale(inventory.fetchFinishedAtUtc(),threshold(ic))?"STALE":"VALID";
        LocalDate fullFrom=sales==null?null:sales.completeDayFrom(),fullUntil=sales==null?null:sales.completeDayUntil();
@@ -240,7 +246,7 @@ public class AnalyticsQueryService {
            s.context.inventoryRunId(),s.range.startDate(),s.range.endDateExclusive(),s.range.anchorDate(),s.range.anchorDate().minusDays(s.q.rankingDays()),
            s.fullFrom,s.fullUntil,s.sales==null?null:s.sales.coveredFromUtc(),s.sales==null?null:s.sales.coveredUntilUtc(),s.sourceKnown,s.sourceKnown,
            s.trusted&&complete(s,s.range.startDate(),s.range.endDateExclusive()),
-            s.trusted&&complete(s,s.range.anchorDate().minusDays(30),s.range.anchorDate()),"VALID".equals(s.context.inventoryStatus()),s.q.offset(),size,kind);
+            s.trusted&&complete(s,s.range.anchorDate().minusDays(30),s.range.anchorDate()),"VALID".equals(s.context.inventoryStatus()),s.q.offset(),size,kind,SalesQueryCoverage.intervals(s.sales));
    }
    Instant inventoryFreshUntil(Snapshot s){
        return s.inventory()==null||s.inventory().fetchFinishedAtUtc()==null?null:
@@ -255,12 +261,12 @@ public class AnalyticsQueryService {
            ||!scope.warehouseCode().equals(run.warehouseCode())||!scope.scopeHash().equals(run.scopeHash())
            ||requested!=null&&!run.scopeHash().equals(hash)||!"BUSINESS".equals(run.executionPurpose())||!"PUBLISHED".equals(run.status()))
            throw new BizException("版本不可用或范围不符，请刷新整页");
-        if(!"COMPLETE".equals(run.coverageStatus()))throw new BizException("发布版本缺少完整覆盖证明");
+        if(!"COMPLETE".equals(run.coverageStatus())&&SalesQueryCoverage.intervals(run).isEmpty())throw new BizException("发布版本缺少完整覆盖证明");
        return run;
    }
    private Profile profile(Snapshot s,Run run){return run==null?null:mapper.profile(run.id(),s.q.sourceAccount(),s.q.warehouseCode(),s.q.customerCode(),s.q.sku());}
-    private static boolean complete(Snapshot s,LocalDate from,LocalDate until){return s.sourceKnown&&s.sales!=null&&s.sales.coveredFromUtc()!=null&&s.sales.coveredUntilUtc()!=null
-        &&s.sales.coveredFromUtc().isBefore(s.sales.coveredUntilUtc())&&s.fullFrom!=null&&s.fullUntil!=null&&!from.isBefore(s.fullFrom)&&!until.isAfter(s.fullUntil);}
+    private static boolean complete(Snapshot s,LocalDate from,LocalDate until){return s.sourceKnown&&!until.isAfter(s.range.anchorDate())
+        &&SalesQueryCoverage.complete(SalesQueryCoverage.intervals(s.sales),from,until);}
    private JsonNode config(Run r){if(r==null)return json.createObjectNode();try{JsonNode n=json.readTree(r.scopeSnapshotJson());if(n==null||!n.isObject())throw new IllegalArgumentException();return n;}catch(Exception e){throw new BizException("版本配置快照无效");}}
    private static int threshold(JsonNode c){int n=c.path("staleAfterSeconds").asInt();if(n<=0)throw new BizException("版本过期阈值无效");return n;}
    private boolean stale(LocalDateTime t,int seconds){return t==null||utc(t).isAfter(clock.instant())||!clock.instant().isBefore(utc(t).plusSeconds(seconds));}
@@ -30,6 +30,7 @@ public final class AnalyticsReadSql {
    }
    private static String cte(AnalyticsSqlRequest request){
        String kind=request.kind();
+        if("SUMMARY".equals(kind)&&!request.coverage().isEmpty())return coveredSummary(request);
        if("HOT".equals(kind))return hotSalesCte();
        if("SLOW".equals(kind))return rankingCte();
        String localEvidence="CONVERT_TZ(evidence_at_utc,'+00:00','+08:00')";
@@ -152,6 +153,35 @@ public final class AnalyticsReadSql {
        }
        return sql;
    }
+    private static String coveredSummary(AnalyticsSqlRequest p){
+        var intervals=new java.util.ArrayList<String>();
+        for(int i=0;i<p.coverage().size();i++)intervals.add("SELECT CAST(#{p.coverage["+i+"].from} AS DATETIME) lo,CAST(#{p.coverage["+i+"].until} AS DATETIME) hi");
+        var dates=new java.util.ArrayList<String>();
+        for(var day=p.from();day.isBefore(p.until());day=day.plusDays(1))dates.add("SELECT CAST('"+day+"' AS DATE) business_date");
+        String start="TIMESTAMPADD(HOUR,-8,CAST(calendar.business_date AS DATETIME))",end="TIMESTAMPADD(DAY,1,"+start+")";
+        String full="EXISTS (SELECT 1 FROM coverage v WHERE v.lo<="+start+" AND v.hi>="+end+")";
+        String observed="EXISTS (SELECT 1 FROM coverage v WHERE GREATEST(v.lo,b.evidence_at_utc,"+start+")<LEAST(v.hi,"+end+"))";
+        String whole="calendar.business_date<#{p.anchor} AND "+full+" AND (d.id IS NOT NULL OR b.evidence_at_utc<="+start+")";
+        return candidates(true)+" coverage AS ("+String.join(" UNION ALL ",intervals)+"),calendar AS ("+String.join(" UNION ALL ",dates)+"),"
+            +" sort_sales AS (SELECT customer_code,sku_code,SUM(outbound_quantity) observed_sum,MIN(quantity_unit) min_unit,MAX(quantity_unit) max_unit,MIN(outbound_quantity) min_qty"
+            +" FROM ll_selfop_sku_daily_sales WHERE sales_run_id=#{p.salesRun} AND"+KEY_FILTER+" AND business_date>=#{p.from} AND business_date<#{p.until} GROUP BY customer_code,sku_code),"
+            +" rankable AS (SELECT e.*,CASE WHEN e.profile_unit_valid=1 AND (s.min_unit IS NULL OR (s.min_unit=e.quantity_unit AND s.max_unit=e.quantity_unit AND s.min_qty>=0)) THEN 1 ELSE 0 END unit_valid,"
+            +"CASE WHEN #{p.salesKnown}=1 AND e.profile_unit_valid=1 AND (s.min_unit IS NULL OR (s.min_unit=e.quantity_unit AND s.max_unit=e.quantity_unit AND s.min_qty>=0))"
+            +" AND (s.observed_sum IS NOT NULL OR EXISTS (SELECT 1 FROM coverage v WHERE GREATEST(v.lo,e.evidence_at_utc,TIMESTAMPADD(HOUR,-8,CAST(#{p.from} AS DATETIME)))<LEAST(v.hi,TIMESTAMPADD(HOUR,-8,CAST(#{p.until} AS DATETIME))))) THEN COALESCE(s.observed_sum,0) END observed_quantity"
+            +" FROM evidence e LEFT JOIN sort_sales s ON s.customer_code=e.customer_code AND s.sku_code=e.sku_code),"
+            +" paged_keys AS (SELECT * FROM rankable ORDER BY observed_quantity IS NULL,observed_quantity DESC,sku_code,customer_code LIMIT #{p.size} OFFSET #{p.offset}),"
+            +" day_counts AS (SELECT b.customer_code,b.sku_code,SUM(d.outbound_order_count) sku_order_count,"
+            +"SUM(CASE WHEN #{p.formalSales}=1 AND "+whole+" THEN 1 ELSE 0 END) complete_count,"
+            +"SUM(CASE WHEN #{p.salesKnown}=1 AND (d.id IS NOT NULL OR "+observed+") THEN 1 ELSE 0 END) observed_count"
+            +" FROM paged_keys b CROSS JOIN calendar LEFT JOIN ll_selfop_sku_daily_sales d ON d.sales_run_id=#{p.salesRun} AND d.source_account=#{p.account} AND d.warehouse_code=#{p.warehouse}"
+            +" AND d.customer_code=b.customer_code AND d.sku_code=b.sku_code AND d.business_date=calendar.business_date GROUP BY b.customer_code,b.sku_code),"
+            +" result AS (SELECT b.customer_code,b.sku_code,b.product_name,b.quantity_unit,b.evidence_at_utc,b.unit_valid,b.observed_quantity,n.sku_order_count,"
+            +"CASE WHEN b.unit_valid=1 THEN n.complete_count ELSE 0 END complete_day_count,"
+            +"CASE WHEN b.unit_valid=1 THEN n.observed_count-n.complete_count ELSE 0 END partial_day_count,"
+            +"TIMESTAMPDIFF(DAY,#{p.from},#{p.until})-CASE WHEN b.unit_valid=1 THEN n.observed_count ELSE 0 END unknown_day_count,"
+            +"NULL window_quantity,NULL window_sku_order_count,NULL sales30_quantity,NULL today_quantity,NULL physical_quantity,NULL available_quantity,NULL frozen_quantity,NULL transit_quantity,"
+            +"NULL inventory_observed_at_utc,'UNKNOWN' inventory_quality,NULL inventory_reason FROM paged_keys b JOIN day_counts n ON n.customer_code=b.customer_code AND n.sku_code=b.sku_code) ";
+    }
    /** Ranking consumes the 30-day aggregate, not chart day-span counts. Its 7/30-day
     * window is contained in this scan, so the same unit proof covers both windows. */
    private static String rankingCte(){
@@ -7,4 +7,5 @@ import java.time.LocalDateTime;
 public record AnalyticsSqlRequest(String account,String warehouse,String customer,String sku,long salesRun,long inventoryRun,
    LocalDate from,LocalDate until,LocalDate anchor,LocalDate windowFrom,LocalDate completeFrom,LocalDate completeUntil,
    LocalDateTime coveredFrom,LocalDateTime coveredUntil,boolean salesKnown,boolean formalSales,boolean completeWindow,
-    boolean complete30,boolean stockFresh,long offset,int size,String kind) {}
+    boolean complete30,boolean stockFresh,long offset,int size,String kind,
+    java.util.List<com.baiyue.lastleg.selfoperated.analytics.model.SelfopSalesCoverage.Interval> coverage) {}
@@ -105,7 +105,7 @@ public class CollectionQueryService {
                String freshness=published==null?"NOT_COLLECTED":scope.staleAfterSeconds()==null||scope.staleAfterSeconds()<=0?"UNKNOWN"
                        :observation==null?"UNKNOWN":!clock.instant().isBefore(observation.plusSeconds(scope.staleAfterSeconds()))?"STALE":"FRESH";
                rows.add(new CollectionResponses.Warehouse(warehouse,warehouse,family,enabled,publication,blocked,
-                        enabled&&"production".equals(config.getEnvironment())?nextSchedule(warehouse,family):null,clock.instant(),scope==null?null:utc(scope.initialFromUtc()),
+                        enabled&&("production".equals(config.getEnvironment())||continuousBusiness())?nextSchedule(warehouse,family):null,clock.instant(),scope==null?null:utc(scope.initialFromUtc()),
                        scope==null?null:utc(scope.initialUntilUtc()),initial,published==null?null:utc(published.coveredFromUtc()),
                        published==null?null:utc(published.coveredUntilUtc()),last==null?null:last.taskNo(),last==null||context.latest().get(last.id())==null?null:utc(context.latest().get(last.id()).finishedAtUtc()),
                        published==null?null:utc(published.publishedAtUtc()),"INVENTORY".equals(family)&&published!=null?utc(published.fetchStartedAtUtc()):null,
@@ -234,7 +234,12 @@ public class CollectionQueryService {
                resultState,t.resultPublishedRunId(),utc(t.nextRetryAtUtc()),
                blocked==null?List.of("RETRY"):List.of(),blocked,utc(t.progressUpdatedAtUtc()),progress(t.progressJson()),
                validation(root==null?t:root),attempt,dependency,waiting,
-                "WAITING_CONFIRMED".equals(dependency)?parent.taskNo():t.taskNo(),stageEstimate(t,latest,attempt,waiting));
+                "WAITING_CONFIRMED".equals(dependency)?parent.taskNo():t.taskNo(),stageEstimate(t,latest,attempt,waiting),utc(t.createdAt()),taskElapsed(t));
+    }
+    private Long taskElapsed(Task task) {
+        Instant start=utc(task.createdAt());
+        Instant end=ACTIVE.contains(task.status())?clock.instant():utc(task.updatedAt());
+        return start==null||end==null||end.isBefore(start)?null:Duration.between(start,end).getSeconds();
    }
    private String dependencyState(Task task,Task parent) {
        if(task.upstreamTaskId()==null)return "WAITING_SOURCE".equals(task.status())?"WAITING_UNCONFIRMED":"NONE";
@@ -280,7 +285,7 @@ public class CollectionQueryService {
        if(!"WAREHOUSE_SYNC".equals(task.taskRole()))return "ACCOUNT_SHARED_TASK";
        String disabled=executionBlock(task.warehouseCode(),protection);if(disabled!=null)return disabled;
        if(!"BUSINESS".equals(config.getCollectionMode()))return "COLLECTION_MODE_NOT_BUSINESS";
-        if("test".equals(config.getEnvironment())&&"SCHEDULED".equals(task.taskType()))return "TEST_MANUAL_TASK_REQUIRED";
+        if("test".equals(config.getEnvironment())&&!continuousBusiness()&&"SCHEDULED".equals(task.taskType()))return "TEST_MANUAL_TASK_REQUIRED";
        if("test".equals(config.getEnvironment())) {
            String budgetBlock=businessBudgetBlock(task);if(budgetBlock!=null)return budgetBlock;
        }
@@ -295,13 +300,16 @@ public class CollectionQueryService {
        JsonNode budget=boundedObject(task.businessRootBudgetJson());
        Long maximum=number(budget,"maxHttp"),reserved=number(budget,"http"),confirmed=number(budget,"confirmedHttp");
        Instant started=budgetInstant(budget,"startedAtUtc"),deadline=budgetInstant(budget,"deadlineAtUtc");
+        boolean continuous=budget!=null&&budget.path("continuous").isBoolean()&&budget.path("continuous").booleanValue();
+        if(continuous&&!continuousBusiness())return "CONTINUOUS_COLLECTION_DISABLED";
        if(budget==null||maximum==null||maximum<0||maximum>Integer.MAX_VALUE||reserved==null||reserved>Integer.MAX_VALUE
-                ||confirmed==null||confirmed>reserved||started==null||deadline==null||!deadline.isAfter(started))
+                ||confirmed==null||confirmed>reserved||started==null
+                ||(continuous?budget.hasNonNull("deadlineAtUtc"):deadline==null||!deadline.isAfter(started)))
            return "BUSINESS_SESSION_BUDGET_MISSING";
        JsonNode stopped=budget.get("stopReason");
        if(stopped!=null&&!stopped.isNull())
            return "BUSINESS_SESSION_STOPPED";
-        if(!clock.instant().isBefore(deadline)&&!Boolean.TRUE.equals(task.businessLocalWorkReady()))return "BUSINESS_SESSION_DEADLINE";
+        if(!continuous&&!clock.instant().isBefore(deadline)&&!Boolean.TRUE.equals(task.businessLocalWorkReady()))return "BUSINESS_SESSION_DEADLINE";
        return null;
    }
    private CollectionResponses.Attempt attemptView(Attempt run,Task task,boolean retryAllowed) {
@@ -346,14 +354,21 @@ public class CollectionQueryService {
        if(protection.accountPauseReason()!=null&&!protection.accountPauseReason().isBlank())return "ACCOUNT_PAUSED";
        if(protection.accountCooldownUntil()!=null&&clock.instant().isBefore(protection.accountCooldownUntil()))return "ACCOUNT_COOLDOWN";
        if(!config.isExclusiveHandoffVerified()||!config.isPriorEgressClosed()||!config.isInflightDrained()||!config.isTransferredBudgetVerified()
-                ||config.getExclusiveGrantExpiresAtUtc()==null||!clock.instant().isBefore(utc(config.getExclusiveGrantExpiresAtUtc())))return "EXCLUSIVE_GRANT_NOT_VALID";
+                ||(!continuousBusiness()&&(config.getExclusiveGrantExpiresAtUtc()==null||!clock.instant().isBefore(utc(config.getExclusiveGrantExpiresAtUtc())))))return "EXCLUSIVE_GRANT_NOT_VALID";
        if("BUSINESS".equals(config.getCollectionMode()))return null;
        if(config.getDailyRequestBudget()<=0)return "DAILY_BUDGET_NOT_CONFIGURED";
        if(protection.reservedHttpRequests()!=null&&clock.instant().atZone(BUSINESS_ZONE).toLocalDate().toString().equals(protection.budgetDate())
                &&protection.reservedHttpRequests()>=config.getDailyRequestBudget())return "DAILY_BUDGET_EXHAUSTED";
        return null;
    }
+    private boolean continuousBusiness() {
+        return com.baiyue.lastleg.selfoperated.analytics.model.SelfopCollectionPolicy.continuousBusiness(config);
+    }
    private Instant nextSchedule(String warehouse,String family) {
+        if(continuousBusiness()&&"OUTBOUND".equals(family)) {
+            var slot=clock.instant().atZone(BUSINESS_ZONE).toLocalDate().atStartOfDay(BUSINESS_ZONE).plusMinutes(5);
+            return (slot.toInstant().isAfter(clock.instant())?slot:slot.plusDays(1)).toInstant();
+        }
        var w=config.getWarehouses().get(warehouse);
        try {var next=CronExpression.parse("OUTBOUND".equals(family)?w.getOutboundCron():w.getInventoryCron()).next(clock.instant().atZone(BUSINESS_ZONE));return next==null?null:next.toInstant();}
        catch(IllegalArgumentException ex){return null;}
@@ -31,7 +31,7 @@ public final class CollectionResponses {
            List<String> allowedActions, String actionBlockReason, Instant progressUpdatedAt,
            Map<String,Object> progress, Validation validation, Attempt latestAttempt,
            String dependencyState, String blockingReason, String progressSourceTaskNo,
-            SelfopStageEstimate.Estimate stageEstimate) {}
+            SelfopStageEstimate.Estimate stageEstimate, Instant createdAt, Long elapsedSeconds) {}
    public record TaskDetail(Task task, Task sharedUpstream, Protection protection, Instant observedAt) {}
    public record Attempt(Long runId, String taskNo, Integer attemptNo, String status, String phase,
            String executionPurpose, Instant startedAt, Instant finishedAt, Instant phaseStartedAt,
@@ -22,18 +22,18 @@ public class DashboardQueryService {
            throw new com.baiyue.common.core.exception.BizException("该销量版本尚未生成仓库每日汇总，请重新发布统计后再查询");
        var base=contextService.request(snapshot,"DASHBOARD");LocalDate fixedFrom=c.anchorDate().minusDays(30),fixedUntil=c.anchorDate();
        var p=new AnalyticsSqlRequest(base.account(),base.warehouse(),null,null,base.salesRun(),base.inventoryRun(),fixedFrom,fixedUntil,base.anchor(),fixedFrom,
-            base.completeFrom(),base.completeUntil(),base.coveredFrom(),base.coveredUntil(),base.salesKnown(),base.formalSales(),base.complete30(),base.complete30(),base.stockFresh(),0,10,"DASHBOARD");
+            base.completeFrom(),base.completeUntil(),base.coveredFrom(),base.coveredUntil(),base.salesKnown(),base.formalSales(),base.complete30(),base.complete30(),base.stockFresh(),0,10,"DASHBOARD",base.coverage());
        var o=new AnalyticsInventoryOptions(null,null,null,null,null,null,null,null,null,null);
        var days=mapper.daily(p,c.rangeStartDate(),c.rangeEndDateExclusive());
        var report=mapper.report(p,o).stream().map(DashboardQueryService::normalize).toList();
        Map<String,Object> stats=report.get(0);var ranked=report.stream().filter(row->row.get("rank_code")!=null).map(DashboardQueryService::ranked).toList();
-        var trendPoints=DashboardSalesPolicy.points(c,snapshot.sourceKnown(),c.rangeStartDate(),c.rangeEndDateExclusive(),days);
+        var trendPoints=DashboardSalesPolicy.points(c,snapshot.sourceKnown(),c.rangeStartDate(),c.rangeEndDateExclusive(),days,SalesQueryCoverage.intervals(snapshot.sales()));
        var trendSummary=SalesBucketPolicy.aggregate(trendPoints);
-        var fixedPoints=DashboardSalesPolicy.points(c,snapshot.sourceKnown(),fixedFrom,fixedUntil,days);var fixed=SalesBucketPolicy.aggregate(fixedPoints);
+        var fixedPoints=DashboardSalesPolicy.points(c,snapshot.sourceKnown(),fixedFrom,fixedUntil,days,SalesQueryCoverage.intervals(snapshot.sales()));var fixed=SalesBucketPolicy.aggregate(fixedPoints);
        Map<String,Object> hotStats=stats;List<DashboardReadMapper.Ranked> hotRanked=ranked;
        if(!c.rangeStartDate().equals(fixedFrom)||!c.rangeEndDateExclusive().equals(fixedUntil)||!base.complete30()){
            var hotRequest=new AnalyticsSqlRequest(base.account(),base.warehouse(),null,null,base.salesRun(),base.inventoryRun(),c.rangeStartDate(),c.rangeEndDateExclusive(),base.anchor(),fixedFrom,
-                base.completeFrom(),base.completeUntil(),base.coveredFrom(),base.coveredUntil(),base.salesKnown(),base.formalSales(),base.completeWindow(),base.complete30(),base.stockFresh(),0,10,"HOT_DASHBOARD");
+                base.completeFrom(),base.completeUntil(),base.coveredFrom(),base.coveredUntil(),base.salesKnown(),base.formalSales(),base.completeWindow(),base.complete30(),base.stockFresh(),0,10,"HOT_DASHBOARD",base.coverage());
            var hotReport=mapper.report(hotRequest,o).stream().map(DashboardQueryService::normalize).toList();
            hotStats=hotReport.get(0);hotRanked=hotReport.stream().filter(row->row.get("rank_code")!=null).map(DashboardQueryService::ranked).toList();
        }
@@ -8,20 +8,23 @@ import java.util.*;
 /** Warehouse coverage is independent of SKU existence and inventory observation eligibility. */
 public final class DashboardSalesPolicy {
    private DashboardSalesPolicy(){}
-    public static List<SalesBucketPolicy.Point> points(AnalyticsResponses.Context c,boolean sourceKnown,LocalDate from,LocalDate until,List<DashboardReadMapper.Day> rows){
+    public static List<SalesBucketPolicy.Point> points(AnalyticsResponses.Context c,boolean sourceKnown,LocalDate from,LocalDate until,List<DashboardReadMapper.Day> rows,
+        List<com.baiyue.lastleg.selfoperated.analytics.model.SelfopSalesCoverage.Interval> coverage){
+        var intervals=coverage.isEmpty()?SalesQueryCoverage.scalar(c.salesCoveredFrom()==null?null:LocalDateTime.ofInstant(c.salesCoveredFrom(),ZoneOffset.UTC),
+            c.salesCoveredUntil()==null?null:LocalDateTime.ofInstant(c.salesCoveredUntil(),ZoneOffset.UTC)):coverage;
        Map<LocalDate,DashboardReadMapper.Day> days=new HashMap<>();for(var row:rows)days.put(row.businessDate(),row);
        List<SalesBucketPolicy.Point> result=new ArrayList<>();ZoneId zone=ZoneId.of("Asia/Shanghai");
        for(LocalDate day=from;day.isBefore(until);day=day.plusDays(1)){
            var actual=days.get(day);Instant start=day.atStartOfDay(zone).toInstant(),end=day.plusDays(1).atStartOfDay(zone).toInstant();
-            Instant lower=c.salesCoveredFrom()==null?null:(c.salesCoveredFrom().isAfter(start)?c.salesCoveredFrom():start);
-            Instant upper=c.salesCoveredUntil()==null?null:(c.salesCoveredUntil().isBefore(end)?c.salesCoveredUntil():end);
+            var interval=SalesQueryCoverage.at(intervals,day);
+            Instant lower=interval==null?null:(interval.from().toInstant(ZoneOffset.UTC).isAfter(start)?interval.from().toInstant(ZoneOffset.UTC):start);
+            Instant upper=interval==null?null:(interval.until().toInstant(ZoneOffset.UTC).isBefore(end)?interval.until().toInstant(ZoneOffset.UTC):end);
            boolean intersects=lower!=null&&upper!=null&&lower.isBefore(upper);
            boolean invalid=actual!=null&&actual.invalidRowCount()!=null&&actual.invalidRowCount()>0;
            // Global recency describes how far collection has progressed, not whether an
            // already published historical day has complete coverage. Keep STALE on context.
            boolean trusted=("VALID".equals(c.salesStatus())||"STALE".equals(c.salesStatus()))&&!c.rebuildRequired();
-            boolean complete=sourceKnown&&trusted&&!invalid&&day.isBefore(c.anchorDate())&&c.completeDayFrom()!=null&&c.completeDayUntil()!=null
-                &&!day.isBefore(c.completeDayFrom())&&day.isBefore(c.completeDayUntil())&&intersects&&lower.equals(start)&&upper.equals(end);
+            boolean complete=sourceKnown&&trusted&&!invalid&&day.isBefore(c.anchorDate())&&SalesQueryCoverage.complete(intervals,day,day.plusDays(1));
            BigDecimal known=sourceKnown&&actual!=null?actual.observedQuantity():null;
            if(known==null&&sourceKnown&&!invalid&&(complete||intersects))known=BigDecimal.ZERO;
            String status=complete?"COMPLETE":known!=null?"PARTIAL":"UNKNOWN";List<String> reasons=new ArrayList<>();
@@ -0,0 +1,33 @@
+package com.baiyue.lastleg.selfoperated.analytics.query;
+
+import com.baiyue.lastleg.selfoperated.analytics.model.*;
+import java.time.*;
+import java.util.*;
+
+/** Reads only immutable evidence already stored in the selected published run. */
+public final class SalesQueryCoverage {
+    private SalesQueryCoverage() { }
+    public static List<SelfopSalesCoverage.Interval> intervals(AnalyticsReadRows.Run selected){
+        if(selected==null)return List.of();
+        var run=new SelfopSyncRun();run.setId(selected.id());run.setScopeId(selected.scopeId());run.setScopeHash(selected.scopeHash());run.setScopeSnapshotJson(selected.scopeSnapshotJson());
+        var evidence=SelfopSalesCoverage.evidence(run);
+        if(evidence!=null)return evidence.intervals();
+        // The existing scalar is itself a frozen continuous proof; never extend it from live tasks.
+        return scalar(selected.coveredFromUtc(),selected.coveredUntilUtc());
+    }
+    public static List<SelfopSalesCoverage.Interval> scalar(LocalDateTime from,LocalDateTime until){
+        return from!=null&&until!=null&&from.isBefore(until)?List.of(new SelfopSalesCoverage.Interval(from,until)):List.of();
+    }
+    public static SelfopSalesCoverage.Interval at(List<SelfopSalesCoverage.Interval> intervals,LocalDate date){
+        var start=date.atStartOfDay().minusHours(8);var end=start.plusDays(1);
+        return intervals.stream().filter(i->i.from().isBefore(end)&&i.until().isAfter(start)).findFirst().orElse(null);
+    }
+    public static boolean complete(List<SelfopSalesCoverage.Interval> intervals,LocalDate from,LocalDate until){
+        var start=from.atStartOfDay().minusHours(8);var end=until.atStartOfDay().minusHours(8);
+        return start.isBefore(end)&&intervals.stream().anyMatch(i->!i.from().isAfter(start)&&!i.until().isBefore(end));
+    }
+    public static LocalDate fullFrom(SelfopSalesCoverage.Interval i){
+        if(i==null)return null;var local=i.from().plusHours(8);return local.toLocalTime().equals(LocalTime.MIDNIGHT)?local.toLocalDate():local.toLocalDate().plusDays(1);
+    }
+    public static LocalDate fullUntil(SelfopSalesCoverage.Interval i){return i==null?null:i.until().plusHours(8).toLocalDate();}
+}
@@ -8,6 +8,7 @@ public final class SelfopBusinessBudget {
    /** Recorded field only; BUSINESS has no HTTP-count ceiling. New sessions write zero. */
    public int maxHttp;
    public int http, confirmedHttp;
+    public boolean continuous;
    public LocalDateTime startedAtUtc, deadlineAtUtc;
    public String stopReason;

@@ -20,9 +21,15 @@ public final class SelfopBusinessBudget {
        return SelfopSupport.json(budget);
    }

+    public static String freezeContinuous(LocalDateTime now) {
+        if(now==null)throw new IllegalArgumentException("BUSINESS_SESSION_BUDGET_INVALID");
+        var budget=new SelfopBusinessBudget();budget.startedAtUtc=now;budget.continuous=true;
+        return SelfopSupport.json(budget);
+    }
+
    public void validate() {
        if (http < 0 || confirmedHttp < 0 || confirmedHttp > http
-            || startedAtUtc == null || deadlineAtUtc == null || !deadlineAtUtc.isAfter(startedAtUtc)
+            || startedAtUtc == null || (continuous ? deadlineAtUtc != null : deadlineAtUtc == null || !deadlineAtUtc.isAfter(startedAtUtc))
            || stopReason != null && !stopReason.matches("[A-Z0-9_]{1,64}"))
            throw new EgressRejectedException("BUSINESS_SESSION_BUDGET_INVALID", false, null);
    }
@@ -30,7 +37,7 @@ public final class SelfopBusinessBudget {
    public String limit(LocalDateTime now) {
        validate();
        if (stopReason != null) return stopReason;
-        if (!now.isBefore(deadlineAtUtc)) return "BUSINESS_SESSION_DEADLINE";
+        if (!continuous && !now.isBefore(deadlineAtUtc)) return "BUSINESS_SESSION_DEADLINE";
        return null;
    }
 }
@@ -18,9 +18,9 @@ public class SelfopEgressService {
  if(context==null||context.rootTaskId()==null||context.taskId()==null||context.runId()==null||!Objects.equals(context.sessionId(),"selfop-"+context.rootTaskId()))throw reject("EGRESS_CONTEXT_INVALID");
  LocalDateTime now=now();gate(context.sourceAccount(),context.executionPurpose(),context.exclusiveGrantId(),now);
  if(operation==null||requestHash==null||!requestHash.matches("[a-f0-9]{64}"))throw reject("EGRESS_REQUEST_INVALID");
-  var account=account(context.sourceAccount());var budget=budget(account,now);save(account,budget);var root=root(context.rootTaskId(),context.executionPurpose(),context.sourceAccount());
+  var account=account(context.sourceAccount());var budget=budget(account,now,context.executionPurpose());save(account,budget);var root=root(context.rootTaskId(),context.executionPurpose(),context.sourceAccount());
  var task=mapper.lockTask(context.taskId());var run=mapper.lockRun(context.runId());
-  SelfopBusinessBudget business=testBusiness(context.executionPurpose())?business(root):null;
+  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);
  checkAccount(account,budget,now);
@@ -42,9 +42,9 @@ public class SelfopEgressService {
  if("VALIDATION".equals(context.executionPurpose())&&budget.reservedHttp>=budget.dailyLimit)throw new EgressRejectedException("ACCOUNT_DAILY_BUDGET",false,now.plusHours(8).toLocalDate().plusDays(1).atStartOfDay().minusHours(8));
  var outbound=new EgressContext(context.sourceAccount(),context.executionPurpose(),String.valueOf(context.rootTaskId()),context.sessionId(),UUID.randomUUID().toString(),context.exclusiveGrantId());
  var pending=new SelfopApiBudget.Pending();pending.context=outbound;pending.operation=operation;pending.internalRequestHash=requestHash;pending.taskId=context.taskId();pending.runId=context.runId();pending.expiresAtUtc=now.plusSeconds(Math.max(1,Math.min(120,config.getEgressLeaseSeconds())));
-  if(pending.expiresAtUtc.isAfter(config.getExclusiveGrantExpiresAtUtc()))pending.expiresAtUtc=config.getExclusiveGrantExpiresAtUtc();
+  if(!continuousBusiness(context.executionPurpose())&&pending.expiresAtUtc.isAfter(config.getExclusiveGrantExpiresAtUtc()))pending.expiresAtUtc=config.getExclusiveGrantExpiresAtUtc();
  if(validation!=null&&pending.expiresAtUtc.isAfter(validation.deadlineAtUtc))pending.expiresAtUtc=validation.deadlineAtUtc;
-  if(business!=null&&pending.expiresAtUtc.isAfter(business.deadlineAtUtc))pending.expiresAtUtc=business.deadlineAtUtc;
+  if(business!=null&&business.deadlineAtUtc!=null&&pending.expiresAtUtc.isAfter(business.deadlineAtUtc))pending.expiresAtUtc=business.deadlineAtUtc;
  budget.pending=pending;budget.reservedHttp++;budget.lastRequestAtUtc=now;
  if(validation!=null){validation.consume(operation,warehouseCode,detailKey);root.setValidationBudgetJson(SelfopSupport.json(validation));mapper.saveRoot(root);}
  if(business!=null){business.http++;saveBusiness(root,business);}
@@ -53,15 +53,15 @@ public class SelfopEgressService {
 @Transactional(noRollbackFor=EgressRejectedException.class)
 public AttemptPermit authorize(AuthorizationRequest request){
  requireRequest(request);var c=request.context();var now=now();gate(c.sourceAccount(),c.executionPurpose(),c.exclusiveGrantId(),now);
-  var account=account(c.sourceAccount());var budget=budget(account,now);var pending=pendingForAuthorization(budget,c,request.operation());
+  var account=account(c.sourceAccount());var budget=budget(account,now,c.executionPurpose());var pending=pendingForAuthorization(budget,c,request.operation());
  var root=root(parseId(c.rootTaskId()),c.executionPurpose(),c.sourceAccount());
  var task=mapper.lockTask(pending.taskId);var run=mapper.lockRun(pending.runId);
-  if(testBusiness(c.executionPurpose())){
+  if(hasBusinessSession(root)){
   var b=business(root);verifyBusinessTask(task,root);
   // Authorization preserves the root time/stop fence independently of request accounting.
-   String limit=b.stopReason!=null?b.stopReason:!now.isBefore(b.deadlineAtUtc)?"BUSINESS_SESSION_DEADLINE":null;
+   String limit=b.limit(now);
   if(limit!=null){stopBusiness(root,b,limit,now);throw reject(limit);}
-   if(pending.expiresAtUtc.isAfter(b.deadlineAtUtc))pending.expiresAtUtc=b.deadlineAtUtc;
+   if(b.deadlineAtUtc!=null&&pending.expiresAtUtc.isAfter(b.deadlineAtUtc))pending.expiresAtUtc=b.deadlineAtUtc;
  }
  verify(task,run,root,account,now);
  if(pending.consumed||pending.completed||!now.isBefore(pending.expiresAtUtc))throw reject("EGRESS_PERMIT_REPLAY_OR_EXPIRED");
@@ -74,7 +74,7 @@ public class SelfopEgressService {
 @Transactional(noRollbackFor=EgressRejectedException.class)
 public void complete(CompletionRequest request){
  if(request==null||request.context()==null||request.actualHttpAttempts()<0||request.actualHttpAttempts()>1)throw reject("EGRESS_COMPLETION_INVALID");
-  var c=request.context();var now=now();var account=account(c.sourceAccount());var budget=budget(account,now);var pending=pending(budget,c,request.operation(),request.requestHash());
+  var c=request.context();var now=now();var account=account(c.sourceAccount());var budget=budget(account,now,c.executionPurpose());var pending=pending(budget,c,request.operation(),request.requestHash());
  String completionHash=SelfopSupport.hash(request);
  if(pending.completed&&!pending.abandoned){
   if(!Objects.equals(pending.completionHash,completionHash))throw reject("EGRESS_COMPLETION_CONFLICT");
@@ -85,7 +85,8 @@ public class SelfopEgressService {
  var root=root(parseId(c.rootTaskId()),c.executionPurpose(),c.sourceAccount());
  SelfopValidationBudget validation="VALIDATION".equals(c.executionPurpose())?validation(root):null;
  if(validation!=null){validation.confirmedHttp+=request.actualHttpAttempts();root.setValidationBudgetJson(SelfopSupport.json(validation));mapper.saveRoot(root);}
-  if(testBusiness(c.executionPurpose())){var b=business(root);b.confirmedHttp+=request.actualHttpAttempts();b.validate();saveBusiness(root,b);}
+  // An already authorized response remains accountable after continuous mode is disabled.
+  if(hasBusinessSession(root)){var b=business(root);b.confirmedHttp+=request.actualHttpAttempts();b.validate();saveBusiness(root,b);}
  boolean auth=Integer.valueOf(401).equals(request.upstreamStatus())||Integer.valueOf(403).equals(request.upstreamStatus())
   ||"LINGXING_WMS_AUTHORIZATION_REJECTED".equals(request.errorCode())||"LINGXING_WMS_SIGNATURE_REJECTED".equals(request.errorCode());
  boolean rate=!auth&&(Integer.valueOf(429).equals(request.upstreamStatus())||Integer.valueOf(31001).equals(request.sourceCode())
@@ -126,16 +127,21 @@ public class SelfopEgressService {
  if("OFF".equals(config.getCollectionMode())||!Objects.equals(config.getCollectionMode(),purpose)||!("VALIDATION".equals(purpose)||"BUSINESS".equals(purpose)))throw reject("EGRESS_MODE_REJECTED");
  if("BUSINESS".equals(purpose)&&!("production".equals(config.getEnvironment())||"test".equals(config.getEnvironment())))throw reject("BUSINESS_ENVIRONMENT_REJECTED");
  if(!Objects.equals(config.getSourceAccount(),account)||account==null||account.isBlank())throw reject("SOURCE_ACCOUNT_MISMATCH");
-  if(grant==null||!Objects.equals(grant,config.getExclusiveGrantId())||config.getExclusiveGrantExpiresAtUtc()==null||!now.isBefore(config.getExclusiveGrantExpiresAtUtc())||!config.isExclusiveHandoffVerified()||!config.isPriorEgressClosed()||!config.isInflightDrained()||!config.isTransferredBudgetVerified())throw reject("EXCLUSIVE_HANDOFF_UNCONFIRMED");
+  if(grant==null||grant.isBlank()||!Objects.equals(grant,config.getExclusiveGrantId())||(!continuousBusiness(purpose)&&(config.getExclusiveGrantExpiresAtUtc()==null||!now.isBefore(config.getExclusiveGrantExpiresAtUtc())))||!config.isExclusiveHandoffVerified()||!config.isPriorEgressClosed()||!config.isInflightDrained()||!config.isTransferredBudgetVerified())throw reject("EXCLUSIVE_HANDOFF_UNCONFIRMED");
  if(("VALIDATION".equals(purpose)&&config.getDailyRequestBudget()<1)||config.getBusinessMinRequestIntervalMs()<1||config.getRequestBudget()<1||config.getTimeBudgetSeconds()<1||config.getMaxRateLimitFailures()<1||config.getCooldownSeconds()<1)throw reject("ACCOUNT_BUDGET_UNCONFIGURED");
 }
 private SelfopOutboundDiscoveryScope account(String id){var a=mapper.lockAccount(id);if(a==null)throw reject("EGRESS_ACCOUNT_MISSING");return a;}
- private SelfopApiBudget budget(SelfopOutboundDiscoveryScope account,LocalDateTime now){
+ private SelfopApiBudget budget(SelfopOutboundDiscoveryScope account,LocalDateTime now,String purpose){
  var b=account.getApiBudgetJson()==null?new SelfopApiBudget():SelfopSupport.read(account.getApiBudgetJson(),SelfopApiBudget.class);
  if(b.reservedHttp<0||b.confirmedHttp<0||b.confirmedHttp>b.reservedHttp)throw reject("ACCOUNT_BUDGET_INVALID");
  LocalDate day=now.plusHours(8).toLocalDate();
-  if(!Objects.equals(day,b.budgetDate)){/* unknown in-flight is charged again on new day; cooling survives */b.budgetDate=day;b.dailyLimit=config.getDailyRequestBudget();b.reservedHttp=b.pending!=null&&!b.pending.completed?1:0;b.confirmedHttp=0;}
-  if("BUSINESS".equals(config.getCollectionMode()))b.dailyLimit=0;
+  if(!Objects.equals(day,b.budgetDate)){
+   b.budgetDate=day;
+   // BUSINESS records cumulative consumption, including transferred and uncertain attempts.
+   // Only the finite VALIDATION daily quota rolls over; cooling/pending always survive.
+   if(!"BUSINESS".equals(purpose)){b.dailyLimit=config.getDailyRequestBudget();b.reservedHttp=b.pending!=null&&!b.pending.completed?1:0;b.confirmedHttp=0;}
+  }
+  if("BUSINESS".equals(purpose))b.dailyLimit=0;
  else {if(b.dailyLimit<1)b.dailyLimit=config.getDailyRequestBudget();b.dailyLimit=Math.min(b.dailyLimit,config.getDailyRequestBudget());}
  b.minIntervalMs=Math.max(b.minIntervalMs,config.getBusinessMinRequestIntervalMs());
  b.configHash=SelfopSupport.hash(Arrays.asList(b.dailyLimit,b.minIntervalMs,config.getRequestBudget(),config.getTimeBudgetSeconds()));
@@ -161,20 +167,23 @@ public class SelfopEgressService {
 }
 private SelfopSyncTask root(Long id,String purpose,String account){var t=mapper.lockTask(id);if(t==null||!Objects.equals(t.getSourceAccount(),account)||!Objects.equals(t.getExecutionPurpose(),purpose))throw reject("EGRESS_ROOT_INVALID");if("VALIDATION".equals(purpose)&&!Objects.equals(id,t.getValidationRootTaskId()))throw reject("VALIDATION_ROOT_INVALID");return t;}
 private void verify(SelfopSyncTask task,SelfopSyncRun run,SelfopSyncTask root,SelfopOutboundDiscoveryScope account,LocalDateTime now){
-  if(testBusiness(root.getExecutionPurpose())&&"DEAD".equals(root.getStatus()))throw reject("BUSINESS_SESSION_ROOT_STOPPED");
+  if(hasBusinessSession(root)&&"DEAD".equals(root.getStatus()))throw reject("BUSINESS_SESSION_ROOT_STOPPED");
  if(task==null||run==null||!Objects.equals(run.getTaskId(),task.getId())||!Objects.equals(run.getSourceAccount(),root.getSourceAccount())||!Objects.equals(task.getSourceAccount(),root.getSourceAccount())||!Objects.equals(task.getExecutionPurpose(),root.getExecutionPurpose())||!Objects.equals(run.getExecutionPurpose(),task.getExecutionPurpose())||!"RUNNING".equals(task.getStatus())||!Set.of("FETCHING","VALIDATED").contains(run.getStatus()))throw reject("EGRESS_TASK_NOT_ACTIVE");
  if("VALIDATION".equals(root.getExecutionPurpose())&&(!Objects.equals(task.getValidationRootTaskId(),root.getId())||Set.of("DEAD","SAMPLE_DONE").contains(root.getStatus())))throw reject("VALIDATION_ROOT_STOPPED");
  if("ACCOUNT_DISCOVERY".equals(task.getTaskRole())){if(!Objects.equals(task.getDiscoveryScopeId(),account.getId())||!Objects.equals(account.getFenceToken(),run.getFenceToken())||account.getLeaseUntilUtc()==null||!now.isBefore(account.getLeaseUntilUtc()))throw reject("EGRESS_FENCE_LOST");}
  else if(!Objects.equals(task.getScopeId(),run.getScopeId())||!Objects.equals(task.getWarehouseCode(),run.getWarehouseCode())||mapper.warehouseFence(task.getScopeId(),run.getFenceToken(),task.getId(),now)!=1)throw reject("EGRESS_FENCE_LOST");
 }
 private SelfopValidationBudget validation(SelfopSyncTask root){if(root==null||!"VALIDATION".equals(root.getExecutionPurpose())||!Objects.equals(root.getId(),root.getValidationRootTaskId())||root.getValidationBudgetJson()==null)throw reject("VALIDATION_ROOT_INVALID");var b=SelfopSupport.read(root.getValidationBudgetJson(),SelfopValidationBudget.class);b.validate();return b;}
+ private boolean continuousBusiness(String purpose){return "BUSINESS".equals(purpose)&&config.isContinuousBusinessCollectionEnabled();}
 private boolean testBusiness(String purpose){return "test".equals(config.getEnvironment())&&"BUSINESS".equals(purpose);}
+ private boolean hasBusinessSession(SelfopSyncTask root){return "BUSINESS".equals(root.getExecutionPurpose())&&(testBusiness(root.getExecutionPurpose())||root.getBusinessBudgetJson()!=null||SelfopCollectionPolicy.continuousBusinessRoot(root));}
 private SelfopBusinessBudget business(SelfopSyncTask root){
  if(root==null||!"BUSINESS".equals(root.getExecutionPurpose())||!Objects.equals(root.getId(),root.getBusinessRootTaskId())||root.getBusinessBudgetJson()==null)throw reject("BUSINESS_SESSION_ROOT_INVALID");
  SelfopBusinessBudget b;try{b=SelfopSupport.read(root.getBusinessBudgetJson(),SelfopBusinessBudget.class);}catch(RuntimeException e){throw reject("BUSINESS_SESSION_BUDGET_INVALID");}
-  if(b==null)throw reject("BUSINESS_SESSION_BUDGET_INVALID");b.validate();return b;
+  if(b==null)throw reject("BUSINESS_SESSION_BUDGET_INVALID");b.validate();
+  if(b.continuous&&!SelfopCollectionPolicy.continuousBusinessRoot(root))throw reject("BUSINESS_SESSION_ROOT_INVALID");return b;
 }
- private void verifyBusinessTask(SelfopSyncTask task,SelfopSyncTask root){if(task==null||!Objects.equals(task.getBusinessRootTaskId(),root.getId())||!Objects.equals(task.getSourceAccount(),root.getSourceAccount())||!"BUSINESS".equals(task.getExecutionPurpose()))throw reject("BUSINESS_SESSION_ROOT_INVALID");}
+ private void verifyBusinessTask(SelfopSyncTask task,SelfopSyncTask root){if(task==null||!Objects.equals(task.getBusinessRootTaskId(),root.getId())||!Objects.equals(task.getSourceAccount(),root.getSourceAccount())||!"BUSINESS".equals(task.getExecutionPurpose())||business(root).continuous&&!continuousBusiness(root.getExecutionPurpose()))throw reject("BUSINESS_SESSION_ROOT_INVALID");}
 private void stopBusiness(SelfopSyncTask root,SelfopBusinessBudget b,String reason,LocalDateTime now){
  b.stopReason=reason;root.setStatus("DEAD");root.setNextRetryAtUtc(null);saveBusiness(root,b);
  mapper.terminateBusinessTasks(root.getId(),root.getSourceAccount());mapper.terminateBusinessRuns(root.getId(),root.getSourceAccount(),reason,now);
@@ -37,6 +37,7 @@ public class SelfopExecutionService {
    }
    public void execute(long taskId){
        var task=tasks.selectById(taskId);require(task!=null,"SELFOP_TASK_MISSING");SelfopCollectionPolicy.requireExecution(config,task.getExecutionPurpose());
+        if(SelfopCollectionPolicy.continuousBusiness(config))require(SelfopCollectionPolicy.controlledBusinessTask(task,tasks::selectById,true),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
        if("SUCCEEDED".equals(task.getStatus())||"DEAD".equals(task.getStatus()))return;
        String owner=UUID.randomUUID().toString();Map<Long,SelfopLeaseService.Lease> held=new LinkedHashMap<>();SelfopLeaseService.Lease primary;
        if("ACCOUNT_DISCOVERY".equals(task.getTaskRole())){primary=leases.acquireDiscovery(task.getDiscoveryScopeId(),owner);held.put(primary.scopeId(),primary);}
@@ -34,9 +34,9 @@ public class SelfopSalesPublisher {
        require(locks.barriers(scope.getSourceAccount(),scope.getWarehouseCode())==0,"HALF_APPLIED_FACTS");
        require(Objects.equals(run.getCalculatedFactRevision(),scope.getFactRevision()),"SELFOP_CANDIDATE_OBSOLETE");
        require(run.getBuildStartedAtUtc()!=null&&aggregates.observationsChangedSince(scope.getSourceAccount(),scope.getWarehouseCode(),run.getBuildStartedAtUtc())==0,"SELFOP_CANDIDATE_OBSERVATION_CHANGED");
-        require("COMPLETE".equals(run.getCoverageStatus())&&run.getCoveredFromUtc()!=null&&run.getCoveredUntilUtc()!=null,"SELFOP_COVERAGE_UNPROVEN");
        var task=locks.lockTask(run.getTaskId());require("APPLIED".equals(task.getApplyState())||"REBUILD".equals(task.getTaskType()),"SELFOP_TASK_NOT_APPLIED");
        SelfopCollectionPolicy.requireBusiness(task);
+        SelfopSalesCoverage.requireTask(run,task);
        if(scope.getPublishedRunId()!=null)runs.update(null,new com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper<SelfopSyncRun>()
            .eq("id",scope.getPublishedRunId()).isNull("superseded_at_utc").set("superseded_at_utc",now).set("updated_at",now));
        scope.setPublishedRunId(runId);scope.setPublishedFactRevision(run.getCalculatedFactRevision());scope.setRebuildRequired(0);
@@ -59,6 +59,7 @@ public class SelfopSalesPublisher {
        require(Objects.equals(scope.getPublishedRunId(),publishedId)&&"PUBLISHED".equals(published.getStatus())&&Objects.equals(published.getCalculatedFactRevision(),scope.getFactRevision())
            &&locks.barriers(scope.getSourceAccount(),scope.getWarehouseCode())==0,"SELFOP_REUSE_OBSOLETE");
        require("APPLIED".equals(task.getApplyState())||"REBUILD".equals(task.getTaskType()),"SELFOP_TASK_NOT_APPLIED");
+        SelfopSalesCoverage.requireTask(published,task);
        require(run.getFetchStartedAtUtc()!=null&&aggregates.observationsChangedSince(scope.getSourceAccount(),scope.getWarehouseCode(),run.getFetchStartedAtUtc())==0,"SELFOP_REUSE_OBSERVATION_CHANGED");
        var now=utcNow();run.setCalculatedFactRevision(published.getCalculatedFactRevision());SelfopProgressProjection.freezeRun(task,run);run.setStatus("REUSED");run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run);
        task.setResultPublishedRunId(publishedId);task.setStatus("SUCCEEDED");task.setUpdatedAt(now);tasks.update(null,new com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper<SelfopSyncTask>()
@@ -32,6 +32,15 @@ public class SelfopSchedulePlanner {
    public void tick(){
        if("OFF".equals(config.getCollectionMode()))return;
        if("BUSINESS".equals(config.getCollectionMode()))SelfopCollectionPolicy.requireExecution(config,"BUSINESS");
+        if(SelfopCollectionPolicy.continuousBusiness(config)){
+            planContinuous(utcNow());
+            // One dispatch per tick: after a safe yield, re-plan and re-read priority.
+            for(var task:taskService.due(1)){
+                if(!SelfopCollectionPolicy.controlledBusinessTask(task,tasks::selectById,true))continue;
+                try{execution.execute(task.getId());}catch(IllegalStateException leaseBusy){/* Durable lease owner continues. */}
+            }
+            return;
+        }
        List<SelfopSyncScope> configured;
        if("VALIDATION".equals(config.getCollectionMode())){taskService.initializeAccount();configured=List.of();}
        else if("test".equals(config.getEnvironment()))configured=List.of();
@@ -48,6 +57,80 @@ public class SelfopSchedulePlanner {
            try{execution.execute(task.getId());}catch(IllegalStateException leaseBusy){/* Another worker owns the durable lease. */}
        }
    }
+    public void planContinuous(LocalDateTime now){
+        require(SelfopCollectionPolicy.continuousBusiness(config),"SELFOP_CONTINUOUS_COLLECTION_DISABLED");
+        require("Asia/Shanghai".equals(config.getSourceQueryTimezone()),"SELFOP_SOURCE_TIMEZONE_REQUIRED");
+        require(!config.getWarehouses().isEmpty(),"SELFOP_WAREHOUSE_UNCONFIGURED");
+        var account=taskService.initializeAccount();
+        tx.executeWithoutResult(s->{
+            locks.lockDiscovery(account.getId());
+            // Read existing scopes only. Pool creation must never move the published coverage anchor.
+            var configured=scopes.selectList(new QueryWrapper<SelfopSyncScope>().eq("source_account",config.getSourceAccount())
+                .eq("data_family","OUTBOUND").in("warehouse_code",config.getWarehouses().keySet()));
+            require(configured.size()==config.getWarehouses().size()&&configured.stream().allMatch(scope->
+                "Asia/Shanghai".equals(scope.getSourceTimezone())&&"Asia/Shanghai".equals(scope.getBusinessTimezone())),"SELFOP_SCOPE_MISSING");
+            var local=now.plusHours(8);var today=local.toLocalDate();
+            var last=today.minusDays(local.toLocalTime().isBefore(LocalTime.of(0,5))?2:1);
+            var first=today.minusMonths(2);
+            String digest=hash(config.getWarehouses());
+            var roots=tasks.selectList(new QueryWrapper<SelfopSyncTask>().eq("source_account",config.getSourceAccount())
+                .eq("execution_purpose","BUSINESS").eq("task_role","ACCOUNT_DISCOVERY").eq("task_type","BACKFILL")
+                .eq("discovery_scope_id",account.getId()).isNull("upstream_task_id").isNull("validation_root_task_id")
+                .ge("query_from_utc",first.atStartOfDay().minusHours(8)).le("query_until_utc",last.plusDays(1).atStartOfDay().minusHours(8)));
+            var acceptedHashes=new HashSet<String>();acceptedHashes.add(digest);
+            configured.forEach(scope->acceptedHashes.add(scope.getScopeHash()));
+            for(var date=last;!date.isBefore(first);date=date.minusDays(1)){
+                var from=date.atStartOfDay().minusHours(8);var until=from.plusDays(1);
+                String key="continuous:outbound:"+date+":"+digest.substring(0,12);
+                // Examine every root: an old DEAD attempt must not mask a later successful/manual root.
+                boolean exists=roots.stream().anyMatch(root->key.equals(root.getRequestKey())||
+                    (from.equals(root.getQueryFromUtc())&&until.equals(root.getQueryUntilUtc())&&acceptedHashes.contains(root.getScopeHash())
+                     &&Set.of("READY","RUNNING","RETRY_WAIT","SUCCEEDED").contains(root.getStatus())&&wholeDayQuery(root,from)));
+                if(exists)continue;
+                var task=new SelfopSyncTask();task.setSourceAccount(config.getSourceAccount());task.setExecutionPurpose("BUSINESS");
+                task.setTaskRole("ACCOUNT_DISCOVERY");task.setTaskType("BACKFILL");task.setDiscoveryScopeId(account.getId());
+                task.setRequestedBy(SelfopCollectionPolicy.CONTINUOUS_PLANNER);task.setScopeHash(digest);task.setRequestKey(key);
+                task.setQueryFromUtc(from);task.setQueryUntilUtc(until);
+                task.setQueryScopeJson(json(new SelfopDiscoveryService.Query(from.plusHours(8).toString(),until.plusHours(8).toString(),null,null,null)));
+                task.setRequestHash(hash(Arrays.asList(task.getQueryScopeJson(),digest)));taskService.create(task);
+            }
+            planContinuousInventory(now,today);
+        });
+    }
+    private void planContinuousInventory(LocalDateTime now,LocalDate today){
+        var inventory=scopes.selectList(new QueryWrapper<SelfopSyncScope>().eq("source_account",config.getSourceAccount())
+            .eq("data_family","INVENTORY").in("warehouse_code",config.getWarehouses().keySet()));
+        require(inventory.size()==config.getWarehouses().size(),"SELFOP_SCOPE_MISSING");
+        for(var scope:inventory){
+            require("INVENTORY".equals(scope.getDataFamily())&&"Asia/Shanghai".equals(scope.getBusinessTimezone()),"SELFOP_INVENTORY_SCOPE_INVALID");
+            var midnight=today.atStartOfDay(ZoneId.of("Asia/Shanghai"));
+            var slot=CronExpression.parse(config.getWarehouses().get(scope.getWarehouseCode()).getInventoryCron()).next(midnight.minusNanos(1));
+            if(slot==null||!slot.toLocalDate().equals(today))continue;
+            var planned=slot.withZoneSameInstant(ZoneOffset.UTC).toLocalDateTime();
+            if(planned.isAfter(now))continue;
+            String key="continuous:inventory:"+scope.getWarehouseCode()+":"+planned+":"+scope.getScopeHash().substring(0,12);
+            var existing=tasks.selectList(new QueryWrapper<SelfopSyncTask>().eq("source_account",config.getSourceAccount())
+                .eq("scope_id",scope.getId()).eq("scope_hash",scope.getScopeHash()).eq("execution_purpose","BUSINESS")
+                .eq("task_role","WAREHOUSE_SYNC").eq("task_type","SCHEDULED")
+                .ge("planned_at_utc",midnight.withZoneSameInstant(ZoneOffset.UTC).toLocalDateTime())
+                .lt("planned_at_utc",midnight.plusDays(1).withZoneSameInstant(ZoneOffset.UTC).toLocalDateTime()));
+            if(existing.stream().anyMatch(t->key.equals(t.getRequestKey())||
+                (Objects.equals(t.getScopeId(),scope.getId())&&Objects.equals(t.getScopeHash(),scope.getScopeHash())
+                 &&t.getPlannedAtUtc()!=null&&t.getPlannedAtUtc().plusHours(8).toLocalDate().equals(today)
+                 &&SelfopCollectionPolicy.continuousBusinessRoot(t))))continue;
+            var task=new SelfopSyncTask();task.setSourceAccount(config.getSourceAccount());task.setExecutionPurpose("BUSINESS");
+            task.setTaskRole("WAREHOUSE_SYNC");task.setTaskType("SCHEDULED");task.setScopeId(scope.getId());task.setWarehouseCode(scope.getWarehouseCode());
+            task.setRequestedBy(SelfopCollectionPolicy.CONTINUOUS_PLANNER);task.setScopeHash(scope.getScopeHash());task.setRequestKey(key);
+            task.setPlannedAtUtc(planned);task.setNextRetryAtUtc(planned);task.setQueryScopeJson(scope.getScopeJson());
+            task.setRequestHash(hash(Arrays.asList(scope.getScopeHash(),planned,task.getQueryScopeJson())));taskService.create(task);
+        }
+    }
+    private static boolean wholeDayQuery(SelfopSyncTask root,LocalDateTime from){
+        try{var q=read(root.getQueryScopeJson(),SelfopDiscoveryService.Query.class);
+            return q!=null&&q.deliveryNo()==null&&q.customerCode()==null&&q.parentTaskId()==null
+                &&from.plusHours(8).equals(LocalDateTime.parse(q.startTime()))&&from.plusDays(1).plusHours(8).equals(LocalDateTime.parse(q.endTime()));
+        }catch(RuntimeException invalid){return false;}
+    }
    public void plan(SelfopSyncScope preview,LocalDateTime now){
        SelfopCollectionPolicy.requireAutomaticPlanning(config);
        tx.executeWithoutResult(s->{var scope=locks.lockScope(preview.getId());ZoneId business=ZoneId.of(scope.getBusinessTimezone());
@@ -110,15 +110,23 @@ public class SelfopTaskService {
        }
        if(testBusiness(task)) {
            require(task.getBusinessRootTaskId()==null&&task.getBusinessBudgetJson()==null,"BUSINESS_SESSION_IDENTITY_INVALID");
-            if(task.getRequestKey().startsWith("manual:")&&task.getUpstreamTaskId()==null) {
-                require(controlledBusinessTask(task,tasks::selectById),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
+            if(continuousBusiness(config)&&continuousBusinessRoot(task)){
+                if("WAREHOUSE_SYNC".equals(task.getTaskRole())){
+                    var scope=scopes.selectById(task.getScopeId());
+                    require(scope!=null&&"INVENTORY".equals(scope.getDataFamily())&&Objects.equals(scope.getSourceAccount(),task.getSourceAccount())
+                        &&Objects.equals(scope.getWarehouseCode(),task.getWarehouseCode())&&Objects.equals(scope.getScopeHash(),task.getScopeHash())
+                        &&Objects.equals(scope.getScopeJson(),task.getQueryScopeJson()),"SELFOP_INVENTORY_SCOPE_INVALID");
+                }
+                task.setBusinessRootTaskId(task.getId());task.setBusinessBudgetJson(SelfopBusinessBudget.freezeContinuous(now));
+            }else if(task.getRequestKey().startsWith("manual:")&&task.getUpstreamTaskId()==null) {
+                require(controlledBusinessTask(task,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
                task.setBusinessRootTaskId(task.getId());
                task.setBusinessBudgetJson(SelfopBusinessBudget.freeze(now,config.getTimeBudgetSeconds()));
            }else {
                var parent=businessParent(task);
-                require(parent!=null&&Objects.equals(parent.getSourceAccount(),task.getSourceAccount())&&controlledBusinessTask(parent,tasks::selectById),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
+                require(parent!=null&&Objects.equals(parent.getSourceAccount(),task.getSourceAccount())&&controlledBusinessTask(parent,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
                if("MANUAL_RETRY".equals(task.getTaskType()))require("SUCCEEDED".equals(task.getStatus())&&task.getRequestKey().startsWith("retry:")&&task.getUpstreamTaskId()!=null,"SELFOP_TEST_MANUAL_TASK_REQUIRED");
-                require("MANUAL_RETRY".equals(task.getTaskType())||controlledBusinessTask(task,tasks::selectById),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
+                require("MANUAL_RETRY".equals(task.getTaskType())||controlledBusinessTask(task,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
                businessGuard(parent,!"WAREHOUSE_SYNC".equals(task.getTaskRole()));task.setBusinessRootTaskId(parent.getBusinessRootTaskId());
            }
        }
@@ -136,7 +144,7 @@ public class SelfopTaskService {
    @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
    public SelfopSyncRun begin(long taskId,SelfopLeaseService.Lease lease) {
        lockBusinessAccount(taskId);leases.verify(lease);
-        if("test".equals(config.getEnvironment())){var preview=tasks.selectById(taskId);businessGuard(preview,!localBusinessWork(preview));}
+        if("test".equals(config.getEnvironment())||continuousBusiness(config)){var preview=tasks.selectById(taskId);businessGuard(preview,!localBusinessWork(preview));}
        var task=locks.lockTask(taskId);require(task!=null&&!"SUCCEEDED".equals(task.getStatus())&&!"DEAD".equals(task.getStatus()),"SELFOP_TASK_NOT_EXECUTABLE");
        checkTaskLease(task,lease);
        requireExecution(config,task.getExecutionPurpose());
@@ -190,6 +198,7 @@ public class SelfopTaskService {
        run.setErrorMessage("自营同步失败，需核对运行记录");run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run);
        // A yielded execution is not a failed request; finite failures are counted separately.
        long failures=runs.selectCount(new QueryWrapper<SelfopSyncRun>().eq("task_id",taskId).eq("status","FAILED"));
+        if(task.getBusinessRootTaskId()!=null&&continuousBusinessRoot(tasks.selectById(task.getBusinessRootTaskId())))retryable=false;
        task.setStatus(retryable&&failures<config.getMaxAttempts()?"RETRY_WAIT":"DEAD");
        if("APPLYING".equals(task.getApplyState())) task.setApplyState("BLOCKED");
        if(task.getScopeId()!=null&&"INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())) {
@@ -200,7 +209,7 @@ public class SelfopTaskService {
            task.setRoutingManifestJson(null);
            tasks.update(null,new UpdateWrapper<SelfopSyncTask>().eq("id",taskId).set("routing_manifest_json",null));
        }
-        task.setNextRetryAtUtc(now.plusSeconds(Math.min(3600,5L<<Math.min(9,(int)failures))));task.setUpdatedAt(now);tasks.updateById(task);
+        task.setNextRetryAtUtc("DEAD".equals(task.getStatus())?null:now.plusSeconds(Math.min(3600,5L<<Math.min(9,(int)failures))));task.setUpdatedAt(now);tasks.updateById(task);
    }
    @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
    public SelfopSyncTask retry(long id,String requestKey,String operator,String reason) {
@@ -228,10 +237,12 @@ public class SelfopTaskService {
        audit.setRequestKey("retry:"+requestKey);audit.setRequestHash(hash(List.of(id,requestKey,reason)));audit.setTaskType("MANUAL_RETRY");audit.setScopeHash(original.getScopeHash());
        audit.setQueryScopeJson(json(Map.of("resumesTaskNo",original.getTaskNo(),"reason",reason)));audit.setStatus("SUCCEEDED");audit.setApplyState("NOT_APPLICABLE");audit.setRequestedBy(operator);
        create(audit);
-        original.setStatus("READY");original.setNextRetryAtUtc(utcNow());original.setRequestedBy(operator);original.setUpdatedAt(utcNow());tasks.updateById(original);
+        original.setStatus("READY");original.setNextRetryAtUtc(utcNow());
+        if(!continuousBusinessRoot(original))original.setRequestedBy(operator);
+        original.setUpdatedAt(utcNow());tasks.updateById(original);
        return original;
    }
-    public List<SelfopSyncTask> due(int limit){return locks.due(utcNow(),limit,config.getCollectionMode());}
+    public List<SelfopSyncTask> due(int limit){return continuousBusiness(config)?locks.dueContinuous(utcNow(),limit,config.getSourceAccount()):locks.due(utcNow(),limit,config.getCollectionMode());}
    @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
    public void candidateFinished(long taskId,long runId,SelfopLeaseService.Lease lease) {
        lockBusinessAccount(taskId);leases.verify(lease);guardBeforeTaskLock(taskId,false);var task=locks.lockTask(taskId);var run=locks.lockRun(runId);
@@ -243,7 +254,7 @@ public class SelfopTaskService {
    }
    @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
    public void waitForBudget(long taskId,long runId,LocalDateTime until,SelfopLeaseService.Lease lease) {
-        lockBusinessAccount(taskId);leases.verify(lease);if("test".equals(config.getEnvironment())){var preview=tasks.selectById(taskId);businessGuard(preview,!localBusinessWork(preview));}var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run);
+        lockBusinessAccount(taskId);leases.verify(lease);if("test".equals(config.getEnvironment())||continuousBusiness(config)){var preview=tasks.selectById(taskId);businessGuard(preview,!localBusinessWork(preview));}var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run);
        SelfopProgressProjection.freezeRun(task,run);run.setLastPhase(run.getStatus());run.setStatus("YIELDED");run.setFinishedAtUtc(utcNow());run.setUpdatedAt(utcNow());runs.updateById(run);
        task.setStatus("RETRY_WAIT");task.setNextRetryAtUtc(until);task.setUpdatedAt(utcNow());tasks.updateById(task);
    }
@@ -252,7 +263,7 @@ public class SelfopTaskService {
    @Transactional(propagation=org.springframework.transaction.annotation.Propagation.MANDATORY)
    public CollectionCheckpoint checkpoint(SelfopSyncTask observedTask,SelfopSyncRun observedRun,SelfopLeaseService.Lease lease) {
        long taskId=observedTask.getId();lockBusinessAccount(taskId);leases.verify(lease);
-        if("test".equals(config.getEnvironment())) {
+        if("test".equals(config.getEnvironment())||continuousBusiness(config)) {
            var preview=tasks.selectById(taskId);require(preview!=null,"SELFOP_TASK_MISSING");businessGuard(preview,false,false);
        }
        var task=locks.lockTask(taskId);var run=locks.lockRun(observedRun.getId());checkRunLease(task,run,lease);
@@ -266,21 +277,21 @@ public class SelfopTaskService {
        BusinessSessionStoppedException(String reason){super("本次测试采集会话已结束或预算无效，请核对记录后新建有界任务");this.reason=reason;}
        public String reason(){return reason;}
    }
-    private boolean testBusiness(SelfopSyncTask task){return task!=null&&"test".equals(config.getEnvironment())&&"BUSINESS".equals(task.getExecutionPurpose());}
+    private boolean testBusiness(SelfopSyncTask task){return task!=null&&("test".equals(config.getEnvironment())||continuousBusiness(config))&&"BUSINESS".equals(task.getExecutionPurpose());}
    private SelfopSyncTask businessParent(SelfopSyncTask task){
        Long id=task.getUpstreamTaskId();
        if(id==null&&task.getQueryScopeJson()!=null){var query=read(task.getQueryScopeJson(),Map.class);if(query.get("parentTaskId") instanceof Number n)id=n.longValue();}
        return id==null?null:tasks.selectById(id);
    }
    private void lockBusinessAccount(long id){
-        if(!"test".equals(config.getEnvironment()))return;
+        if(!"test".equals(config.getEnvironment())&&!continuousBusiness(config))return;
        var task=tasks.selectById(id);if(!testBusiness(task))return;
        require(task.getBusinessRootTaskId()!=null,"BUSINESS_SESSION_ROOT_REQUIRED");
        var account=accounts.selectOne(new QueryWrapper<SelfopOutboundDiscoveryScope>().eq("source_account",task.getSourceAccount()));
        require(account!=null,"SELFOP_SCOPE_MISSING");locks.lockDiscovery(account.getId());
    }
    private void guardBeforeTaskLock(long id,boolean checkDeadline){
-        if(!"test".equals(config.getEnvironment()))return;
+        if(!"test".equals(config.getEnvironment())&&!continuousBusiness(config))return;
        var task=tasks.selectById(id);businessGuard(task,checkDeadline);
    }
    private boolean localBusinessWork(SelfopSyncTask task){
@@ -301,22 +312,22 @@ public class SelfopTaskService {
    private void businessGuard(SelfopSyncTask task,boolean checkDeadline){businessGuard(task,checkDeadline,true);}
    private void businessGuard(SelfopSyncTask task,boolean checkDeadline,boolean terminate){
        if(!testBusiness(task))return;
-        require(controlledBusinessTask(task,tasks::selectById),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
+        require(controlledBusinessTask(task,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
        require(task.getBusinessRootTaskId()!=null,"BUSINESS_SESSION_ROOT_REQUIRED");
        var ancestor=task;
-        while(!ancestor.getRequestKey().startsWith("manual:"))ancestor=businessParent(ancestor);
+        while(!ancestor.getRequestKey().startsWith("manual:")&&!continuousBusinessRoot(ancestor))ancestor=businessParent(ancestor);
        require(Objects.equals(ancestor.getId(),task.getBusinessRootTaskId()),"BUSINESS_SESSION_ROOT_INVALID");
        // Serialize with egress in account -> root -> task order, before taking any task row lock.
        var account=accounts.selectOne(new QueryWrapper<SelfopOutboundDiscoveryScope>().eq("source_account",task.getSourceAccount()));
        require(account!=null,"SELFOP_SCOPE_MISSING");locks.lockDiscovery(account.getId());
        var root=locks.lockTask(task.getBusinessRootTaskId());
        require(root!=null&&Objects.equals(root.getId(),root.getBusinessRootTaskId())&&Objects.equals(root.getSourceAccount(),task.getSourceAccount())
-            &&root.getRequestKey()!=null&&root.getRequestKey().startsWith("manual:")&&controlledBusinessTask(root,tasks::selectById),"BUSINESS_SESSION_ROOT_INVALID");
+            &&root.getRequestKey()!=null&&(root.getRequestKey().startsWith("manual:")||continuousBusinessRoot(root))&&controlledBusinessTask(root,tasks::selectById,continuousBusiness(config)),"BUSINESS_SESSION_ROOT_INVALID");
        require(root.getBusinessBudgetJson()!=null,"BUSINESS_SESSION_BUDGET_INVALID");
        var budget=read(root.getBusinessBudgetJson(),SelfopBusinessBudget.class);budget.validate();
-        // HTTP exhaustion is enforced at egress, allowing the final permitted response to finish local work.
+        require(!budget.continuous||(continuousBusiness(config)&&continuousBusinessRoot(root)),"SELFOP_CONTINUOUS_COLLECTION_DISABLED");
        String reason=budget.stopReason;
-        if(reason==null&&checkDeadline&&!utcNow().isBefore(budget.deadlineAtUtc))reason="BUSINESS_SESSION_DEADLINE";
+        if(reason==null&&checkDeadline&&!budget.continuous&&!utcNow().isBefore(budget.deadlineAtUtc))reason="BUSINESS_SESSION_DEADLINE";
        if(reason==null&&"DEAD".equals(root.getStatus()))reason="BUSINESS_SESSION_STOPPED";
        if(reason==null)return;
        if(!terminate)throw new BusinessSessionStoppedException(reason);