feat(web): switch unattended migration to product workflows

my-tools/run_web_product_migration.py

@@ -0,0 +1,252 @@
+#!/usr/bin/env python3
+"""Implement the Web product backlog through bounded non-interactive Codex turns."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import signal
+import subprocess
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+from wbv2_runtime import ROOT
+
+
+CONTROL = ROOT / "tests/golden/WBV2"
+BACKLOG = CONTROL / "web-product-backlog.json"
+STATUS = CONTROL / "web-product-migration-status.json"
+LOCK = CONTROL / "web-product-migration.lock"
+PID = CONTROL / "web-product-migration.pid"
+LOG_ROOT = CONTROL / "product-migration"
+ACTIVE_PROCESS: subprocess.Popen[str] | None = None
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def read_json(path: Path) -> dict:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def write_json(path: Path, value: object) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + f".tmp-{os.getpid()}")
+    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    temporary.replace(path)
+
+
+def alive(pid: object) -> bool:
+    if not isinstance(pid, int) or pid <= 0:
+        return False
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+    return True
+
+
+def acquire_lock() -> None:
+    try:
+        handle = os.open(LOCK, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
+    except FileExistsError:
+        try:
+            owner = read_json(LOCK).get("pid")
+        except (OSError, json.JSONDecodeError):
+            owner = None
+        if not alive(owner):
+            LOCK.unlink(missing_ok=True)
+            return acquire_lock()
+        raise SystemExit(f"web product migration already running: pid={owner}")
+    os.write(handle, (json.dumps({"pid": os.getpid(), "startedAt": now()}) + "\n").encode())
+    os.close(handle)
+
+
+def file_hash(path: str) -> str | None:
+    absolute = ROOT / path
+    if not absolute.is_file():
+        return None
+    return hashlib.sha256(absolute.read_bytes()).hexdigest()
+
+
+def refresh_modules(backlog: dict) -> None:
+    states = {task["id"]: task["status"] for task in backlog.get("tasks", [])}
+    for module in backlog.get("workflowModules", []):
+        task_ids = module.get("taskIds", [])
+        if not task_ids:
+            continue
+        task_states = [states.get(task_id, "pending") for task_id in task_ids]
+        if all(state == "implemented" for state in task_states):
+            module["status"] = "implemented"
+        elif any(state == "active" for state in task_states):
+            module["status"] = "active"
+        elif any(state == "blocked" for state in task_states):
+            module["status"] = "blocked"
+        else:
+            module["status"] = "partial"
+
+
+def write_status(backlog: dict, status: str, **extra: object) -> None:
+    tasks = backlog.get("tasks", [])
+    modules = backlog.get("workflowModules", [])
+    write_json(STATUS, {
+        "schemaVersion": 1,
+        "pid": os.getpid(),
+        "status": status,
+        "updatedAt": now(),
+        "tasks": {
+            "total": len(tasks),
+            "implemented": sum(task.get("status") == "implemented" for task in tasks),
+            "active": [task["id"] for task in tasks if task.get("status") == "active"],
+            "pending": sum(task.get("status") == "pending" for task in tasks),
+            "blocked": sum(task.get("status") == "blocked" for task in tasks),
+        },
+        "workflowModules": {
+            "total": len(modules),
+            "implemented": sum(module.get("status") == "implemented" for module in modules),
+            "active": [module["id"] for module in modules if module.get("status") == "active"],
+            "blocked": [module["id"] for module in modules if module.get("status") == "blocked"],
+        },
+        **extra,
+    })
+
+
+def prompt(task: dict) -> str:
+    return f"""Implement exactly one Web Blender product workflow task in {ROOT}.
+
+Task: {task['id']} - {task['title']}
+Commands to expose: {', '.join(task['commands'])}
+Owned paths: {', '.join(task['ownedPaths'])}
+Implementation requirement: {task['implementation']}
+
+Rules:
+- Follow the repository AGENTS.md minimal-context instructions first.
+- Preserve all existing dirty changes. Do not revert or rewrite unrelated work.
+- Read only the owned paths and their direct imported types/helpers.
+- Implement production behavior through the existing Blender Main/WASM command routes; do not simulate Blender behavior in JavaScript.
+- Do not run tests, builds, typecheck, lint, Chromium, full validators, or coverage binding. The user explicitly deferred verification to real usage.
+- Do not modify docs/EXECUTION_QUEUE.md, nextTask.md, source coverage ledgers, manifests, or this backlog.
+- Do not commit or push. The outer unattended scheduler owns rotation and repository handoff.
+- Finish the implementation in this turn. Do not stop at a plan.
+- At the end write {task['resultPath']} as JSON with schemaVersion=1, task={task['id']!r}, status='implemented', changedFiles, and a concise summary.
+"""
+
+
+def run_task(task: dict, heartbeat_seconds: float) -> tuple[int, bool]:
+    global ACTIVE_PROCESS
+    before = {path: file_hash(path) for path in task["ownedPaths"]}
+    result_path = ROOT / task["resultPath"]
+    result_path.unlink(missing_ok=True)
+    log_path = LOG_ROOT / f"{task['id']}.codex.jsonl"
+    LOG_ROOT.mkdir(parents=True, exist_ok=True)
+    command = [
+        "codex", "exec", "--approve-for-me", "--sandbox", "workspace-write",
+        "--ephemeral", "--cd", str(ROOT), "--json", "-",
+    ]
+    with log_path.open("w", encoding="utf-8") as log:
+        process = subprocess.Popen(
+            command,
+            cwd=ROOT,
+            stdin=subprocess.PIPE,
+            stdout=log,
+            stderr=subprocess.STDOUT,
+            text=True,
+            start_new_session=True,
+        )
+        ACTIVE_PROCESS = process
+        assert process.stdin is not None
+        process.stdin.write(prompt(task))
+        process.stdin.close()
+        while process.poll() is None:
+            write_status(read_json(BACKLOG), "implementing", activeTask=task["id"], childPid=process.pid, log=str(log_path.relative_to(ROOT)))
+            time.sleep(heartbeat_seconds)
+        code = process.returncode or 0
+        ACTIVE_PROCESS = None
+    after = {path: file_hash(path) for path in task["ownedPaths"]}
+    changed = any(before[path] != after[path] for path in before)
+    result_ok = False
+    if result_path.is_file():
+        try:
+            result = read_json(result_path)
+            result_ok = result.get("task") == task["id"] and result.get("status") == "implemented"
+        except json.JSONDecodeError:
+            pass
+    return code, changed and result_ok
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--once", action="store_true", help="run at most one task")
+    parser.add_argument("--max-attempts", type=int, default=3)
+    parser.add_argument("--heartbeat-seconds", type=float, default=5.0)
+    args = parser.parse_args()
+    if args.max_attempts < 1 or args.heartbeat_seconds <= 0:
+        parser.error("limits must be positive")
+    acquire_lock()
+    PID.write_text(f"{os.getpid()}\n", encoding="ascii")
+    stopped = False
+    def stop(signum: int, _frame: object) -> None:
+        nonlocal stopped
+        stopped = True
+        if ACTIVE_PROCESS and ACTIVE_PROCESS.poll() is None:
+            try:
+                os.killpg(ACTIVE_PROCESS.pid, signum)
+            except ProcessLookupError:
+                pass
+
+    signal.signal(signal.SIGINT, stop)
+    signal.signal(signal.SIGTERM, stop)
+    try:
+        ran = 0
+        while not stopped:
+            backlog = read_json(BACKLOG)
+            for task in backlog.get("tasks", []):
+                if task.get("status") == "active":
+                    task["status"] = "pending"
+            refresh_modules(backlog)
+            pending = next((task for task in backlog.get("tasks", []) if task.get("status") == "pending"), None)
+            if pending is None:
+                refresh_modules(backlog)
+                write_json(BACKLOG, backlog)
+                blocked = [task for task in backlog.get("tasks", []) if task.get("status") == "blocked"]
+                write_status(backlog, "blocked" if blocked else "implementation_complete")
+                return 2 if blocked else 0
+            pending["status"] = "active"
+            pending["attempts"] = int(pending.get("attempts", 0)) + 1
+            pending["startedAt"] = now()
+            refresh_modules(backlog)
+            write_json(BACKLOG, backlog)
+            write_status(backlog, "implementing", activeTask=pending["id"])
+            code, implemented = run_task(pending, args.heartbeat_seconds)
+            backlog = read_json(BACKLOG)
+            task = next(item for item in backlog["tasks"] if item["id"] == pending["id"])
+            if code == 0 and implemented:
+                task["status"] = "implemented"
+                task["completedAt"] = now()
+                task.pop("lastError", None)
+            elif int(task.get("attempts", 0)) >= args.max_attempts:
+                task["status"] = "blocked"
+                task["lastError"] = f"codex exit={code}; implementation result or owned-path change missing"
+            else:
+                task["status"] = "pending"
+                task["lastError"] = f"codex exit={code}; retry scheduled"
+            refresh_modules(backlog)
+            write_json(BACKLOG, backlog)
+            ran += 1
+            if args.once and ran >= 1:
+                write_status(backlog, "idle", lastTask=pending["id"], exitCode=code)
+                return 0 if implemented else 2
+    finally:
+        PID.unlink(missing_ok=True)
+        LOCK.unlink(missing_ok=True)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

my-tools/show_blender_web_scheduler.py

@@ -13,6 +13,7 @@ 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
@@ -39,15 +40,22 @@ def main() -> int:
             "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"] = (
-            "PLANNING" if status.get("state") == "IDLE" and watchdog_alive and watchdog.get("status") == "waiting_for_source_batch"
+            "IMPLEMENTING" if product_alive and product.get("status") == "implementing"
+            else "COMPLETE" if product.get("status") == "implementation_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["progress"] = project_progress()
         status["fastPath"] = fast_path_status(status)
         if args.json:
             print(json.dumps(status, ensure_ascii=False, separators=(",", ":")))
@@ -74,11 +82,16 @@ def main() -> int:
         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") == "PLANNING":
+        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', '')}")
@@ -109,6 +122,8 @@ def _queue_task() -> str | 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)
@@ -166,6 +181,20 @@ def project_progress() -> dict:
                 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(),
@@ -192,6 +221,23 @@ def project_progress() -> dict:
             "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,
+        },
     }
 
 
@@ -201,6 +247,7 @@ def fast_path_status(status: dict) -> dict:
     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": {
@@ -225,7 +272,7 @@ def fast_path_status(status: dict) -> dict:
             "nextAction": planner.get("nextAction"),
         },
         "commandBudget": {"maxCommands": 48, "maxReadCommands": 16},
-        "verificationPolicy": "incremental-context-only-until-final-audit",
+        "verificationPolicy": product_backlog.get("verificationPolicy", "incremental-context-only-until-final-audit"),
         "wasmModules": modules,
     }
 
@@ -277,17 +324,17 @@ def format_progress(progress: dict, fast_path: dict | None = None) -> list[str]:
     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 [
         "",
-        "Project migration progress",
-        f"  dynamic units: {progress.get('verifiedUnits', 0)}/{progress.get('totalUnits', 0)} verified ({progress.get('verifiedPercent', 0)}%), remaining={progress.get('unverifiedUnits', 0)}",
+        "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"  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')}",
+        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')}",

my-tools/start_unattended.sh

@@ -10,6 +10,7 @@ WATCHDOG_INTERVAL="${WBV2_WATCHDOG_INTERVAL:-10}"
 WATCHDOG_RESTART_DELAY="${WBV2_WATCHDOG_RESTART_DELAY:-5}"
 AUTO_AUTHORIZE="${WBV2_AUTO_AUTHORIZE:-1}"
 WAIT_ON_EMPTY="${WBV2_WAIT_ON_EMPTY:-1}"
+MIGRATION_MODE="${WBV2_MIGRATION_MODE:-product}"
 
 # Keep repository evidence on disk while placing compiler/transient caches in
 # RAM. Eight compile jobs is the repository's stability limit.
@@ -75,5 +76,25 @@ start_watchdog() {
   printf 'watchdog launch pid: %s\n' "$!"
 }
 
-start_scheduler
-start_watchdog
+start_product_migration() {
+  local pid_file="$CONTROL_DIR/web-product-migration.pid"
+  if test -f "$pid_file"; then
+    local product_pid
+    product_pid="$(tr -d '[:space:]' < "$pid_file")"
+    if test -n "$product_pid" && kill -0 "$product_pid" 2>/dev/null; then
+      printf 'product migration already running: %s\n' "$product_pid"
+      return
+    fi
+    mv "$pid_file" "$pid_file.stale-$(date +%s)"
+  fi
+  nohup setsid python3 "$ROOT_DIR/my-tools/run_web_product_migration.py" \
+    > "$CONTROL_DIR/web-product-migration.stdout.log" 2>&1 < /dev/null &
+  printf 'product migration launch pid: %s\n' "$!"
+}
+
+if test "$MIGRATION_MODE" = "source-verification"; then
+  start_scheduler
+  start_watchdog
+else
+  start_product_migration
+fi