clock.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/clock.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_storage_foundation.py"
"""
Injectable clock service for registry code.
 
Purpose
-------
Centralize all wall-clock and monotonic reads used by the registry so
tests can pin time and production code never reads time directly.
 
Responsibilities
----------------
- Define the two-read :class:`Clock` contract.
- Provide :class:`RealClock` backed by Python timers.
- Provide :class:`FakeClock` for deterministic tests.
- Provide wall-time arithmetic helpers so other registry modules do not
  import time or datetime directly.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- This is the only module under ``oc_chat.registry`` allowed to import
  ``time`` or ``datetime``.
- ``now_wall`` returns explicit RFC3339 UTC ending in ``Z``.
- ``state_version`` ordering is an integer counter; monotonic timing is
  available for timeout logic and wall time is only stored as text.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- time (wall-clock and monotonic nanosecond reads for RealClock)
- datetime (RFC3339 UTC formatting, parsing and wall-time arithmetic)
"""
from __future__ import annotations
 
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
import time
 
from oc_diagnostics import _dbg
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
class Clock(ABC):
    """Two-read clock contract required by the registry spec."""
 
    @abstractmethod
    def now_wall(self) -> str:
        """Return the current wall time as RFC3339 UTC."""
 
    @abstractmethod
    def now_monotonic_ns(self) -> int:
        """Return a monotonic nanosecond counter."""
 
 
def _rfc3339_from_time_ns(value: int) -> str:
    seconds, nanoseconds = divmod(value, 1_000_000_000)
    dt = datetime.fromtimestamp(seconds, tz=timezone.utc).replace(
        microsecond=nanoseconds // 1_000
    )
    return dt.isoformat(timespec="microseconds").replace("+00:00", "Z")
 
 
def wall_plus_seconds(wall: str, seconds: int) -> str:
    """Return ``wall`` plus ``seconds`` as RFC3339 UTC."""
 
    if seconds < 0:
        raise ValueError("seconds must be non-negative")
    normalized = wall[:-1] + "+00:00" if wall.endswith("Z") else wall
    dt = datetime.fromisoformat(normalized)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    else:
        dt = dt.astimezone(timezone.utc)
    return (
        (dt + timedelta(seconds=seconds))
        .isoformat(timespec="microseconds")
        .replace("+00:00", "Z")
    )
 
 
class RealClock(Clock):
    """Clock backed by Python's wall and monotonic timers."""
 
    def now_wall(self) -> str:
        return _rfc3339_from_time_ns(time.time_ns())
 
    def now_monotonic_ns(self) -> int:
        return time.monotonic_ns()
 
 
@dataclass
class FakeClock(Clock):
    """Pinned clock for deterministic registry tests."""
 
    wall: str = "2026-05-20T18:30:41.123456Z"
    monotonic_ns: int = 0
 
    def now_wall(self) -> str:
        return self.wall
 
    def now_monotonic_ns(self) -> int:
        return self.monotonic_ns
 
    def set_wall(self, value: str) -> None:
        self.wall = value
 
    def set_monotonic_ns(self, value: int) -> None:
        self.monotonic_ns = value