ElebobaandClaude Opus 5.5
711c646876
CI / ruff · pytest (pull_request) Successful in 18s
- app/core/clock.py: now()/today(), virtual offset (advance/set_now/reset), pluggable offset persistence hook (ADR-005) - config/rules.yaml + app/core/rules.py: YAML declares rules (params, actions, levels, reason_ru), Python evaluators implement predicates by id; dedup of open alerts, crisis never suppressed; ADR-006 colours (ADR-007) - tests: per-rule unit tests, dedup, clock, 13-day demo persona trajectory Refs #5 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
123 lines
3.8 KiB
Python
123 lines
3.8 KiB
Python
"""The only source of time in the system (ADR-005, guardrail #4).
|
|
|
|
Real mode (`DEMO_CLOCK=false`): `now()` is the wall clock in `settings.tz`.
|
|
Virtual mode (`DEMO_CLOCK=true`): `now()` is the wall clock shifted by an offset. The demo console moves
|
|
time forward with `advance(days=1)` ("next day"), so 13 days of therapy fit into a 3-minute demo while the
|
|
scheduler and the rule windows ("3 days in a row", "7 days") keep working on the shifted time.
|
|
|
|
The clock keeps an *offset*, not a frozen instant: virtual time keeps ticking between button presses, so
|
|
scheduled jobs and timestamps remain ordered. The offset can be persisted through a pluggable hook
|
|
(`register_persistence`) so the DB layer can store it and survive restarts; by default it lives in memory.
|
|
|
|
Business logic must never call `datetime.now()` directly — it receives time from here (or as a parameter).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from collections.abc import Callable
|
|
from datetime import date, datetime, timedelta, timezone, tzinfo
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
from app.config import get_settings
|
|
|
|
LoadOffset = Callable[[], timedelta | None]
|
|
SaveOffset = Callable[[timedelta], None]
|
|
|
|
_lock = threading.RLock()
|
|
_offset = timedelta(0)
|
|
_save: SaveOffset | None = None
|
|
|
|
# Moscow has had a fixed UTC+3 without DST since 2014; used only if the OS has no tz database
|
|
# (slim Docker images may ship without tzdata).
|
|
_FALLBACK_OFFSETS = {"Europe/Moscow": timedelta(hours=3)}
|
|
|
|
|
|
def _tz() -> tzinfo:
|
|
name = get_settings().tz
|
|
try:
|
|
return ZoneInfo(name)
|
|
except ZoneInfoNotFoundError:
|
|
return timezone(_FALLBACK_OFFSETS.get(name, timedelta(0)), name)
|
|
|
|
|
|
def _virtual() -> bool:
|
|
return get_settings().demo_clock
|
|
|
|
|
|
def _real_now() -> datetime:
|
|
# The single sanctioned wall-clock read in the codebase.
|
|
return datetime.now(_tz())
|
|
|
|
|
|
def _require_virtual() -> None:
|
|
if not _virtual():
|
|
raise RuntimeError("Virtual clock is disabled (DEMO_CLOCK=false); time cannot be moved")
|
|
|
|
|
|
def _store(offset: timedelta) -> None:
|
|
global _offset
|
|
_offset = offset
|
|
if _save is not None:
|
|
_save(offset)
|
|
|
|
|
|
def now() -> datetime:
|
|
"""Current time, tz-aware in `settings.tz`; shifted by the demo offset in virtual mode."""
|
|
real = _real_now()
|
|
if not _virtual():
|
|
return real
|
|
with _lock:
|
|
return real + _offset
|
|
|
|
|
|
def today() -> date:
|
|
"""Current local date according to `now()`."""
|
|
return now().date()
|
|
|
|
|
|
def offset() -> timedelta:
|
|
"""Current virtual offset (always zero in real mode)."""
|
|
if not _virtual():
|
|
return timedelta(0)
|
|
with _lock:
|
|
return _offset
|
|
|
|
|
|
def advance(days: float = 0, hours: float = 0, minutes: float = 0) -> datetime:
|
|
"""Move virtual time forward (or backward with negative values). Returns the new `now()`."""
|
|
_require_virtual()
|
|
with _lock:
|
|
_store(_offset + timedelta(days=days, hours=hours, minutes=minutes))
|
|
return now()
|
|
|
|
|
|
def set_now(dt: datetime) -> datetime:
|
|
"""Jump virtual time to `dt` (naive values are interpreted in `settings.tz`). Returns the new `now()`."""
|
|
_require_virtual()
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=_tz())
|
|
with _lock:
|
|
_store(dt - _real_now())
|
|
return now()
|
|
|
|
|
|
def reset() -> None:
|
|
"""Return virtual time to the wall clock (offset = 0)."""
|
|
with _lock:
|
|
_store(timedelta(0))
|
|
|
|
|
|
def register_persistence(load: LoadOffset | None, save: SaveOffset | None) -> None:
|
|
"""Plug in offset storage (e.g. a DB row). `load` is called once immediately; `save` on every change.
|
|
|
|
Pass `(None, None)` to go back to in-memory only.
|
|
"""
|
|
global _offset, _save
|
|
with _lock:
|
|
_save = save
|
|
if load is not None:
|
|
stored = load()
|
|
if stored is not None:
|
|
_offset = stored
|