live_app.py

"""Live-demo ASGI entrypoint for the self-reflective-apis research server.

    uvicorn live_app:app --host 127.0.0.1 --port 8100

This file is OURS. Nothing inside ``_clone/`` is modified, ever -- it is only
imported from, and extra routes are composed onto the upstream FastAPI app
object at runtime.

What it does
------------
1. Puts ``_clone/apis`` on ``sys.path`` and imports the upstream recipe app
   (``recipe.main:app``), the self-contained domain on the pinned default
   branch.
2. Imports the Acme ``required_promo`` policy module, which registers the Acme
   CSM/promo table ``active_csm_codes`` in the shared ``TableRegistry``
   singleton. Both apps and every ``table_versions`` block read that same
   singleton, so the Acme table becomes visible in recipe responses and
   reloadable through the recipe app's ``/admin/reload-table`` -- without
   importing the Acme FastAPI app, which does not import at this commit
   (see RUNBOOK_live.md, "Upstream surprises").
3. Rebinds the registry path for the drifting table from the read-only clone
   to a writable copy under ``runtime_data/`` so that drift rotation never
   writes into ``_clone/``. ``drift_cron.py`` swaps that same file. Which
   table drifts is chosen by the LIVE_DRIFT_* variables; the default is the
   recipe table the API reads when it builds a recovery suggestion, so a
   rotation can change an answer a client is already holding.
4. Adds two routes of our own: ``GET /live/status``, describing what is loaded
   and on what public cadence it drifts, and ``GET /live``, a read-only status
   page that renders the same payload.

The live table file is the single source of truth for drift position: its
``version`` major number is the snapshot index. There is no separate cursor to
get out of sync with, and a restart never rewinds a rotation.
"""
from __future__ import annotations

import json
import os
import shutil
import sys
from datetime import datetime, timezone
from pathlib import Path

LIVE_ROOT = Path(__file__).resolve().parent

# Never write .pyc files into the read-only upstream clone. Belt and braces:
# dont_write_bytecode suppresses the write; pycache_prefix redirects any write
# that still happens (a launcher may reset the flag) into our own tree. Both
# are consulted at import time, so they are set before anything under _clone/
# is imported.
sys.dont_write_bytecode = True
sys.pycache_prefix = str(LIVE_ROOT / "runtime_data" / "pycache")
os.environ.setdefault("PYTHONDONTWRITEBYTECODE", "1")

# The startup banner prints the install path, which contains non-ASCII on this
# host; a cp1252 console would raise UnicodeEncodeError and abort the server.
for _stream in (sys.stdout, sys.stderr):
    try:
        _stream.reconfigure(encoding="utf-8", errors="replace")
    except (AttributeError, ValueError):  # pragma: no cover
        pass

UPSTREAM_ROOT = Path(os.environ.get("LIVE_UPSTREAM_ROOT", LIVE_ROOT / "_clone")).resolve()
CLONE_APIS = UPSTREAM_ROOT / "apis"
SNAPSHOT_DIR = Path(os.environ.get("LIVE_SNAPSHOT_DIR", LIVE_ROOT / "snapshots")).resolve()
RUNTIME_DATA = Path(os.environ.get("LIVE_RUNTIME_DATA", LIVE_ROOT / "runtime_data")).resolve()
# Set to 1 to re-seed the writable mirror from the clone at startup, i.e. to
# rewind drift back to v1. Off by default so a restart never loses a rotation.
RESET_TABLES = os.environ.get("LIVE_RESET_TABLES", "") == "1"

# Which table drifts. TABLE_NAME is the registry key; FILE_STEM is the file
# name on disk, and the two differ for the recipe table. The acme profile is
# LIVE_DRIFT_TABLE=active_csm_codes LIVE_DRIFT_FILE=active_csm_codes
# LIVE_DRIFT_SRC=acme_billing/data LIVE_DRIFT_MIRROR=acme.
TABLE_NAME = os.environ.get("LIVE_DRIFT_TABLE", "incompatible_ingredients")
FILE_STEM = os.environ.get("LIVE_DRIFT_FILE", "incompatible_combinations")
MIRROR_DIR = os.environ.get("LIVE_DRIFT_MIRROR", "recipe")
SRC_REL = os.environ.get("LIVE_DRIFT_SRC", "recipe/data")
# Snapshot the mirror starts from, i.e. where drift resumes after a reset.
SEED_INDEX = int(os.environ.get("LIVE_SEED_INDEX", "1"))
LIVE_TABLE = RUNTIME_DATA / MIRROR_DIR / (FILE_STEM + ".json")

