feat(my-tools): show migration progress and duration estimate

my-tools/show_blender_web_scheduler.py

@@ -9,9 +9,12 @@ from __future__ import annotations
 
 import argparse
 import json
+import statistics
+import re
+import time
 from pathlib import Path
 
-from wbv2_runtime import CONTROL_ROOT, live_status, read_json
+from wbv2_runtime import CONTROL_ROOT, ROOT, live_status, read_json
 
 
 def main() -> int:
@@ -20,45 +23,186 @@ def main() -> int:
     parser.add_argument("--watch", action="store_true")
     parser.add_argument("--interval-ms", type=int, default=2000)
     args = parser.parse_args()
-    result = live_status(json_only=True, watch=args.watch, interval_ms=args.interval_ms)
-    if result.returncode != 0:
-        if result.stderr:
-            print(result.stderr, end="")
-        return result.returncode
-    if args.watch:
-        # Preserve the Node tool's ANSI refresh and output contract for long
-        # running terminal sessions.
-        print(result.stdout, end="")
-        return 0
+    while True:
+        result = live_status(json_only=True, watch=False, interval_ms=args.interval_ms)
+        if result.returncode != 0:
+            if result.stderr:
+                print(result.stderr, end="")
+            return result.returncode
+        try:
+            status = json.loads(result.stdout)
+        except json.JSONDecodeError:
+            print(result.stdout, end="")
+            return 0
+        status["myTools"] = {
+            "watchdogStatus": read_json(CONTROL_ROOT / "my-tools-watchdog.json"),
+            "schedulerWrapperStatus": read_json(CONTROL_ROOT / "my-tools-scheduler.json"),
+        }
+        status["progress"] = project_progress()
+        if args.json:
+            print(json.dumps(status, ensure_ascii=False, separators=(",", ":")))
+            if not args.watch:
+                return 0
+            try:
+                time.sleep(args.interval_ms / 1000)
+            except KeyboardInterrupt:
+                return 0
+            continue
+        if args.watch:
+            print("\x1b[2J\x1b[H", end="")
+        print("Blender Web/WBV2 scheduler")
+        print(f"  state: {status.get('state', 'UNKNOWN')}")
+        print(f"  phase: {status.get('phase', 'unknown')}")
+        queue = status.get("queue", {})
+        print(f"  current task: {queue.get('task') or 'none'}")
+        print(f"  parent manifest: {queue.get('parentManifest') or 'none'}")
+        scheduler = status.get("scheduler", {})
+        print(f"  scheduler: pid={scheduler.get('pid') or 'none'} alive={scheduler.get('alive', False)} state={scheduler.get('state') or 'unknown'}")
+        guardian = status.get("guardian", {})
+        supervisor = status.get("supervisor", {})
+        print(f"  guardian: pid={guardian.get('pid') or 'none'} alive={guardian.get('alive', False)} state={guardian.get('state') or 'unknown'}")
+        print(f"  supervisor: pid={supervisor.get('pid') or 'none'} alive={supervisor.get('alive', False)} state={supervisor.get('state') or 'unknown'}")
+        print(f"  latest handoff: {status.get('latestHandoff') or 'none'}")
+        print(f"  local HEAD: {status.get('localHead') or 'unknown'}")
+        for line in format_progress(status["progress"]):
+            print(line)
+        print(f"  message: {status.get('message', '')}")
+        if not args.watch:
+            return 0
+        try:
+            time.sleep(args.interval_ms / 1000)
+        except KeyboardInterrupt:
+            return 0
+
+
+def _read_json(path: Path) -> dict:
     try:
