feat: expose reliable collection activity and stage estimates

2026-10-13 16:50:53 +08:00
parent 03601ab0aa
commit 146d468ff4
15 changed files with 557 additions and 46 deletions
@@ -14,16 +14,29 @@ public final class SelfopProgressProjection {
        "VALIDATION", "MERGE", "BUILD", "PUBLISH", "WAITING", "SAMPLE_DONE", "FAILED");
    /** Pure projection for callers already holding the scope/task/run locks and fence. */
    public static void applyCheckpoint(SelfopSyncTask task,SelfopSyncRun run,String phase,String unit,long completed,Long total,int scan) {
+        applyCheckpoint(task,run,phase,unit,completed,total,scan,true);
+    }
+    public static void applyCheckpoint(SelfopSyncTask task,SelfopSyncRun run,String phase,String unit,long completed,Long total,int scan,boolean realWork) {
        require(PHASES.contains(phase),"SELFOP_PROGRESS_PHASE_INVALID");
        require(Set.of("PAGE","ORDER","INVENTORY_ROW","BUSINESS_KEY","BATCH","NONE").contains(unit),"SELFOP_PROGRESS_UNIT_INVALID");
        require(completed>=0 && scan>=0 && (total==null || total>=completed),"SELFOP_PROGRESS_COUNTS_INVALID");
        Map<String,Object> p=progress(task);
-        if(!Objects.equals(number(p.get("attemptNo")),run.getAttemptNo().longValue()))p=new LinkedHashMap<>();
+        if(!Objects.equals(number(p.get("attemptNo")),run.getAttemptNo().longValue())){
+            var prior=p;p=new LinkedHashMap<>();
+            if(prior.get("localWorkReady") instanceof Boolean
+                &&Objects.equals(prior.get("localWorkReadyScopeHash"),task.getScopeHash())
+                &&Objects.equals(prior.get("localWorkReadyUpstreamTaskId"),task.getUpstreamTaskId()==null?null:task.getUpstreamTaskId().toString()))
+                for(String key:List.of("localWorkReady","localWorkReadyScopeHash","localWorkReadyUpstreamTaskId"))p.put(key,prior.get(key));
+        }
+        String priorIdentity=SelfopStageEstimate.identity(p);
        String countScope=scope(task); Object started=p.get("stageStartedAt");
        if(!phase.equals(p.get("phase")) || started==null)started=instant(utcNow());
+        p.put("taskNo",task.getTaskNo());
        p.put("attemptNo",run.getAttemptNo()); p.put("phase",phase); p.put("progressScope",countScope);
        p.put("stageStartedAt",started); p.put("unit",unit); p.put("completed",completed);
        p.put("total",total); p.put("totalKnown",total!=null); p.put("scanRound",scan); p.put("countBasis",Set.of("INVENTORY","BUILD").contains(phase)?"ATTEMPT_ONLY":"TASK_CUMULATIVE"); p.put("sliceFrom",instant(task.getQueryFromUtc())); p.put("sliceUntil",instant(task.getQueryUntilUtc())); p.put("stageFinishedAt",null);
+        if(!priorIdentity.equals(SelfopStageEstimate.identity(p)))p.put("stageStartedAt",instant(utcNow()));
+        SelfopStageEstimate.checkpoint(p,utcNow().toInstant(ZoneOffset.UTC),realWork);
        // A zero denominator is meaningful evidence, but cannot produce a percentage.
        p.put("phasePercent",total==null||total==0?null:BigDecimal.valueOf(completed).multiply(BigDecimal.valueOf(100))
            .divide(BigDecimal.valueOf(total),1,RoundingMode.DOWN));
@@ -42,6 +55,13 @@ public final class SelfopProgressProjection {

    /** Pure terminal snapshot; historical attempt evidence is immutable. */
    public static void freezeRun(SelfopSyncTask task,SelfopSyncRun run) {
+        // A yielded inventory attempt resumes with the same attempt number. Its live
+        // sampling baseline must not survive a budget pause or execution interruption.
+        if(task.getProgressJson()!=null){
+            var live=progress(task);
+            if(Objects.equals(number(live.get("attemptNo")),run.getAttemptNo().longValue())&&live.remove("etaSamples")!=null)
+                task.setProgressJson(bounded(live));
+        }
        if(run.getProgressSnapshotJson()!=null)return;
        if(task.getProgressJson()==null) {
            var fallback=new LinkedHashMap<String,Object>();fallback.put("attemptNo",run.getAttemptNo());
@@ -50,16 +70,50 @@ public final class SelfopProgressProjection {
            fallback.put("unit","NONE");fallback.put("completed",0);fallback.put("total",null);fallback.put("totalKnown",false);
            fallback.put("sliceFrom",instant(task.getQueryFromUtc()));fallback.put("sliceUntil",instant(task.getQueryUntilUtc()));
            fallback.put("stageStartedAt",instant(run.getFetchStartedAtUtc()));fallback.put("stageFinishedAt",null);fallback.put("stages",List.of());
+            fallback.put("taskNo",task.getTaskNo());fallback.put("summaryAsOf",instant(utcNow()));
            run.setProgressSnapshotJson(bounded(fallback));
            return;
        }
        var p=progress(task);
        // A later attempt's projection must never be copied into an earlier historical run.
        require(Objects.equals(number(p.get("attemptNo")),run.getAttemptNo().longValue()),"SELFOP_PROGRESS_ATTEMPT_MISMATCH");
+        p.remove("etaSamples");p.put("summaryAsOf",instant(utcNow()));
        run.setProgressSnapshotJson(bounded(p));run.setUpdatedAt(utcNow());
    }

-    private static String bounded(Map<String,Object> p) {String value=json(p);require(value.getBytes(StandardCharsets.UTF_8).length<=8192,"SELFOP_PROGRESS_TOO_LARGE");return value;}
+    /** Route commits this proof together with its immutable effective-stage references. */
+    public static void localWorkReady(SelfopSyncTask task,boolean ready) {
+        var p=progress(task);p.put("localWorkReady",ready);
+        p.put("localWorkReadyScopeHash",task.getScopeHash());
+        p.put("localWorkReadyUpstreamTaskId",task.getUpstreamTaskId()==null?null:task.getUpstreamTaskId().toString());
+        task.setProgressJson(bounded(p));
+    }
+
+    private static String bounded(Map<String,Object> p) {
+        // Reserve the terminal timestamp before accepting a live checkpoint. Optional
+        // ETA and derived display fields never displace business/audit checkpoints.
+        int target=p.containsKey("summaryAsOf")?8192:8192-64;
+        String value=json(p);
+        if(bytes(value)>target){p.remove("etaSamples");value=json(p);}
+        if(bytes(value)>target){p.remove("phasePercent");value=json(p);}
+        if(bytes(value)>target&&p.get("stages") instanceof List<?> prior){
+            var stages=new ArrayList<Map<String,Object>>();
+            for(Object entry:prior){var stage=map(entry);stage.remove("phasePercent");stages.add(stage);}
+            p.put("stages",stages);value=json(p);
+            // The current top-level stage clock remains available. Duplicated stage
+            // display clocks can be omitted while retaining every stage count/identity.
+            if(bytes(value)>target){
+                for(var stage:stages){stage.remove("stageStartedAt");stage.remove("stageFinishedAt");}
+                value=json(p);
+            }
+        }
+        // A pre-existing legacy checkpoint may have used the whole 8 KiB with no
+        // optional display fields. Missing optional snapshot time is preferable to
+        // failing its business termination; never invent a replacement timestamp.
+        if(bytes(value)>8192&&p.containsKey("summaryAsOf")){p.remove("summaryAsOf");value=json(p);}
+        require(bytes(value)<=8192,"SELFOP_PROGRESS_TOO_LARGE");return value;
+    }
+    private static int bytes(String value){return value.getBytes(StandardCharsets.UTF_8).length;}
    private static String scope(SelfopSyncTask task){return "ACCOUNT_DISCOVERY".equals(task.getTaskRole())?"ACCOUNT_SHARED":"WAREHOUSE";}
    @SuppressWarnings("unchecked") private static Map<String,Object> progress(SelfopSyncTask task){return task.getProgressJson()==null?new LinkedHashMap<>():read(task.getProgressJson(),LinkedHashMap.class);}
    @SuppressWarnings("unchecked") private static Map<String,Object> map(Object value){return value instanceof Map<?,?>?new LinkedHashMap<>((Map<String,Object>)value):new LinkedHashMap<>();}
@@ -0,0 +1,114 @@
+package com.baiyue.lastleg.selfoperated.analytics.model;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.*;
+import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*;
+
+/** Bounded committed-checkpoint sampling; reads never manufacture progress. */
+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;
+    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) { }
+
+    public static String identity(Map<String,Object> p) {
+        var values=new ArrayList<Object>();
+        for(String key:List.of("taskNo","attemptNo","phase","scanRound","progressScope","unit",
+            "countBasis","sliceFrom","sliceUntil","total"))values.add(p.get(key));
+        return hash(values);
+    }
+
+    /** A reuse/uncertain checkpoint resets the baseline, never contributes throughput. */
+    public static void checkpoint(Map<String,Object> p,Instant now,boolean realWork) {
+        try {
+            Long completed=number(p.get("completed")),total=number(p.get("total"));
+            if(total==null||total<=0||completed==null||completed<0||completed>=total||!Boolean.TRUE.equals(p.get("totalKnown"))||"NONE".equals(p.get("unit"))) {
+                p.remove("etaSamples");return;
+            }
+            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()
+                ||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();
+            points.removeIf(point->second-point.get(0)>WINDOW_SECONDS);
+            long advances=points.isEmpty()?0:points.get(points.size()-1).get(2);
+            long lastAdvance=numberOr(old.get("a"),second);
+            if(points.isEmpty()) {points.add(List.of(second,completed,0L));lastAdvance=second;}
+            else {
+                var last=points.get(points.size()-1);
+                if(completed>last.get(1)) {advances++;lastAdvance=second;}
+                var point=List.of(second,completed,advances);
+                if(last.get(0)/BUCKET_SECONDS==second/BUCKET_SECONDS)points.set(points.size()-1,point);
+                else points.add(point);
+            }
+            while(points.size()>MAX_POINTS)points.remove(0);
+            var sample=new LinkedHashMap<String,Object>();sample.put("i",id);sample.put("a",lastAdvance);sample.put("p",points);
+            while(json(sample).getBytes(StandardCharsets.UTF_8).length>MAX_BYTES&&points.size()>1)points.remove(0);
+            if(json(sample).getBytes(StandardCharsets.UTF_8).length<=MAX_BYTES)p.put("etaSamples",sample);
+            else p.remove("etaSamples");
+        } catch(RuntimeException malformedOptionalSample) {p.remove("etaSamples");}
+    }
+
+    public static Estimate estimate(Map<String,Object> p,Instant now,boolean healthy,boolean paused,boolean terminal) {
+        String id=identity(p);Long completed=number(p.get("completed")),total=number(p.get("total"));
+        String reason=terminal?"TERMINAL":paused?"PAUSED":!healthy?"HEALTH_UNCONFIRMED":
+            total==null||!Boolean.TRUE.equals(p.get("totalKnown"))?"TOTAL_UNKNOWN":completed==null||completed<0||total<completed||"NONE".equals(p.get("unit"))?"INVALID_PROGRESS":
+            total.equals(completed)?"COUNT_COMPLETE":null;
+        Instant from=null;Long delta=null;BigDecimal rate=null;Long remaining=null;Instant until=null;
+        try {
+            var sample=map(p.get("etaSamples"));var points=points(sample.get("p"));
+            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);
+                if(points.isEmpty()||second-lastAdvance>=STALE_SECONDS)reason="NO_RECENT_PROGRESS";
+                else {
+                    var first=points.get(0);var last=points.get(points.size()-1);
+                    long elapsed=second-first.get(0),advances=last.get(2)-first.get(2);
+                    if(second<last.get(0)||!completed.equals(last.get(1)))reason="INVALID_PROGRESS";
+                    else if(elapsed<MIN_SECONDS||advances<MIN_ADVANCES)reason="SAMPLING";
+                    else {
+                        delta=last.get(1)-first.get(1);
+                        if(delta<=0)reason="NO_RECENT_PROGRESS";
+                        else {
+                            from=Instant.ofEpochSecond(first.get(0));
+                            rate=BigDecimal.valueOf(delta).divide(BigDecimal.valueOf(elapsed),6,RoundingMode.DOWN);
+                            remaining=BigDecimal.valueOf(total-completed).multiply(BigDecimal.valueOf(elapsed))
+                                .divide(BigDecimal.valueOf(delta),0,RoundingMode.CEILING).longValueExact();
+                            until=Instant.ofEpochSecond(Math.min(second+VALID_SECONDS,Math.min(lastAdvance+STALE_SECONDS,first.get(0)+WINDOW_SECONDS)));
+                            if(!until.isAfter(now))reason="SAMPLE_EXPIRED";
+                        }
+                    }
+                }
+            }
+        } catch(RuntimeException malformedOptionalSample) {reason="INVALID_SAMPLE";}
+        if(reason!=null)return new Estimate(paused&&!terminal?"PAUSED":"UNAVAILABLE",id,completed,total,null,null,null,null,null,now,null,reason);
+        return new Estimate("AVAILABLE",id,completed,total,from,now,delta,rate,remaining,now,until,null);
+    }
+    private static Long number(Object value){
+        if(!(value instanceof Number))return null;
+        try{return new BigDecimal(value.toString()).longValueExact();}catch(RuntimeException invalid){return null;}
+    }
+    private static long numberOr(Object value,long fallback){Long n=number(value);return n==null?fallback:n;}
+    @SuppressWarnings("unchecked") private static Map<String,Object> map(Object value){return value instanceof Map<?,?>?(Map<String,Object>)value:Map.of();}
+    private static List<List<Long>> points(Object value) {
+        var result=new ArrayList<List<Long>>();
+        if(value instanceof List<?> list)for(Object item:list) {
+            if(!(item instanceof List<?> point)||point.size()!=3||result.size()>=MAX_POINTS)throw new IllegalArgumentException("Invalid bounded sample");
+            var parsed=List.of(Objects.requireNonNull(number(point.get(0))),Objects.requireNonNull(number(point.get(1))),Objects.requireNonNull(number(point.get(2))));
+            if(parsed.stream().anyMatch(n->n<0))throw new IllegalArgumentException("Negative sample");
+            if(!result.isEmpty()){
+                var prior=result.get(result.size()-1);
+                if(parsed.get(0)<=prior.get(0)||parsed.get(1)<prior.get(1)||parsed.get(2)<prior.get(2))throw new IllegalArgumentException("Nonmonotonic sample");
+            }
+            result.add(parsed);
+        }
+        return result;
+    }
+}
@@ -2,6 +2,7 @@ package com.baiyue.lastleg.selfoperated.analytics.query;

 import com.baiyue.common.core.exception.BizException;
 import com.baiyue.lastleg.selfoperated.analytics.model.SelfopAnalyticsProperties;
