#!/usr/bin/env python3 """Display the live Blender/WBV2 scheduler state. The Node live-status command remains the source of truth for RUNNING, IDLE and BLOCKED decisions. """ from __future__ import annotations import argparse import hashlib import json import os import statistics import re import time from datetime import datetime from pathlib import Path from wbv2_runtime import CONTROL_ROOT, ROOT, live_status, read_json def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--json", action="store_true", help="emit machine-readable status") parser.add_argument("--watch", action="store_true") parser.add_argument("--interval-ms", type=int, default=2000) args = parser.parse_args() 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() watchdog = status["myTools"]["watchdogStatus"] watchdog_pid = watchdog.get("pid") if isinstance(watchdog, dict) else None watchdog_alive = isinstance(watchdog_pid, int) and (Path("/proc") / str(watchdog_pid)).exists() status["myTools"]["watchdogAlive"] = watchdog_alive product = status["progress"].get("productMigration", {}) product_pid = product.get("pid") product_alive = isinstance(product_pid, int) and (Path("/proc") / str(product_pid)).exists() product_complete = ( product.get("workflowTotal", 0) > 0 and product.get("workflowImplemented") == product.get("workflowTotal") and product.get("taskImplemented") == product.get("taskTotal") ) status["myTools"]["productMigrationAlive"] = product_alive status["unattendedState"] = ( "IMPLEMENTING" if product_alive and product.get("status") == "implementing" else "COMPLETE" if product_complete else "BLOCKED" if product.get("status") == "blocked" else "PLANNING" if status.get("state") == "IDLE" and watchdog_alive and watchdog.get("status") == "waiting_for_source_batch" else status.get("state", "UNKNOWN") ) status["fastPath"] = fast_path_status(status) 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('unattendedState', 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'}") watchdog = status.get("myTools", {}).get("watchdogStatus", {}) print(f" watchdog: pid={watchdog.get('pid') or 'none'} alive={status.get('myTools', {}).get('watchdogAlive', False)} status={watchdog.get('status') or 'unknown'} reason={watchdog.get('reason') or 'none'} task={watchdog.get('task') or 'none'}") print(f" product migration: pid={product.get('pid') or 'none'} alive={product_alive} status={product.get('status') or 'not-started'} active={product.get('activeTask') or 'none'}") route_coverage = status["progress"].get("featureRouteCoverage", {}) route_counts = route_coverage.get("routes", {}) print( " feature routes: " f"total={route_coverage.get('unitCount', 0)} " f"WEB_MAIN={route_counts.get('WEB_MAIN', 0)} " f"SERVER_NATIVE={route_counts.get('SERVER_NATIVE', 0)} " f"BROWSER_EQUIVALENT={route_counts.get('BROWSER_EQUIVALENT', 0)} " f"UNAVAILABLE={route_counts.get('UNAVAILABLE', 0)}" ) evidence = route_coverage.get("behavioralEvidence", {}) print( " feature evidence: " f"status={evidence.get('status', 'UNVERIFIED')} " f"verified={evidence.get('verifiedUnits', 0)} " f"unverified={evidence.get('unverifiedUnits', route_coverage.get('unitCount', 0))} " "(route assignment is not behavioral completion)" ) 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"], status["fastPath"]): print(line) if status.get("unattendedState") == "IMPLEMENTING": print(" message: IMPLEMENTING: one workflow task is being edited; verification is deferred to user operation.") elif status.get("unattendedState") == "COMPLETE": print(" message: COMPLETE: the Web workflow implementation backlog is closed; user-reported defects become focused follow-up tasks.") elif status.get("unattendedState") == "PLANNING": print(" message: PLANNING: watchdog heartbeat is active; exact source gaps are queued without false coverage.") else: 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: 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") route_catalog = _read_json(ROOT / "web/app/src/capabilities/blender-feature-route-catalog.json") product_backlog = _read_json(CONTROL_ROOT / "web-product-backlog.json") product_status = _read_json(CONTROL_ROOT / "web-product-migration-status.json") completeness_audit = _read_json(ROOT / "tests/golden/WBV2/feature-completeness-audit.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 = {} stage_labels = {} 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 = str(index + 1) stage_durations.setdefault(stage, []).append(duration) if stage not in stage_labels and isinstance(command.get("command"), str): stage_labels[stage] = command["command"].split("&&", 1)[0].strip()[:120] stage_medians = {stage: round(statistics.median(values)) for stage, values in stage_durations.items() if values} bottleneck_stage = max(stage_medians, key=stage_medians.get) if stage_medians else None bottleneck = { "stage": bottleneck_stage, "label": stage_labels.get(bottleneck_stage) if bottleneck_stage else None, "medianDurationMs": stage_medians.get(bottleneck_stage) if bottleneck_stage else None, } historical = max(historical_candidates, default=("", {}), key=lambda item: item[0])[1] workflow_modules = product_backlog.get("workflowModules", []) if isinstance(product_backlog.get("workflowModules"), list) else [] product_tasks = product_backlog.get("tasks", []) if isinstance(product_backlog.get("tasks"), list) else [] workflow_implemented = sum(module.get("status") == "implemented" for module in workflow_modules) task_implemented = sum(task.get("status") == "implemented" for task in product_tasks) completed_durations = [] for task in product_tasks: try: started = datetime.fromisoformat(task["startedAt"]) completed = datetime.fromisoformat(task["completedAt"]) except (KeyError, TypeError, ValueError): continue completed_durations.append((completed - started).total_seconds()) remaining_tasks = len(product_tasks) - task_implemented product_remaining_seconds = statistics.median(completed_durations) * remaining_tasks if completed_durations else None route_names = ("WEB_MAIN", "SERVER_NATIVE", "BROWSER_EQUIVALENT", "UNAVAILABLE") catalog_counts = route_catalog.get("counts", {}) if isinstance(route_catalog.get("counts"), dict) else {} route_counts = {name: int(catalog_counts.get(name, 0) or 0) for name in route_names} route_unit_count = int(route_catalog.get("unitCount", sum(route_counts.values())) or 0) classified_count = int(route_catalog.get("classifiedCount", sum(route_counts.values())) or 0) catalog_evidence = route_catalog.get("behavioralEvidence", {}) if isinstance(route_catalog.get("behavioralEvidence"), dict) else {} feature_route_coverage = { "command": "featureRouteCoverage", "sourceTreeSha256": route_catalog.get("sourceTreeSha256"), "unitCount": route_unit_count, "classifiedCount": classified_count, "unclassifiedCount": int(route_catalog.get("unclassifiedCount", max(0, route_unit_count - classified_count)) or 0), "routes": route_counts, "routeAssignmentComplete": route_unit_count > 0 and classified_count == route_unit_count, "behavioralEvidence": { "status": catalog_evidence.get("status", "PARTIAL"), "verifiedUnits": int(catalog_evidence.get("verifiedUnits", verified) or 0), "unverifiedUnits": int(catalog_evidence.get("unverifiedUnits", max(0, route_unit_count - verified)) or 0), "policy": catalog_evidence.get("policy", "route classification does not replace exact behavioral evidence"), }, } feature_route_gaps = { "command": "featureRouteGaps", "routeAssignmentGaps": max(0, route_unit_count - classified_count), "unavailableUnits": route_counts["UNAVAILABLE"], "unverifiedBehavioralEvidence": int(catalog_evidence.get("unverifiedUnits", max(0, route_unit_count - verified)) or 0), "behavioralEvidenceStatus": catalog_evidence.get("status", "PARTIAL"), } 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, "bottleneck": bottleneck, "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, }, "featureRouteCoverage": feature_route_coverage, "featureRouteGaps": feature_route_gaps, "completenessAudit": { "status": completeness_audit.get("status", "MISSING"), "implementationTasks": completeness_audit.get("implementation", {}).get("productTasks", {}), "behavioralEvidence": completeness_audit.get("behavioralEvidence", {}), "smoke": completeness_audit.get("smoke", {}), "actualMissingOrBlocked": completeness_audit.get("actualMissingOrBlocked", []), }, "productMigration": { **product_status, "strategy": product_backlog.get("strategy"), "verificationPolicy": product_backlog.get("verificationPolicy"), "workflowTotal": len(workflow_modules), "workflowImplemented": workflow_implemented, "workflowPercent": round(workflow_implemented / len(workflow_modules) * 100, 2) if workflow_modules else 0.0, "workflowActive": [module.get("id") for module in workflow_modules if module.get("status") == "active"], "workflowPartial": [module.get("id") for module in workflow_modules if module.get("status") == "partial"], "workflowBlocked": [module.get("id") for module in workflow_modules if module.get("status") == "blocked"], "taskTotal": len(product_tasks), "taskImplemented": task_implemented, "taskPending": sum(task.get("status") == "pending" for task in product_tasks), "taskActive": [task.get("id") for task in product_tasks if task.get("status") == "active"], "taskBlocked": [task.get("id") for task in product_tasks if task.get("status") == "blocked"], "estimatedRemainingMinutes": round(product_remaining_seconds / 60, 1) if product_remaining_seconds is not None else None, }, } def fast_path_status(status: dict) -> dict: scheduler = status.get("scheduler", {}) if isinstance(status, dict) else {} latest_batch = scheduler.get("latestBatch") if isinstance(scheduler.get("latestBatch"), dict) else {} cache = _read_json(CONTROL_ROOT / "context-cache.json") ledger = _read_json(CONTROL_ROOT / "failure-ledger.json") planner = _read_json(CONTROL_ROOT / "auto-planner.json") product_backlog = _read_json(CONTROL_ROOT / "web-product-backlog.json") modules = wasm_module_status() return { "batch": { "batchId": latest_batch.get("batchId"), "selectedRecordCount": len(latest_batch.get("selectedRecordIds", [])), "completed": bool(latest_batch.get("completed")), }, "contextCache": { "status": cache.get("status", "MISSING"), "task": cache.get("task"), "schemaVersion": cache.get("schemaVersion"), "inputCount": len(cache.get("inputs", {})) if isinstance(cache.get("inputs"), dict) else 0, }, "failureLedger": { "entryCount": len(ledger.get("entries", [])) if isinstance(ledger.get("entries"), list) else 0, "hydratedLogs": len(ledger.get("hydratedLogs", [])) if isinstance(ledger.get("hydratedLogs"), list) else 0, }, "autoPlanner": { "status": planner.get("status", "MISSING"), "reason": planner.get("reason"), "pending": planner.get("pending", {}).get("count") if isinstance(planner.get("pending"), dict) else None, "nextAction": planner.get("nextAction"), }, "commandBudget": {"maxCommands": 48, "maxReadCommands": 16}, "verificationPolicy": product_backlog.get("verificationPolicy", "incremental-context-only-until-final-audit"), "wasmModules": modules, } def wasm_module_status() -> dict: """Report module preload policy and artifact integrity without fetching or rebuilding.""" manifest_path = ROOT / "web/app/public/wasm-module-manifest.json" manifest = _read_json(manifest_path) cache_path = CONTROL_ROOT / "wasm-module-status-cache.json" cache = _read_json(cache_path) cached_entries = cache.get("entries", {}) if isinstance(cache.get("entries"), dict) else {} next_cache = {"schemaVersion": 1, "manifestMtimeNs": None, "entries": {}} try: next_cache["manifestMtimeNs"] = str(manifest_path.stat().st_mtime_ns) except OSError: pass entries = [] for entry in manifest.get("modules", []) if isinstance(manifest.get("modules"), list) else []: resources = entry.get("resources", {}) if isinstance(entry, dict) else {} variants = {} for variant, resource in resources.items() if isinstance(resources, dict) else []: if not isinstance(resource, dict) or not isinstance(resource.get("url"), str): continue relative = resource["url"].lstrip("/") path = ROOT / "web/app/public" / relative actual = None bytes_count = 0 mtime_ns = None try: stat = path.stat() bytes_count = stat.st_size mtime_ns = str(stat.st_mtime_ns) except OSError: stat = None cache_key = str(path.relative_to(ROOT)) cached = next_cache["entries"].get(cache_key) if not isinstance(cached, dict): cached = cached_entries.get(cache_key, {}) if isinstance(cached_entries.get(cache_key), dict) else {} if stat and cached.get("bytes") == bytes_count and cached.get("mtimeNs") == mtime_ns and isinstance(cached.get("sha256"), str): actual = cached["sha256"] elif stat: actual = hashlib.sha256(path.read_bytes()).hexdigest() next_cache["entries"][cache_key] = {"bytes": bytes_count, "mtimeNs": mtime_ns, "sha256": actual} variants[variant] = { "url": resource["url"], "bytes": bytes_count, "integrity": actual == resource.get("sha256") if actual else False, } entries.append({ "id": entry.get("id"), "lifecycle": entry.get("lifecycle"), "preload": entry.get("preload") is True, "variants": variants, }) try: temporary = cache_path.with_suffix(f".tmp-{os.getpid()}") temporary.write_text(json.dumps(next_cache, ensure_ascii=False, separators=(",", ":")) + "\n", encoding="utf-8") temporary.replace(cache_path) except OSError: # Status reporting remains read-only from the caller's perspective if # the control directory is unavailable; the next poll simply rehashes. pass return { "manifest": "web/app/public/wasm-module-manifest.json" if manifest else None, "moduleCount": len(entries), "preloadCompleted": sum(1 for item in entries if item["lifecycle"] == "complete" and item["preload"]), "debugOnDemand": sum(1 for item in entries if item["lifecycle"] == "debug" and not item["preload"]), "entries": entries, } def format_progress(progress: dict, fast_path: dict | None = None) -> 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", {}) fast_path = fast_path or {} batch = fast_path.get("batch", {}) context_cache = fast_path.get("contextCache", {}) ledger = fast_path.get("failureLedger", {}) planner = fast_path.get("autoPlanner", {}) product = progress.get("productMigration", {}) product_estimate = product.get("estimatedRemainingMinutes") return [ "", "Web product implementation progress", f" workflow modules: {product.get('workflowImplemented', 0)}/{product.get('workflowTotal', 0)} implemented ({product.get('workflowPercent', 0)}%), active={','.join(product.get('workflowActive', [])) or 'none'}, partial={','.join(product.get('workflowPartial', [])) or 'none'}", f" implementation tasks: {product.get('taskImplemented', 0)}/{product.get('taskTotal', 0)} done, active={','.join(product.get('taskActive', [])) or 'none'}, pending={product.get('taskPending', 0)}, blocked={','.join(product.get('taskBlocked', [])) or 'none'}", f" implementation estimate: {product_estimate if product_estimate is not None else 'learning'} minutes remaining; verification={product.get('verificationPolicy') or 'deferred'}", f" source traceability reference: {progress.get('verifiedUnits', 0)}/{progress.get('totalUnits', 0)} bound; excluded from product completion denominator", f" completeness audit: {progress.get('completenessAudit', {}).get('status', 'MISSING')} blocked={sum(item.get('status') == 'BLOCKED' for item in progress.get('completenessAudit', {}).get('actualMissingOrBlocked', []))} behavioral-unverified={progress.get('completenessAudit', {}).get('behavioralEvidence', {}).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" old verification estimate: remaining={estimate_text}, total={total_text} (reference only; strategy retired)", f" observed bottleneck: stage={progress.get('bottleneck', {}).get('stage') or 'n/a'} median={progress.get('bottleneck', {}).get('medianDurationMs') or 0}ms command={progress.get('bottleneck', {}).get('label') or 'n/a'}", f" fast path: batch={batch.get('batchId') or 'none'} selected={batch.get('selectedRecordCount', 0)} completed={batch.get('completed', False)}", f" context cache: {context_cache.get('status', 'MISSING')} task={context_cache.get('task') or 'none'} inputs={context_cache.get('inputCount', 0)}; failure ledger entries={ledger.get('entryCount', 0)} hydratedLogs={ledger.get('hydratedLogs', 0)}", f" command budget: max={fast_path.get('commandBudget', {}).get('maxCommands', 48)}, read/exploration max={fast_path.get('commandBudget', {}).get('maxReadCommands', 16)}; verification={fast_path.get('verificationPolicy', 'n/a')}", f" retired source verifier: {planner.get('status', 'MISSING')} pending={planner.get('pending', 'n/a')} (traceability reference only)", f" wasm modules: manifest={fast_path.get('wasmModules', {}).get('manifest') or 'missing'} completed-preload={fast_path.get('wasmModules', {}).get('preloadCompleted', 0)} debug-on-demand={fast_path.get('wasmModules', {}).get('debugOnDemand', 0)}", ] if __name__ == "__main__": try: raise SystemExit(main()) except BrokenPipeError: # Quietly support consumers such as ``head`` closing a JSON stream. pass