INNER CODE UNIT · Python

initialize_progress

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

def initialize_progress(path: str | os.PathLike[str], target: int) -> Path:
    progress_path = Path(path)
    atomic_write_json(
        progress_path,
        {
            "completed": 0,
            "target": max(0, int(target)),
            "updated_at": time.time(),
        },
    )
    return progress_path


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

View source record →