INNER CODE UNIT · Python

mark_slot_completed

lij768423-svg/grok-register-panel · batch_supervisor.py:99

def mark_slot_completed(slots: int = 1) -> None:
    """Persist completed task slots for the supervising parent process."""
    raw_path = str(os.environ.get(PROGRESS_ENV, "") or "").strip()
    if not raw_path:
        return
    increment = max(0, int(slots or 0))
    if increment <= 0:
        return

    path = Path(raw_path)
    lock_path = path.with_name(f"{path.name}.lock")
    with _PROGRESS_LOCK:
        with exclusive_file_lock(lock_path):
            try:
                data = json.loads(path.read_text(encoding="utf-8"))
            except (OSError, ValueError, TypeError, json.JSONDecodeError):
                data = {}
            completed = max(0, int(data.get("completed", 0) or 0)) + increment

View source record →