# Declarative only: what we advertise as the rotation cadence. Keep it truthful
# -- it must match whatever schedules drift_cron.py.
DRIFT_SCHEDULE = os.environ.get("LIVE_DRIFT_SCHEDULE", "daily at 03:00 host local time")

# Upstream (read-only) sources used once to seed our snapshot ring.
_UPSTREAM_DATA = CLONE_APIS.joinpath(*SRC_REL.split("/"))
_UPSTREAM_SNAPSHOTS = _UPSTREAM_DATA / "snapshots"

if not CLONE_APIS.is_dir():
    raise SystemExit(
        "Upstream clone not found at {}. Run:\n"
        "  git clone https://github.com/arquicanedo/self-reflective-apis _clone".format(CLONE_APIS)
    )
if str(CLONE_APIS) not in sys.path:
    sys.path.insert(0, str(CLONE_APIS))


def seed_snapshots() -> list[Path]:
    """Mirror the upstream snapshot ring into snapshots/ on first run.

    One-way copy (clone -> ours); nothing is ever written back. If snapshots/
    is already populated, it is left alone so an operator can extend or edit
    the ring without the server clobbering it.
    """
    SNAPSHOT_DIR.mkdir(parents=True, exist_ok=True)
    ring = sorted(SNAPSHOT_DIR.glob(FILE_STEM + "_v*.json"))
    if ring:
        return ring

    shipped = _UPSTREAM_DATA / (FILE_STEM + ".json")
    if shipped.is_file():
        # The repo's shipped table is version 1.0.0 == snapshot v1.
        shutil.copyfile(shipped, SNAPSHOT_DIR / (FILE_STEM + "_v1.json"))
    for src in sorted(_UPSTREAM_SNAPSHOTS.glob(FILE_STEM + "_v*.json")):
        shutil.copyfile(src, SNAPSHOT_DIR / src.name)

    ring = sorted(SNAPSHOT_DIR.glob(FILE_STEM + "_v*.json"))
    if len(ring) < 2:
        raise SystemExit(
            "Need at least two snapshots of {!r} in {} for drift to mean anything.".format(
                TABLE_NAME, SNAPSHOT_DIR
            )
        )
    return ring


def seed_live_table() -> Path:
    """Create the writable live table if absent.

    An existing file is never overwritten unless LIVE_RESET_TABLES=1, so a
    server restart resumes at whatever snapshot drift_cron last rotated to.
    """
    LIVE_TABLE.parent.mkdir(parents=True, exist_ok=True)
    if RESET_TABLES or not LIVE_TABLE.exists():
        shutil.copyfile(SNAPSHOT_DIR / (FILE_STEM + "_v%d.json" % SEED_INDEX), LIVE_TABLE)
    return LIVE_TABLE


# --------------------------------------------------------------------------
# Compose the app
# --------------------------------------------------------------------------
_ring = seed_snapshots()
seed_live_table()

from recipe.main import app                   # noqa: E402  upstream app object
from shared.table_registry import registry    # noqa: E402  shared singleton

try:
    from acme_billing.policies import required_promo  # noqa: F401,E402
    ACME_TABLE_LOADED = True
except Exception as _exc:  # pragma: no cover
    ACME_TABLE_LOADED = False
    print("[live] WARNING: Acme promo table not loaded: %r" % (_exc,), file=sys.stderr)

# Point the registry at our writable copy instead of the file inside _clone/,
# then re-read so the served version reflects whatever drift_cron last wrote.
if TABLE_NAME in registry._paths:  # noqa: SLF001 -- deliberate, documented
    registry._paths[TABLE_NAME] = LIVE_TABLE
    registry.reload(TABLE_NAME)
else:
    registry.load(TABLE_NAME, LIVE_TABLE)

print(
    "[live] %s v%s from %s | ring of %d | drift: %s"
    % (TABLE_NAME, registry.version(TABLE_NAME), LIVE_TABLE, len(_ring), DRIFT_SCHEDULE),
    file=sys.stderr,
    flush=True,
)


from fastapi.responses import HTMLResponse  # noqa: E402

