356 lines
20 KiB
Python
Executable File
356 lines
20 KiB
Python
Executable File
#!/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 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()
|
|
status["myTools"]["productMigrationAlive"] = product_alive
|
|
status["unattendedState"] = (
|
|
"IMPLEMENTING" if product_alive and product.get("status") == "implementing"
|
|
else "COMPLETE" if product.get("status") == "implementation_complete" or (
|
|
product.get("workflowTotal", 0) > 0
|
|
and product.get("workflowImplemented") == product.get("workflowTotal")
|
|
and product.get("taskImplemented") == product.get("taskTotal")
|
|
)
|
|
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'}")
|
|
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")
|
|
product_backlog = _read_json(CONTROL_ROOT / "web-product-backlog.json")
|
|
product_status = _read_json(CONTROL_ROOT / "web-product-migration-status.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]
|
|
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
|
|
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,
|
|
},
|
|
"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)
|
|
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
|
|
if path.is_file():
|
|
actual = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
variants[variant] = {
|
|
"url": resource["url"],
|
|
"bytes": path.stat().st_size if path.is_file() else 0,
|
|
"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,
|
|
})
|
|
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" 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" 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
|