+import com.baiyue.lastleg.selfoperated.analytics.model.SelfopStageEstimate;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.nio.charset.StandardCharsets;
@@ -23,6 +24,7 @@ public class CollectionQueryService {
    private final Clock clock;
    private static final ZoneId BUSINESS_ZONE = ZoneId.of("Asia/Shanghai");
    private static final Set<String> TERMINAL = Set.of("ROUTED","PUBLISHED","REUSED","YIELDED","FAILED","ABORTED","SAMPLE_DONE");
+    private static final Set<String> ACTIVE = Set.of("RUNNING","READY","QUEUED","WAITING","WAITING_SOURCE","RETRY_WAIT","YIELDED");
    private static final Set<String> PROGRESS_FIELDS = Set.of("progressScope","countBasis","attemptNo","scanRound","phase",
            "completed","total","totalKnown","unit","sliceFrom","sliceUntil","stageStartedAt","stageFinishedAt",
            "validationState","invalidCount","unknownCount","pendingReviewCount","countUnit","summaryAsOf","directoryComplete");
@@ -76,14 +78,20 @@ public class CollectionQueryService {
            for(String family:List.of("OUTBOUND","INVENTORY")) {
                if(query.dataFamily()!=null&&!family.equals(query.dataFamily()))continue;
                Scope scope=persisted.get(warehouse+":"+family);
-                Task current=currentTasks.stream().filter(t->warehouse.equals(t.warehouseCode())&&family.equals(t.dataFamily())).findFirst().orElse(null);
+                List<Task> candidates=new ArrayList<>(currentTasks.stream().filter(t->warehouse.equals(t.warehouseCode())&&family.equals(t.dataFamily())).toList());
                Task sampled=validationTasks.stream().filter(t->warehouse.equals(t.warehouseCode())).findFirst().orElse(null);
-                if(sampled!=null&&(current==null||sampled.createdAt()!=null&&current.createdAt()!=null&&sampled.createdAt().isAfter(current.createdAt())))current=sampled;
+                if(sampled!=null)candidates.add(sampled);
+                candidates.sort(Comparator.comparingInt((Task t)->"RUNNING".equals(t.status())?0:ACTIVE.contains(t.status())?1:2)
+                        .thenComparing(Task::createdAt,Comparator.nullsLast(Comparator.reverseOrder()))
+                        .thenComparing(Task::id,Comparator.nullsLast(Comparator.reverseOrder())));
+                Task current=candidates.isEmpty()?null:candidates.get(0);
+                List<CollectionResponses.Task> activity=candidates.stream().filter(t->ACTIVE.contains(t.status())).limit(8)
+                        .map(t->taskView(t,scope,protection,context,family)).toList();
                Task last=successes.stream().filter(t->warehouse.equals(t.warehouseCode())&&family.equals(t.dataFamily())).findFirst().orElse(null);
                Attempt published=scope==null?null:context.published().get(scope.publishedRunId());
                if(published!=null&&(!warehouse.equals(published.warehouseCode())||!Objects.equals(scope.id(),published.scopeId())
                        ||!Objects.equals(scope.scopeHash(),published.scopeHash())))published=null;
-                Task shared="OUTBOUND".equals(family)&&sharedWindowRelated(active,scope)?active:null;
+                Task shared=current==null?null:relatedUpstream(current,context);
                String blocked=executionBlock(warehouse,protection);
                boolean enabled=blocked==null;
                boolean publication="BUSINESS".equals(config.getCollectionMode())&&config.isExecutionEnabled()
@@ -104,10 +112,11 @@ public class CollectionQueryService {
                        "INVENTORY".equals(family)&&published!=null?utc(published.fetchFinishedAtUtc()):null,freshness,
                        "OUTBOUND".equals(family)&&scope!=null&&(Integer.valueOf(1).equals(scope.rebuildRequired())||barriers.contains(warehouse)),reviewStatus(scope),
                        current==null?null:taskView(current,scope,protection,context,family),shared==null?null:taskView(shared,null,protection,context),protection,
-                        "INVENTORY".equals(family)?inventorySamples.get(warehouse):null));
+                        "INVENTORY".equals(family)?inventorySamples.get(warehouse):null,activity));
            }
        }
