ai-agent/backend/deploy_status.py

"""
Live deploy-progress status for the ai-agent viewer.

The blue-green deploy script (``services/ai-agent/deploy.sh``) runs on the *host*
and, alongside its ``/tmp`` journal, writes a JSON snapshot of the in-flight
deploy into the service's own data dir (``/data/deploy-status.json`` in the
container — bind-mounted from ``services/ai-agent/data``). It rewrites that file
as each phase begins and stamps it ``done``/``failed`` when the deploy ends.

This module surfaces that file to the PWA:

* :func:`read_status` parses the snapshot into a small dict (or ``None`` when
  there is no active/recent deploy), applied to the API + SSE payloads.
* :class:`DeployWatcher` polls the file every second and publishes a ``deploy``
  SSE event whenever the snapshot changes, so the sticky in-app deploy banner
  updates live without a manual refresh.

The file is a plain JSON object the script writes; we tolerate a partial write
(mid-flush) by treating a parse error as "no change".

Snapshot shape (all optional except ``phaseNum``/``totalPhases``/``status``)::

    {
      "label":       "deploy-status-notif",   # worktree / branch label
      "status":      "running" | "done" | "failed",
      "phaseNum":    3,                        # 0-based-ish current phase (1..N)
      "totalPhases": 6,
      "phase":       "waiting for standby ...", # current phase description
      "steps": [ {"n": 1, "label": "..."}, ... ],  # all phase labels, in order
      "startedAt":   1731000000,               # epoch seconds
      "updatedAt":   1731000123,
      "sessionId":   "f7baa782-...",           # calling conversation
      "convUrl":     "https://ai-agent.lab.../conversation/<enc>/<sid>.jsonl",
      "convTitle":   "Deploy status notification"
    }

We derive each step's per-step state (``pending`` / ``active`` / ``done`` /
``failed``) from ``phaseNum`` + ``status`` here so the frontend just renders dots.
"""

from __future__ import annotations

import json
import os
import pathlib
import threading


DEPLOY_STATUS_PATH = pathlib.Path(
    os.environ.get("DEPLOY_STATUS_PATH", "/data/deploy-status.json")
)

# A finished/failed snapshot lingers so the banner can show the final state
# briefly; the script leaves the file in place and the frontend dismisses it.
# We still surface a terminal snapshot until the file is removed or replaced.


def _derive_steps(snap: dict) -> list[dict]:
    """Attach a per-step ``state`` to each declared step from phaseNum/status."""
    steps = snap.get("steps") or []
    phase_num = int(snap.get("phaseNum") or 0)
    status = snap.get("status") or "running"
    out: list[dict] = []
    for s in steps:
        n = int(s.get("n") or 0)
        if status == "done":
            state = "done"
        elif status == "failed":
            # every step before the current one succeeded; the current one failed.
            state = "done" if n < phase_num else ("failed" if n == phase_num else "pending")
        else:  # running
            if n < phase_num:
                state = "done"
            elif n == phase_num:
                state = "active"
            else:
                state = "pending"
        out.append({"n": n, "label": s.get("label", ""), "state": state})
    return out


def read_status() -> dict | None:
    """Parse the current deploy snapshot, or ``None`` if there is none.

    Never raises: a missing file, an unreadable file, or a half-written file
    (concurrent flush from the script) all read as "no active deploy" so a
    transient parse error can't take down ``/api/*``.
    """
    try:
        raw = DEPLOY_STATUS_PATH.read_text()
    except (OSError, ValueError):
        return None
    try:
        snap = json.loads(raw)
    except ValueError:
        return None
    if not isinstance(snap, dict):
        return None
    snap = dict(snap)
    snap["steps"] = _derive_steps(snap)
    # Normalize the fields the UI relies on so a partial file still renders.
    snap.setdefault("status", "running")
    snap.setdefault("totalPhases", len(snap["steps"]) or 0)
    snap.setdefault("phaseNum", 0)
    return snap


class DeployWatcher(threading.Thread):
    """Poll the deploy snapshot; publish a ``deploy`` SSE event on any change."""

    def __init__(self, hub, interval: float = 1.0):
        super().__init__(daemon=True)
        self.hub = hub
        self.interval = interval
        self._sig: tuple | None = None
        self._stop = threading.Event()

    def stop(self) -> None:
        self._stop.set()

    def _signature(self) -> tuple:
        """(mtime, size) of the snapshot file, or a sentinel when it's absent."""
        try:
            st = DEPLOY_STATUS_PATH.stat()
            return (st.st_mtime, st.st_size)
        except OSError:
            return ()

    def run(self) -> None:
        self._sig = self._signature()  # prime so we don't fire on boot
        while not self._stop.wait(self.interval):
            try:
                sig = self._signature()
                if sig != self._sig:
                    self._sig = sig
                    self.hub.publish({"type": "deploy"})
            except Exception:
                pass