execution_trace.py

"""Durable, secret-safe progress reporting for maintainer controller runs."""

from __future__ import annotations

import fcntl
import json
import uuid
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterator


UTC = timezone.utc
TRACE_FILE_NAME = "execution.json"
MAX_EVENTS = 160
MAX_HISTORY = 30
TERMINAL_STATUSES = {"completed", "failed", "blocked"}


def now_iso() -> str:
    return datetime.now(UTC).isoformat()


def default_trace() -> dict[str, Any]:
    return {"schema_version": 1, "current": None, "history": []}


class ExecutionTrace:
    """A small cross-process event log shared by the live worker and Console."""

    def __init__(self, state_path: Path, run_id: str | None = None, trace_path: Path | None = None):
        self.state_path = Path(state_path).expanduser()
        self.path = (trace_path or self.state_path / TRACE_FILE_NAME).expanduser()
        self.lock_path = self.path.with_suffix(self.path.suffix + ".lock")
        self.run_id = run_id

    def _read_unlocked(self) -> dict[str, Any]:
        if not self.path.exists():
            return default_trace()
        try:
            value = json.loads(self.path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            return default_trace()
        if not isinstance(value, dict):
            return default_trace()
        merged = default_trace()
        merged.update(value)
        if not isinstance(merged.get("history"), list):
            merged["history"] = []
        return merged

    @contextmanager
    def _locked(self) -> Iterator[None]:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        with self.lock_path.open("a+", encoding="utf-8") as lock_file:
            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
            try:
                yield
            finally:
                fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)

    def _write_unlocked(self, value: dict[str, Any]) -> None:
        temporary = self.path.with_suffix(self.path.suffix + ".tmp")
        temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        temporary.replace(self.path)

    def snapshot(self) -> dict[str, Any]:
        with self._locked():
            return self._read_unlocked()

    @staticmethod
    def _event(
        current: dict[str, Any],
        phase: str,
        status: str,
        message: str,
        command: str | None,
        details: dict[str, Any] | None,
    ) -> dict[str, Any]:
        events = list(current.get("events") or [])
        event = {
            "sequence": len(events) + 1,
            "at": now_iso(),
            "phase": phase,
            "status": status,
            "message": message,
        }
        if command:
            event["command"] = command
        if details:
            event["details"] = dict(details)
        events.append(event)
        current["events"] = events[-MAX_EVENTS:]
        current["phase"] = phase
        current["status"] = status if phase in {"completed", "failed", "blocked"} else "running"
        current["message"] = message
        current["updated_at"] = event["at"]
        if details:
            context = dict(current.get("context") or {})
            context.update(details)
            current["context"] = context
        return event

    def start(
        self,
        kind: str = "live_cycle",
        trigger: str = "systemd timer",
        message: str = "Maintainer cycle started",
    ) -> str:
        self.run_id = self.run_id or f"{kind}-{uuid.uuid4().hex[:12]}"
        started_at = now_iso()
        current: dict[str, Any] = {
            "id": self.run_id,
            "kind": kind,
            "trigger": trigger,
            "status": "running",
            "phase": "starting",
            "message": message,
            "started_at": started_at,
            "updated_at": started_at,
            "finished_at": None,
            "context": {},
            "events": [],
        }
        self._event(current, "starting", "running", message, "live_cycle.py", None)
        with self._locked():
            state = self._read_unlocked()
            state["current"] = current
            self._write_unlocked(state)
        return self.run_id

    def event(
        self,
        phase: str,
        status: str = "running",
        message: str = "",
        command: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> dict[str, Any] | None:
        with self._locked():
            state = self._read_unlocked()
            current = state.get("current")
            if not isinstance(current, dict):
                return None
            if self.run_id and current.get("id") != self.run_id:
                return None
            self._event(current, phase, status, message, command, details)
            if phase in {"completed", "failed", "blocked"} and status in TERMINAL_STATUSES:
                current["finished_at"] = current.get("updated_at") or now_iso()
                history = list(state.get("history") or [])
                history.append(dict(current))
                state["history"] = history[-MAX_HISTORY:]
            state["current"] = current
            self._write_unlocked(state)
            return dict(current)

    def finish(
        self,
        status: str,
        message: str,
        details: dict[str, Any] | None = None,
    ) -> dict[str, Any] | None:
        phase = "completed" if status == "completed" else "failed" if status == "failed" else "blocked"
        return self.event(phase, status, message, details=details)