-        status = json.loads(result.stdout)
-    except json.JSONDecodeError:
-        print(result.stdout, end="")
-        return 0
-    status["myTools"] = {
-        "watchdogStatus": read_json(CONTROL_ROOT / "my-tools-watchdog.json"),
-        "schedulerWrapperStatus": read_json(CONTROL_ROOT / "my-tools-scheduler.json"),
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except (FileNotFoundError, json.JSONDecodeError):
+        return {}
+    return value if isinstance(value, dict) else {}
+
+
+def _queue_task() -> str | None:
+    try:
+        match = re.search(r"\| 当前任务 \| `([^`]+)` \|", (ROOT / "docs/EXECUTION_QUEUE.md").read_text(encoding="utf-8"))
+    except OSError:
+        return None
+    return match.group(1) if match else None
+
+
+def project_progress() -> dict:
+    """Summarize progress from existing receipts without running any checks."""
+    inventory = _read_json(CONTROL_ROOT / "dynamic-inventory.json")
+    total = int(inventory.get("summary", {}).get("unitCount", 0) or 0)
+    verified = int(inventory.get("coverage", {}).get("verifiedUnits", 0) or 0)
+    unverified = int(inventory.get("coverage", {}).get("unverifiedUnits", max(0, total - verified)) or 0)
+    index = _read_json(CONTROL_ROOT / "task-index.json")
+    source = index.get("source", {}).get("path")
+    plan = _read_json(ROOT / source) if source else {}
+    tasks = plan.get("tasks", []) if isinstance(plan.get("tasks", []), list) else []
+    done_tasks = sum(1 for task in tasks if task.get("status") == "done")
+    active_tasks = [task.get("id") for task in tasks if task.get("ready") is True]
+    blocked_tasks = sum(1 for task in tasks if task.get("status") == "blocked")
+
+    receipts = []
+    historical_candidates = []
+    for task_dir in ROOT.glob("tests/golden/WBV2-P*-001"):
+        receipt = _read_json(task_dir / "module-receipt.json")
+        historical = receipt.get("inventory", {}).get("historical", {})
+        if historical:
+            historical_candidates.append((task_dir.name, historical))
+        if receipt.get("status") != "PASS":
+            continue
+        dynamic = receipt.get("inventory", {}).get("dynamic", {})
+        duration = receipt.get("efficiency", {}).get("totalDurationMs")
+        if not isinstance(duration, (int, float)) or duration < 0:
+            duration = sum(item.get("durationMs", 0) for item in receipt.get("commands", []) if isinstance(item, dict))
+        current_verified = dynamic.get("verifiedUnits")
+        if isinstance(current_verified, int):
+            receipts.append({
+                "task": receipt.get("task", task_dir.name),
+                "verifiedUnits": current_verified,
+                "durationMs": duration,
+                "reused": bool(receipt.get("efficiency", {}).get("existingEvidenceReused")),
+                "commands": receipt.get("commands", []),
+            })
+    receipts.sort(key=lambda item: item["task"])
+    productive = [item for item in receipts if not item["reused"] and item["durationMs"] > 0]
+    deltas = []
+    previous = None
+    for item in receipts:
+        if previous is not None and item["verifiedUnits"] > previous and not item["reused"] and item["durationMs"] > 0:
+            deltas.append({"units": item["verifiedUnits"] - previous, "durationMs": item["durationMs"], "task": item["task"]})
+        previous = max(previous or 0, item["verifiedUnits"])
+    if deltas:
+        rates = [item["units"] / (item["durationMs"] / 3_600_000) for item in deltas if item["durationMs"] > 0]
+        unit_rate = statistics.median(rates) if rates else 0.0
+        batch_units = statistics.median(item["units"] for item in deltas)
+    else:
+        unit_rate = 0.0
+        batch_units = 0.0
+    remaining_hours = unverified / unit_rate if unit_rate > 0 else None
+    stage_durations = {}
+    for item in productive:
+        for index, command in enumerate(item.get("commands", [])):
+            duration = command.get("durationMs") if isinstance(command, dict) else None
+            if isinstance(duration, (int, float)) and duration >= 0:
+                stage_durations.setdefault(str(index + 1), []).append(duration)
+    stage_medians = {stage: round(statistics.median(values)) for stage, values in stage_durations.items() if values}
+    historical = max(historical_candidates, default=("", {}), key=lambda item: item[0])[1]
+    return {
+        "scope": "dynamic-source-inventory",
+        "currentTask": _queue_task(),
+        "totalUnits": total,
+        "verifiedUnits": verified,
+        "unverifiedUnits": unverified,
+        "verifiedPercent": round((verified / total) * 100, 2) if total else 0.0,
+        "taskPlan": {"tasks": len(tasks), "done": done_tasks, "active": active_tasks, "blocked": blocked_tasks},
+        "observedBatches": len(receipts),
+        "productiveBatches": len(deltas),
+        "reusedBatches": sum(1 for item in receipts if item["reused"]),
+        "observedExecutionMs": round(sum(item["durationMs"] for item in productive)),
+        "stageMedianDurationMs": stage_medians,
+        "medianIncrementalUnitsPerBatch": round(batch_units, 2),
+        "medianIncrementalUnitsPerHour": round(unit_rate, 2),
+        "estimatedRemainingHours": round(remaining_hours, 2) if remaining_hours is not None else None,
+        "estimatedTotalHours": round((verified / unit_rate) + remaining_hours, 2) if unit_rate > 0 and remaining_hours is not None else None,
+        "estimateBasis": "median successful non-reused module receipts; excludes blocked/unrunnable units and final full verification",
+        "lastObservedReceipt": receipts[-1]["task"] if receipts else None,
+        "lastIncrementalBatch": deltas[-1]["task"] if deltas else None,
+        "historicalInventory": {
+            "total": historical.get("total", 6900),
+            "effectiveCompleted": historical.get("effectiveCompleted", 290),
+            "effectivePending": historical.get("effectivePending", 6609),
+            "reproductionRequired": True,
+        },
     }
-    if args.json:
-        print(json.dumps(status, ensure_ascii=False, separators=(",", ":")))
-        return 0
-    print("Blender Web/WBV2 scheduler")
-    print(f"  state: {status.get('state', 'UNKNOWN')}")
-    print(f"  phase: {status.get('phase', 'unknown')}")
-    queue = status.get("queue", {})
-    print(f"  current task: {queue.get('task') or 'none'}")
-    print(f"  parent manifest: {queue.get('parentManifest') or 'none'}")
-    scheduler = status.get("scheduler", {})
-    print(f"  scheduler: pid={scheduler.get('pid') or 'none'} alive={scheduler.get('alive', False)} state={scheduler.get('state') or 'unknown'}")
-    guardian = status.get("guardian", {})
-    supervisor = status.get("supervisor", {})
-    print(f"  guardian: pid={guardian.get('pid') or 'none'} alive={guardian.get('alive', False)} state={guardian.get('state') or 'unknown'}")
-    print(f"  supervisor: pid={supervisor.get('pid') or 'none'} alive={supervisor.get('alive', False)} state={supervisor.get('state') or 'unknown'}")
-    print(f"  latest handoff: {status.get('latestHandoff') or 'none'}")
-    print(f"  local HEAD: {status.get('localHead') or 'unknown'}")
-    print(f"  message: {status.get('message', '')}")
-    return 0
+
+
+def format_progress(progress: dict) -> list[str]:
+    estimate = progress.get("estimatedRemainingHours")
+    estimate_text = f"{estimate} h" if estimate is not None else "n/a"
+    total_estimate = progress.get("estimatedTotalHours")
+    total_text = f"{total_estimate} h" if total_estimate is not None else "n/a"
+    plan = progress.get("taskPlan", {})
+    return [
+        "",
+        "Project migration progress",
+        f"  dynamic units: {progress.get('verifiedUnits', 0)}/{progress.get('totalUnits', 0)} verified ({progress.get('verifiedPercent', 0)}%), remaining={progress.get('unverifiedUnits', 0)}",
+        f"  task plan: {plan.get('done', 0)}/{plan.get('tasks', 0)} done, active={','.join(plan.get('active', [])) or 'none'}, blocked={plan.get('blocked', 0)}",
+        f"  historical ledger: {progress.get('historicalInventory', {}).get('effectiveCompleted', 0)}/{progress.get('historicalInventory', {}).get('total', 6900)} effective, pending={progress.get('historicalInventory', {}).get('effectivePending', 0)} (full reproduction required)",
+        f"  observed batches: {progress.get('observedBatches', 0)} total, productive={progress.get('productiveBatches', 0)}, reused={progress.get('reusedBatches', 0)}",
+        f"  efficiency: median +{progress.get('medianIncrementalUnitsPerBatch', 0)} units/batch, {progress.get('medianIncrementalUnitsPerHour', 0)} units/hour",
+        f"  stage medians: {progress.get('stageMedianDurationMs') or 'n/a'} ms (1=build, 2=typecheck, 3=web build, 4=Chromium)",
+        f"  estimate: remaining={estimate_text}, total={total_text}",
+        f"  estimate basis: {progress.get('estimateBasis', 'n/a')}",
+    ]
 
 
 if __name__ == "__main__":
-    raise SystemExit(main())
+    try:
+        raise SystemExit(main())
+    except BrokenPipeError:
+        # Quietly support consumers such as ``head`` closing a JSON stream.
+        pass