code/self_healing_finetuning/observability/progress.py

from __future__ import annotations

from dataclasses import dataclass


@dataclass(frozen=True)
class ProgressState:
    run_id: str
    cycle: int
    stage: str
    step: str
    progress_step_pct: float
    progress_stage_pct: float
    progress_cycle_pct: float
    train_provider: str
    infer_provider: str
    model_candidate: str
    status: str = "running"
    blocker: str | None = None
    next_action: str | None = None

    def as_event(self) -> dict[str, object]:
        return {
            "event_type": "heartbeat",
            "run_id": self.run_id,
            "cycle": self.cycle,
            "stage": self.stage,
            "step": self.step,
            "progress_step_pct": self.progress_step_pct,
            "progress_stage_pct": self.progress_stage_pct,
            "progress_cycle_pct": self.progress_cycle_pct,
            "train_provider": self.train_provider,
            "infer_provider": self.infer_provider,
            "model_candidate": self.model_candidate,
            "status": self.status,
            "blocker": self.blocker,
            "next_action": self.next_action,
        }

    def console_line(self) -> str:
        blocker = self.blocker or "none"
        next_action = self.next_action or "none"
        return (
            f"[SHFT heartbeat] run={self.run_id} cycle={self.cycle} stage={self.stage} "
            f"step={self.step} progress_step={self.progress_step_pct:.0f}% "
            f"progress_stage={self.progress_stage_pct:.0f}% progress_cycle={self.progress_cycle_pct:.0f}% "
            f"provider=train:{self.train_provider},infer:{self.infer_provider} model={self.model_candidate} "
            f"status={self.status} blocker={blocker} next=\"{next_action}\""
        )