chatgpt-register/ops/service.py

"""Background task helpers for the single-process zhuce6 runtime."""

from __future__ import annotations

from collections import deque
from datetime import datetime
import threading
import time
from typing import Callable


def _isoformat_timestamp(value: float | None) -> str | None:
    if value is None:
        return None
    return datetime.fromtimestamp(value).isoformat(timespec="seconds")


class RepeatedTask:
    def __init__(self, name: str, fn: Callable[[], None], interval_seconds: int) -> None:
        self.name = name
        self.fn = fn
        self.interval_seconds = max(1, int(interval_seconds))
        self._thread: threading.Thread | None = None
        self._stop_event = threading.Event()
        self._lock = threading.Lock()
        self._run_count = 0
        self._success_count = 0
        self._failure_count = 0
        self._last_started_at: float | None = None
        self._last_finished_at: float | None = None
        self._last_duration_seconds: float | None = None
        self._last_error: str | None = None
        self._next_run_at: float | None = None
        self._is_running = False
        self._recent_runs: deque[dict[str, object]] = deque(maxlen=20)

    def start(self) -> None:
        if self._thread and self._thread.is_alive():
            return
        self._stop_event.clear()
        with self._lock:
            self._next_run_at = time.time()
        self._thread = threading.Thread(target=self._run, daemon=True, name=f"zhuce6-{self.name}")
        self._thread.start()

    def stop(self) -> None:
        self._stop_event.set()
        if self._thread and self._thread.is_alive():
            self._thread.join(timeout=1)

    def _run(self) -> None:
        while not self._stop_event.is_set():
            cycle_started = time.time()
            cycle_error: str | None = None
            with self._lock:
                self._is_running = True
                self._run_count += 1
                self._last_started_at = cycle_started
                self._last_error = None
            try:
                self.fn()
            except Exception as exc:
                with self._lock:
                    self._failure_count += 1
                    self._last_error = str(exc)
                    cycle_error = str(exc)
                print(f"[zhuce6:{self.name}] background task error: {exc}")
            else:
                with self._lock:
                    self._success_count += 1
                    self._last_error = None
            finally:
                cycle_finished = time.time()
                with self._lock:
                    self._is_running = False
                    self._last_finished_at = cycle_finished
                    self._last_duration_seconds = round(cycle_finished - cycle_started, 3)
                    self._recent_runs.append(
                        {
                            "started_at": _isoformat_timestamp(cycle_started),
                            "finished_at": _isoformat_timestamp(cycle_finished),
                            "duration_seconds": self._last_duration_seconds,
                            "status": "failed" if cycle_error else "completed",
                            "error": cycle_error,
                        }
                    )
            elapsed = time.time() - cycle_started
            wait_seconds = max(0.0, self.interval_seconds - elapsed)
            with self._lock:
                self._next_run_at = time.time() + wait_seconds
            if self._stop_event.wait(wait_seconds):
                break

    def snapshot(self) -> dict[str, object]:
        with self._lock:
            if self._is_running:
                status = "running"
            elif self._run_count == 0:
                status = "pending"
            elif self._last_error:
                status = "degraded"
            else:
                status = "healthy"
            return {
                "name": self.name,
                "status": status,
                "interval_seconds": self.interval_seconds,
                "run_count": self._run_count,
                "success_count": self._success_count,
                "failure_count": self._failure_count,
                "is_running": self._is_running,
                "last_started_at": _isoformat_timestamp(self._last_started_at),
                "last_finished_at": _isoformat_timestamp(self._last_finished_at),
                "last_duration_seconds": self._last_duration_seconds,
                "last_error": self._last_error,
                "next_run_at": _isoformat_timestamp(self._next_run_at),
                "recent_runs": list(self._recent_runs),
            }