-        return new CollectionResponses.Overview(query.sourceAccount(),config.getEnvironment(),config.getCollectionMode(),clock.instant(),rows);
+        return new CollectionResponses.Overview(query.sourceAccount(),config.getEnvironment(),config.getCollectionMode(),clock.instant(),rows,
+                active!=null&&ACTIVE.contains(active.status())&&"BUSINESS".equals(active.executionPurpose())?taskView(active,null,protection,context):null);
    }

    private Map<String,CollectionResponses.InventoryDailySamples> inventorySamples(CollectionQuery query) {
@@ -181,14 +190,6 @@ public class CollectionQueryService {
        for(Scope scope:mapper.scopes(q.sourceAccount(),q.warehouseCode()))scopes.put(scope.dataFamily(),scope);
        return scopes;
    }
-    private boolean sharedWindowRelated(Task active,Scope scope) {
-        if(active==null||scope==null||active.queryFromUtc()==null||active.queryUntilUtc()==null)return false;
-        return overlaps(active.queryFromUtc(),active.queryUntilUtc(),scope.initialFromUtc(),scope.initialUntilUtc())
-                ||overlaps(active.queryFromUtc(),active.queryUntilUtc(),scope.discoveryFromUtc(),scope.discoveryUntilUtc());
-    }
-    private static boolean overlaps(LocalDateTime from,LocalDateTime until,LocalDateTime otherFrom,LocalDateTime otherUntil) {
-        return otherFrom!=null&&otherUntil!=null&&from.isBefore(otherUntil)&&until.isAfter(otherFrom);
-    }
    private String reviewStatus(Scope scope) {
        if(scope==null||scope.reviewTargetDays()==null||scope.reviewTargetDays()<1)return "UNKNOWN";
        if(scope.knownOrderRoundFinishedAtUtc()==null||scope.historyRoundFinishedAtUtc()==null)return "PENDING";
@@ -199,6 +200,7 @@ public class CollectionQueryService {
        if(task.upstreamTaskId()==null)return null;
        Task parent=context.tasks().get(task.upstreamTaskId());
        return parent!=null&&"ACCOUNT_DISCOVERY".equals(parent.taskRole())&&Objects.equals(task.sourceAccount(),parent.sourceAccount())
+                &&Objects.equals(task.executionPurpose(),parent.executionPurpose())
                &&(!"VALIDATION".equals(parent.executionPurpose())||validationAccountRoot(parent)&&Objects.equals(task.warehouseCode(),parent.warehouseCode()))?parent:null;
    }
    private static boolean validationAccountRoot(Task task) {
@@ -223,11 +225,55 @@ public class CollectionQueryService {
        String resultState=t.resultPublishedRunId()==null?"NOT_PUBLISHED":result==null||scope==null||!Objects.equals(scope.id(),result.scopeId())
                ||!Objects.equals(t.warehouseCode(),result.warehouseCode())||!Objects.equals(t.scopeHash(),result.scopeHash())?"RESULT_UNAVAILABLE"
                :latest!=null&&"REUSED".equals(latest.status())?"REUSED":latest!=null&&"PUBLISHED".equals(latest.status())?"PUBLISHED":"RESULT_AVAILABLE";
+        Task parent=relatedUpstream(t,context);
+        String dependency=dependencyState(t,parent);
+        String waiting=blockingReason(t,dependency,protection);
+        CollectionResponses.Attempt attempt=latest==null?null:attemptView(latest,t,blocked==null);
        return new CollectionResponses.Task(t.taskNo(),t.taskType(),validationAccountRoot(t)&&family!=null?family:t.dataFamily(),t.warehouseCode(),t.taskRole(),t.executionPurpose(),
                utc(t.queryFromUtc()),utc(t.queryUntilUtc()),t.status(),t.applyState(),t.attemptCount(),latest==null?null:latest.attemptNo(),
                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),latest==null?null:attemptView(latest,t,blocked==null));
+                validation(root==null?t:root),attempt,dependency,waiting,
+                "WAITING_CONFIRMED".equals(dependency)?parent.taskNo():t.taskNo(),stageEstimate(t,latest,attempt,waiting));
+    }
+    private String dependencyState(Task task,Task parent) {
+        if(task.upstreamTaskId()==null)return "WAITING_SOURCE".equals(task.status())?"WAITING_UNCONFIRMED":"NONE";
+        if(parent==null)return "WAITING_UNCONFIRMED";
+        if("SUCCEEDED".equals(parent.status()))return "SATISFIED";
+        if(Set.of("DEAD","FAILED","ABORTED","CANCELLED").contains(parent.status()))return "FAILED";
+        return "WAITING_CONFIRMED";
+    }
+    private String blockingReason(Task task,String dependency,CollectionResponses.Protection protection) {
+        if(!ACTIVE.contains(task.status()))return null;
+        String disabled=executionBlock(task.warehouseCode(),protection);
+        if(disabled!=null)return disabled;
+        if("test".equals(config.getEnvironment())&&"BUSINESS".equals(task.executionPurpose())) {
+            String budget=businessBudgetBlock(task);if(budget!=null)return budget;
+        }
+        if("WAITING_UNCONFIRMED".equals(dependency))return "UPSTREAM_UNCONFIRMED";
+        if("WAITING_CONFIRMED".equals(dependency))return "UPSTREAM_NOT_READY";
+        if("FAILED".equals(dependency))return "UPSTREAM_FAILED";
+        if("RETRY_WAIT".equals(task.status()))return "RETRY_SCHEDULED";
+        if("YIELDED".equals(task.status()))return "EXECUTION_YIELDED";
+        if(Set.of("READY","QUEUED","WAITING").contains(task.status()))return "QUEUED";
+        return null;
+    }
+    private SelfopStageEstimate.Estimate stageEstimate(Task task,Attempt run,CollectionResponses.Attempt attempt,String blocked) {
+        boolean terminal=!ACTIVE.contains(task.status())||run!=null&&TERMINAL.contains(run.status());
+        String source=task.progressJson();
+        JsonNode bounded=boundedObject(source);
+        Map<String,Object> raw=bounded==null?new LinkedHashMap<>():json.convertValue(bounded,new com.fasterxml.jackson.core.type.TypeReference<LinkedHashMap<String,Object>>(){});
+        raw.putAll(progress(source));
+        raw.put("taskNo",task.taskNo());
+        boolean healthy="RUNNING".equals(task.status())&&attempt!=null&&"HEALTHY".equals(attempt.executionHealth())
+                &&Objects.equals(task.attemptCount(),attempt.attemptNo())&&Objects.equals(number(bounded,"attemptNo"),attempt.attemptNo()==null?null:attempt.attemptNo().longValue());
+        var estimate=SelfopStageEstimate.estimate(raw,clock.instant(),healthy,blocked!=null,terminal);
+        if("AVAILABLE".equals(estimate.status())&&attempt.heartbeatAt()!=null) {
+            Instant healthUntil=attempt.heartbeatAt().plusSeconds(config.getLeaseSeconds());
+            if(healthUntil.isBefore(estimate.validUntil()))return new SelfopStageEstimate.Estimate(estimate.status(),estimate.progressIdentity(),estimate.completed(),estimate.total(),
+                    estimate.sampleFrom(),estimate.sampleUntil(),estimate.completedDelta(),estimate.ratePerSecond(),estimate.remainingSeconds(),estimate.estimatedAt(),healthUntil,null);
+        }
+        return estimate;
    }
    private String retryBlock(Task task,Scope scope,CollectionResponses.Protection protection,Context context) {
        if(!"BUSINESS".equals(task.executionPurpose()))return "VALIDATION_NOT_RETRYABLE";
@@ -294,9 +340,9 @@ public class CollectionQueryService {
        if(!Set.of("BUSINESS","VALIDATION").contains(config.getCollectionMode()))return "COLLECTION_MODE_INVALID";
        if("BUSINESS".equals(config.getCollectionMode())&&!com.baiyue.lastleg.selfoperated.analytics.model.SelfopCollectionPolicy.businessEnvironment(config))return "BUSINESS_ENVIRONMENT_REJECTED";
        if(!config.isExecutionEnabled())return "EXECUTION_DISABLED";
-        if(!config.getWarehouses().containsKey(warehouse))return "WAREHOUSE_NOT_CONFIGURED";
-        var w=config.getWarehouses().get(warehouse);
-        if(!"Asia/Shanghai".equals(w.getBusinessTimezone())||!"Asia/Shanghai".equals(w.getSourceTimezone())||!"PCS".equals(config.getQuantityUnit()))return "TIMEZONE_OR_UNIT_NOT_CONFIRMED";
+        if(warehouse!=null&&!config.getWarehouses().containsKey(warehouse))return "WAREHOUSE_NOT_CONFIGURED";
+        var w=warehouse==null?null:config.getWarehouses().get(warehouse);
+        if(w!=null&&(!"Asia/Shanghai".equals(w.getBusinessTimezone())||!"Asia/Shanghai".equals(w.getSourceTimezone()))||!"PCS".equals(config.getQuantityUnit()))return "TIMEZONE_OR_UNIT_NOT_CONFIRMED";
        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()
@@ -39,16 +39,14 @@ public interface CollectionReadMapper {
            + " AND JSON_TYPE(JSON_EXTRACT(t.query_scope_json,'$.warehouses[0]'))='STRING')";
    String TASK_WAREHOUSE = "CASE WHEN " + VALIDATION_ROOT
            + " THEN JSON_UNQUOTE(JSON_EXTRACT(t.query_scope_json,'$.warehouses[0]')) ELSE t.warehouse_code END";
-    String TASK_FAMILY = "CASE WHEN " + VALIDATION_ROOT + " THEN 'OUTBOUND' ELSE s.data_family END";
+    String TASK_FAMILY = "CASE WHEN t.task_role='ACCOUNT_DISCOVERY' THEN 'OUTBOUND' ELSE s.data_family END";
    String TASK_MEMBERSHIP = "((t.task_role='WAREHOUSE_SYNC' AND s.id IS NOT NULL) OR " + VALIDATION_ROOT + ")";
    String BUSINESS_LOCAL_WORK_READY = "CASE WHEN t.task_role='WAREHOUSE_SYNC' AND s.data_family='OUTBOUND'"
            + " AND (t.task_type='REBUILD' OR t.apply_state='APPLIED' OR ("
-            + "JSON_EXTRACT(t.routing_manifest_json,'$.directoryComplete')=CAST('true' AS JSON)"
-            + " AND JSON_TYPE(JSON_EXTRACT(t.routing_manifest_json,'$.objects'))='OBJECT'"
-            + " AND NOT EXISTS (SELECT 1 FROM JSON_TABLE(t.routing_manifest_json,'$.objects.*'"
-            + " COLUMNS(entry JSON PATH '$')) manifest_item"
-            + " WHERE COALESCE(JSON_TYPE(JSON_EXTRACT(manifest_item.entry,'$.effectiveStageId')),'NULL') NOT IN ('INTEGER')"
-            + " OR (CAST(JSON_UNQUOTE(JSON_EXTRACT(manifest_item.entry,'$.effectiveStageId')) AS DECIMAL(65,0)) BETWEEN 1 AND 9223372036854775807) IS NOT TRUE)))"
+            + "OCTET_LENGTH(t.progress_json) BETWEEN 0 AND 8192 AND JSON_EXTRACT(t.progress_json,'$.localWorkReady')=CAST('true' AS JSON)"
+            + " AND JSON_UNQUOTE(JSON_EXTRACT(t.progress_json,'$.localWorkReadyScopeHash'))=t.scope_hash"
+            + " AND ((t.upstream_task_id IS NULL AND JSON_TYPE(JSON_EXTRACT(t.progress_json,'$.localWorkReadyUpstreamTaskId'))='NULL')"
+            + " OR (t.upstream_task_id IS NOT NULL AND JSON_UNQUOTE(JSON_EXTRACT(t.progress_json,'$.localWorkReadyUpstreamTaskId'))=CAST(t.upstream_task_id AS CHAR)))))"
            + " THEN TRUE ELSE FALSE END";
    String TASK = "t.id,t.task_no,t.source_account," + TASK_WAREHOUSE + " AS warehouse_code," + TASK_FAMILY + " AS data_family,t.task_role,t.task_type,"
            + "t.execution_purpose,t.scope_hash,t.status,t.apply_state,t.attempt_count,t.query_from_utc,t.query_until_utc,"
@@ -123,14 +121,17 @@ public interface CollectionReadMapper {
            + " WHERE r.source_account=#{account} AND r.status='PUBLISHED' AND r.execution_purpose='BUSINESS' AND r.run_role='WAREHOUSE_SYNC'"
            + " AND r.id IN <foreach collection='ids' item='id' open='(' close=')' separator=','>#{id}</foreach></script>")
    List<Attempt> publishedRuns(@Param("account") String account,@Param("ids") List<Long> ids);
+    String ACTIVE_STATUS = "('RUNNING','READY','QUEUED','WAITING','WAITING_SOURCE','RETRY_WAIT','YIELDED')";
+    String ACTIVITY_ORDER = "CASE WHEN status='RUNNING' THEN 0 WHEN status IN " + ACTIVE_STATUS + " THEN 1 ELSE 2 END";
    @Select("<script>SELECT " + TASK + TASK_FROM
-            + " JOIN (SELECT id,ROW_NUMBER() OVER(PARTITION BY scope_id ORDER BY created_at DESC,id DESC) rn FROM ll_selfop_sync_task"
-            + " WHERE scope_id IN <foreach collection='ids' item='id' open='(' close=')' separator=','>#{id}</foreach>) ranked"
-            + " ON ranked.id=t.id AND ranked.rn=1 WHERE t.source_account=#{account}</script>")
+            + " JOIN (SELECT id,ROW_NUMBER() OVER(PARTITION BY scope_id ORDER BY " + ACTIVITY_ORDER + ",created_at DESC,id DESC) rn FROM ll_selfop_sync_task"
+            + " WHERE source_account=#{account} AND task_role='WAREHOUSE_SYNC' AND scope_id IN <foreach collection='ids' item='id' open='(' close=')' separator=','>#{id}</foreach>) ranked"
+            + " ON ranked.id=t.id AND ranked.rn &lt;= 8 WHERE t.source_account=#{account} AND (t.status IN " + ACTIVE_STATUS + " OR ranked.rn=1)"
+            + " ORDER BY t.scope_id,ranked.rn</script>")
    List<Task> currentTasks(@Param("account") String account,@Param("ids") List<Long> ids);
    @Select("<script>SELECT " + TASK + TASK_FROM
            + " JOIN (SELECT t.id,ROW_NUMBER() OVER(PARTITION BY " + TASK_WAREHOUSE
-            + " ORDER BY t.created_at DESC,t.id DESC) rn FROM ll_selfop_sync_task t"
+            + " ORDER BY " + ACTIVITY_ORDER + ",t.created_at DESC,t.id DESC) rn FROM ll_selfop_sync_task t"
            + " WHERE t.source_account=#{account} AND " + VALIDATION_ROOT + " AND (" + TASK_WAREHOUSE + ") IN"
            + " <foreach collection='warehouses' item='w' open='(' close=')' separator=','>#{w}</foreach>) ranked"
            + " ON ranked.id=t.id AND ranked.rn=1 WHERE t.source_account=#{account} AND " + VALIDATION_ROOT + "</script>")
@@ -141,7 +142,7 @@ public interface CollectionReadMapper {
            + " <foreach collection='ids' item='id' open='(' close=')' separator=','>#{id}</foreach>) ranked"
            + " ON ranked.id=t.id AND ranked.rn=1 WHERE t.source_account=#{account}</script>")
    List<Task> lastSuccessfulTasks(@Param("account") String account,@Param("ids") List<Long> ids);
-    // The one account active task is separately related to warehouses by actual persisted windows.
+    // Account activity is independent; only upstream_task_id proves a warehouse dependency.
    @Select("SELECT " + TASK + " FROM ll_selfop_outbound_discovery_scope d JOIN ll_selfop_sync_task t ON t.id=d.active_task_id"
            + " AND t.source_account=d.source_account LEFT JOIN ll_selfop_sync_scope s ON s.id=t.scope_id"
            + BUSINESS_ROOT_JOIN
@@ -5,13 +5,14 @@ import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.util.List;
 import java.util.Map;
+import com.baiyue.lastleg.selfoperated.analytics.model.SelfopStageEstimate;

 public final class CollectionResponses {
    private CollectionResponses() {}
    public record Page<T>(int current, int size, long total, Instant listUpperCreatedAt,
            Long listUpperId, Instant observedAt, List<T> records) {}
    public record Overview(String sourceAccount, String environmentLabel, String collectionMode,
-            Instant observedAt, List<Warehouse> warehouses) {}
+            Instant observedAt, List<Warehouse> warehouses, Task accountActivity) {}
    public record Warehouse(String warehouseCode, String warehouseName, String dataFamily,
            boolean executionEnabled, boolean publicationEnabled, String disabledReason,
            Instant nextScheduledAt, Instant observedAt, Instant initialFrom, Instant initialUntil,
@@ -19,7 +20,8 @@ public final class CollectionResponses {
            String lastSuccessfulTaskNo, Instant lastSucceededAt, Instant lastPublishedAt,
            Instant inventoryObservedFrom, Instant inventoryObservedUntil, String dataFreshness,
            boolean rebuildRequired, String revisionReviewStatus, Task currentTask,
-            Task sharedUpstream, Protection protection, InventoryDailySamples inventoryDailySamples) {}
+            Task sharedUpstream, Protection protection, InventoryDailySamples inventoryDailySamples,
+            List<Task> activeTasks) {}
    public record InventoryDailySamples(LocalDate windowFrom, LocalDate windowUntil, int expectedDays,
            int sampleCount, List<LocalDate> missingDates, LocalDate accumulationStartedOn) {}
    public record Task(String taskNo, String taskType, String dataFamily, String warehouseCode,
@@ -27,7 +29,9 @@ public final class CollectionResponses {
            String taskStatus, String applyState, Integer attemptCount, Integer latestAttemptNo,
            String resultPublicationState, Long resultPublishedRunId, Instant nextRetryAt,
            List<String> allowedActions, String actionBlockReason, Instant progressUpdatedAt,
-            Map<String,Object> progress, Validation validation, Attempt latestAttempt) {}
+            Map<String,Object> progress, Validation validation, Attempt latestAttempt,
+            String dependencyState, String blockingReason, String progressSourceTaskNo,
+            SelfopStageEstimate.Estimate stageEstimate) {}
    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,
@@ -11,6 +11,6 @@ public interface SelfopEgressMapper {
 @Update("UPDATE ll_selfop_outbound_discovery_scope SET api_budget_json=#{apiBudgetJson},api_cooldown_until_utc=#{apiCooldownUntilUtc},api_pause_reason=#{apiPauseReason},updated_at=UTC_TIMESTAMP(3) WHERE id=#{id}") int saveAccount(SelfopOutboundDiscoveryScope row);
 @Update("UPDATE ll_selfop_sync_task SET validation_budget_json=#{validationBudgetJson},business_budget_json=#{businessBudgetJson},status=#{status},next_retry_at_utc=#{nextRetryAtUtc},updated_at=UTC_TIMESTAMP(3) WHERE id=#{id}") int saveRoot(SelfopSyncTask row);
 @Update("UPDATE ll_selfop_sync_task SET status='DEAD',next_retry_at_utc=NULL,updated_at=UTC_TIMESTAMP(3) WHERE business_root_task_id=#{root} AND source_account=#{account} AND execution_purpose='BUSINESS' AND status IN ('WAITING_SOURCE','READY','RUNNING','RETRY_WAIT')") int terminateBusinessTasks(@Param("root") Long root,@Param("account") String account);
- @Update("UPDATE ll_selfop_sync_run SET status='FAILED',error_code=#{reason},finished_at_utc=#{now},progress_snapshot_json=(SELECT progress_json FROM ll_selfop_sync_task WHERE id=task_id) WHERE source_account=#{account} AND execution_purpose='BUSINESS' AND task_id IN (SELECT id FROM ll_selfop_sync_task WHERE business_root_task_id=#{root} AND source_account=#{account} AND execution_purpose='BUSINESS') AND status IN ('FETCHING','VALIDATED','YIELDED','READY','BUILDING')") int terminateBusinessRuns(@Param("root") Long root,@Param("account") String account,@Param("reason") String reason,@Param("now") LocalDateTime now);
- @Update("UPDATE ll_selfop_sync_run SET status=#{status},error_code=#{reason},finished_at_utc=#{now},progress_snapshot_json=(SELECT progress_json FROM ll_selfop_sync_task WHERE id=task_id) WHERE task_id IN (SELECT id FROM ll_selfop_sync_task WHERE validation_root_task_id=#{root}) AND status IN ('FETCHING','VALIDATED','YIELDED')") int terminateValidationRuns(@Param("root") Long root,@Param("status") String status,@Param("reason") String reason,@Param("now") LocalDateTime now);
+ @Update("UPDATE ll_selfop_sync_run SET status='FAILED',error_code=#{reason},finished_at_utc=#{now},progress_snapshot_json=(SELECT JSON_SET(JSON_REMOVE(COALESCE(progress_json,JSON_OBJECT()),'$.etaSamples'),'$.summaryAsOf',CONCAT(DATE_FORMAT(#{now},'%Y-%m-%dT%H:%i:%s.%f'),'Z')) FROM ll_selfop_sync_task WHERE id=task_id AND (progress_json IS NULL OR JSON_EXTRACT(progress_json,'$.attemptNo')=attempt_no)) WHERE source_account=#{account} AND execution_purpose='BUSINESS' AND task_id IN (SELECT id FROM ll_selfop_sync_task WHERE business_root_task_id=#{root} AND source_account=#{account} AND execution_purpose='BUSINESS') AND status IN ('FETCHING','VALIDATED','YIELDED','READY','BUILDING')") int terminateBusinessRuns(@Param("root") Long root,@Param("account") String account,@Param("reason") String reason,@Param("now") LocalDateTime now);
+ @Update("UPDATE ll_selfop_sync_run SET status=#{status},error_code=#{reason},finished_at_utc=#{now},progress_snapshot_json=(SELECT JSON_SET(JSON_REMOVE(COALESCE(progress_json,JSON_OBJECT()),'$.etaSamples'),'$.summaryAsOf',CONCAT(DATE_FORMAT(#{now},'%Y-%m-%dT%H:%i:%s.%f'),'Z')) FROM ll_selfop_sync_task WHERE id=task_id AND (progress_json IS NULL OR JSON_EXTRACT(progress_json,'$.attemptNo')=attempt_no)) WHERE task_id IN (SELECT id FROM ll_selfop_sync_task WHERE validation_root_task_id=#{root}) AND status IN ('FETCHING','VALIDATED','YIELDED')") int terminateValidationRuns(@Param("root") Long root,@Param("status") String status,@Param("reason") String reason,@Param("now") LocalDateTime now);
 }
@@ -153,10 +153,11 @@ public class SelfopDiscoveryService {
                    full=stage(finalFetchTask,finalFetchRun,acceptedDetail,existing==null?1:existing.getFirstPageNo(),acceptedObservation,"VALID");
                    if(existing!=null){full.setId(existing.getId());full.setRunId(existing.getRunId());full.setCreatedAt(existing.getCreatedAt());stages.updateById(full);}else stages.insert(full);
                }
-                ref.setEffectiveStageId(full.getId());ref.setWarehouseCode(acceptedDetail.warehouseCode());saveManifest(checkpoint.task(),checkpoint.run(),manifest);
+                ref.setEffectiveStageId(full.getId());ref.setWarehouseCode(acceptedDetail.warehouseCode());saveManifest(checkpoint.task(),checkpoint.run(),manifest,existing==null||!"VALID".equals(existing.getValidationStatus()));
                if(!currentFetchTask.getId().equals(task.getId())){var refreshManifest=new SelfopRoutingManifest();refreshManifest.setDirectoryComplete(true);refreshManifest.setPages(1);refreshManifest.setTotal(1L);
                    var refreshRef=new SelfopRoutingManifest.Entry();refreshRef.setCustomerCode(ref.getCustomerCode());refreshRef.setDeliveryNo(ref.getDeliveryNo());refreshRef.setWarehouseCode(ref.getWarehouseCode());refreshRef.setEffectiveStageId(full.getId());
                    refreshManifest.getObjects().put(entry.getKey(),refreshRef);currentFetchTask.setRoutingManifestJson(json(refreshManifest));
+                    SelfopProgressProjection.applyCheckpoint(currentFetchTask,currentFetchRun,"DETAIL","ORDER",1,1L,refreshManifest.getScan(),existing==null||!"VALID".equals(existing.getValidationStatus()));
                    SelfopProgressProjection.freezeRun(currentFetchTask,currentFetchRun);currentFetchRun.setStatus("ROUTED");currentFetchRun.setFetchFinishedAtUtc(utcNow());currentFetchRun.setFinishedAtUtc(utcNow());currentFetchRun.setUpdatedAt(utcNow());runs.updateById(currentFetchRun);
                    currentFetchTask.setStatus("SUCCEEDED");currentFetchTask.setUpdatedAt(utcNow());tasks.updateById(currentFetchTask);}
            });
@@ -204,10 +205,14 @@ public class SelfopDiscoveryService {
        s.setContentHash(SelfopOutboundFingerprint.order(payload));s.setPayloadJson(json(payload));s.setValidationStatus(status);s.setCreatedAt(utcNow());s.setUpdatedAt(utcNow());return s;
    }
    private void saveManifest(SelfopSyncTask task,SelfopSyncRun active,SelfopRoutingManifest manifest){
+        // Directory pages are freshly fetched. Detail reference reuse cannot establish throughput.
+        saveManifest(task,active,manifest,!manifest.isDirectoryComplete());
+    }
+    private void saveManifest(SelfopSyncTask task,SelfopSyncRun active,SelfopRoutingManifest manifest,boolean realWork){
        task.setRoutingManifestJson(json(manifest));task.setUpdatedAt(utcNow());
        SelfopProgressProjection.applyCheckpoint(task,active,manifest.isDirectoryComplete()?"DETAIL":"DIRECTORY",manifest.isDirectoryComplete()?"ORDER":"PAGE",
            manifest.isDirectoryComplete()?manifest.getObjects().values().stream().filter(e->e.getEffectiveStageId()!=null).count():Math.max(0,manifest.getNextPage()-1),
-            manifest.isDirectoryComplete()?(long)manifest.getObjects().size():manifest.getPages()==null?null:manifest.getPages().longValue(),manifest.getScan());
+            manifest.isDirectoryComplete()?(long)manifest.getObjects().size():manifest.getPages()==null?null:manifest.getPages().longValue(),manifest.getScan(),realWork);
        runs.updateById(active);tasks.updateById(task);
    }
    private void route(SelfopSyncTask parent,SelfopSyncRun parentRun,SelfopRoutingManifest manifest,SelfopLeaseService.Lease lease,boolean detailsResolved){
@@ -238,7 +243,8 @@ public class SelfopDiscoveryService {
                child.setQueryScopeJson(scope.getScopeJson());child.setScopeHash(scope.getScopeHash());child.setRequestHash(hash(List.of(parent.getId(),scope.getScopeHash())));child.setStatus(ready?"READY":"WAITING_SOURCE");child.setApplyState("NOT_STARTED");
                var saved=taskService.create(child);
                if(!Set.of("SUCCEEDED","DEAD","SAMPLE_DONE").contains(saved.getStatus())&& !Set.of("APPLYING","APPLIED","BLOCKED").contains(saved.getApplyState())){
-                    saved.setRoutingManifestJson(json(own));saved.setStatus(ready?"READY":"WAITING_SOURCE");saved.setUpdatedAt(utcNow());tasks.updateById(saved);}
+                    saved.setRoutingManifestJson(json(own));saved.setStatus(ready?"READY":"WAITING_SOURCE");
+                    SelfopProgressProjection.localWorkReady(saved,ready);saved.setUpdatedAt(utcNow());tasks.updateById(saved);}
            }
        });
    }