_STARTED = datetime.now(timezone.utc)
_ROTATIONS = []
_LAST = {"version": None, "data": None}
_PAGE = LIVE_ROOT / "live_page.html"
# API calls that did not come from the status page's own polling loop; the page
# marks its probes with an x-live-probe header, so this counts try-it clicks
# and external curls.
_VISITOR_CALLS = {"n": 0}


@app.middleware("http")
async def _count_visitor_calls(request, call_next):
    if request.url.path.startswith("/api/") and "x-live-probe" not in request.headers:
        _VISITOR_CALLS["n"] += 1
    return await call_next(request)


def _iso(dt):
    return dt.astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")


def _upstream_commit():
    """Short-circuit git: read .git/HEAD by hand rather than shelling out."""
    git = UPSTREAM_ROOT / ".git"
    try:
        head = (git / "HEAD").read_text(encoding="utf-8").strip()
    except OSError:
        return None
    if not head.startswith("ref: "):
        return head
    ref = head[5:].strip()
    try:
        return (git / ref).read_text(encoding="utf-8").strip()
    except OSError:
        pass
    try:
        for line in (git / "packed-refs").read_text(encoding="utf-8").splitlines():
            if line.endswith(" " + ref):
                return line.split(" ", 1)[0]
    except OSError:
        pass
    return None


UPSTREAM_COMMIT = _upstream_commit()


def _read_live_table():
    try:
        blob = json.loads(LIVE_TABLE.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None, {}
    return blob.get("version"), blob.get("data") or {}


def _diff_entries(before, after):
    """Diff two table payloads.

    Scalar columns report old/new. Row lists report added/removed, matching the
    shape the upstream reload endpoint returns for the same tables.
    """
    changed = {}
    for key in set(before) | set(after):
        old, new = before.get(key), after.get(key)
        if old == new:
            continue
        if isinstance(old, list) or isinstance(new, list):
            old_rows, new_rows = old or [], new or []
            changed[key] = {
                "removed": [r for r in old_rows if r not in new_rows],
                "added": [r for r in new_rows if r not in old_rows],
            }
        else:
            changed[key] = {"old": old, "new": new}
    return changed


def _track(served, on_disk, entries):
    """Record a rotation once the reload has caught up with the file.

    While the file is ahead of the served copy the cached 'before' state is
    left alone, so the diff is taken against what was actually being served.
    """
    if on_disk != served:
        return
    previous = _LAST["version"]
    if previous is not None and served != previous:
        before = _LAST["data"] or {}
        changed = _diff_entries(before, entries)
        try:
            when = datetime.fromtimestamp(LIVE_TABLE.stat().st_mtime, timezone.utc)
        except OSError:
            when = datetime.now(timezone.utc)
        _ROTATIONS.insert(0, {"ts": _iso(when), "table": TABLE_NAME,
                              "from": previous, "to": served, "changed": changed})
        del _ROTATIONS[12:]
    _LAST["version"] = served
    _LAST["data"] = entries


_seed_version, _seed_entries = _read_live_table()
_track(registry.version(TABLE_NAME), _seed_version, _seed_entries)


@app.get("/live/status")
def live_status():
    """Public status: what is loaded, what drifts, and on what cadence."""
    served = registry.version(TABLE_NAME)
    on_disk, entries = _read_live_table()
    _track(served, on_disk, entries)
    return {
        "service": "self-reflective-apis live demo",
        "upstream": "github.com/arquicanedo/self-reflective-apis (default branch, pinned)",
        "upstream_commit": UPSTREAM_COMMIT,
        "started_at": _iso(_STARTED),
        "table_versions": registry.all_versions(),
        "drift": {
            "table": TABLE_NAME,
            "schedule": DRIFT_SCHEDULE,
            "ring": [p.name for p in sorted(SNAPSHOT_DIR.glob(FILE_STEM + "_v*.json"))],
            "served_version": served,
            "on_disk_version": on_disk,
            "reload_pending": on_disk is not None and on_disk != served,
            "entries": entries,
        },
        "rotations": _ROTATIONS,
        "visitor_requests": _VISITOR_CALLS["n"],
        "acme_promo_table_loaded": ACME_TABLE_LOADED,
        "admin": "routes under /admin require a bearer token at the gate (port 8180)",
    }


@app.get("/live", include_in_schema=False)
@app.get("/live/", include_in_schema=False)
def live_page():
    return HTMLResponse(_PAGE.read_text(encoding="utf-8"),
                        headers={"Cache-Control": "no-store"})