@@ -145,8 +145,8 @@ public class SelfopTaskService {
        var now=utcNow();var active=runs.selectOne(new QueryWrapper<SelfopSyncRun>().eq("task_id",taskId).eq("status","FETCHING").orderByDesc("id").last("LIMIT 1"));
        if(task.getScopeId()!=null && "INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())) {
            var yielded=runs.selectOne(new QueryWrapper<SelfopSyncRun>().eq("task_id",taskId).eq("status","YIELDED").orderByDesc("id").last("LIMIT 1"));
-            if(yielded!=null){yielded.setStatus("FETCHING");yielded.setFinishedAtUtc(null);yielded.setFenceToken(lease.fence());yielded.setUpdatedAt(now);
-                runs.update(null,new UpdateWrapper<SelfopSyncRun>().eq("id",yielded.getId()).set("status","FETCHING").set("fence_token",lease.fence()).set("finished_at_utc",null).set("updated_at",now));
+            if(yielded!=null){yielded.setStatus("FETCHING");yielded.setFinishedAtUtc(null);yielded.setProgressSnapshotJson(null);yielded.setFenceToken(lease.fence());yielded.setUpdatedAt(now);
+                runs.update(null,new UpdateWrapper<SelfopSyncRun>().eq("id",yielded.getId()).set("status","FETCHING").set("fence_token",lease.fence()).set("finished_at_utc",null).set("progress_snapshot_json",null).set("updated_at",now));
                task.setStatus("RUNNING");task.setUpdatedAt(now);tasks.updateById(task);return yielded;}
        }
        if(active!=null){require(!Objects.equals(active.getFenceToken(),lease.fence()),"SELFOP_TASK_ALREADY_RUNNING");