alembic/versions/0090_service_seen.py

@@ -0,0 +1,64 @@
+"""service_seen — the learned roster that makes a stopped part observable.
+
+Milestone 365. Nothing in FabledCurator knew what was SUPPOSED to be running:
+`celery inspect` reports the workers that answer, so a dead worker was a
+shorter list rather than a red light, and the only surface that could tell an
+operator otherwise was Portainer. This table is the memory that turns an
+absence into something the app can see.
+
+Keyed on the queue set for a celery role and on agent_id for the GPU agent —
+NOT on the celery worker name, which here is `celery@<container id>` and is
+minted fresh on every deploy. See the model docstring for why that choice is
+the whole design.
+
+## First migration on the collapsed baseline
+
+0089 is the single generated baseline that replaced revisions 0001..0089
+(milestone 328). This is the first revision written on top of it, so it is
+also the first evidence that the chain steps forward from the collapse rather
+than merely reproducing the schema — which nothing had demonstrated yet.
+
+An existing install is at 0089 because it ran the real 0089; a fresh one is at
+0089 because it ran the baseline. Both arrive here identically, which was the
+property the collapse was designed around.
+
+Revision ID: 0090
+Revises: 0089
+Create Date: 2026-09-02
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0090"
+down_revision: Union[str, None] = "0089"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "service_seen",
+        sa.Column("key", sa.String(length=128), nullable=False),
+        sa.Column("kind", sa.String(length=16), nullable=False),
+        sa.Column("display_name", sa.String(length=64), nullable=False),
+        sa.Column(
+            "first_seen_at", sa.DateTime(timezone=True),
+            server_default=sa.text("now()"), nullable=False,
+        ),
+        sa.Column(
+            "last_seen_at", sa.DateTime(timezone=True),
+            server_default=sa.text("now()"), nullable=False,
+        ),
+        sa.Column("details", sa.JSON(), nullable=False),
+        sa.PrimaryKeyConstraint("key", name=op.f("pk_service_seen")),
+    )
+    # No secondary indexes, deliberately: one row per moving part means every
+    # read is a handful of rows and an index would be write cost buying
+    # nothing (#3301 removed seven of exactly that shape).
+
+
+def downgrade() -> None:
+    op.drop_table("service_seen")

backend/app/api/gpu.py

@@ -21,6 +21,7 @@ from ..services.gallery_service import image_url
 from ..services.ml.gpu_jobs import GpuJobService, error_dedupe_statements
 from ..services.ml.gpu_triage import classify_reason, recover_defective_image
 from ..services.ml.regions import RegionService
+from ..services.service_roster import touch_service
 
 gpu_bp = Blueprint("gpu", __name__, url_prefix="/api/gpu")
 
@@ -256,6 +257,18 @@ async def lease():
         if not await _agent_authed(session):
             return jsonify({"error": "unauthorized"}), 401
         jobs = await GpuJobService(session).lease(agent_id, batch_size=batch)
+        # The agent cannot be polled — it is HTTP-only and pulls from here, so
+        # web never dials it. A lease IS the check-in, and until milestone 365
+        # it was thrown away: an agent sitting idle with nothing to lease left
+        # no trace at all and was indistinguishable from one switched off a
+        # week ago. Recorded on the call that was already happening.
+        await touch_service(
+            session,
+            key=f"agent:{agent_id}",
+            kind="agent",
+            display_name="GPU agent" if agent_id == "agent" else f"GPU agent ({agent_id})",
+            details={"agent_id": agent_id, "last_call": "lease", "leased": len(jobs)},
+        )
         ml = await MLSettings.load(session)
         # image rows for url/mime in one shot
         ids = [j.image_record_id for j in jobs]
@@ -329,6 +342,13 @@ async def heartbeat():
         if not await _agent_authed(session):
             return jsonify({"error": "unauthorized"}), 401
         n = await GpuJobService(session).heartbeat(agent_id, job_ids)
+        await touch_service(
+            session,
+            key=f"agent:{agent_id}",
+            kind="agent",
+            display_name="GPU agent" if agent_id == "agent" else f"GPU agent ({agent_id})",
+            details={"agent_id": agent_id, "last_call": "heartbeat", "extended": n},
+        )
         await session.commit()
     return jsonify({"extended": n})
 

backend/app/models/__init__.py

@@ -32,6 +32,7 @@ from .presentation_review import PresentationReview
 from .series_chapter import SeriesChapter
 from .series_page import SeriesPage
 from .series_suggestion import SeriesSuggestion
+from .service_seen import ServiceSeen
 from .source import Source
 from .subscribestar_failed_media import SubscribeStarFailedMedia
 from .subscribestar_seen_media import SubscribeStarSeenMedia
@@ -63,6 +64,7 @@ __all__ = [
     "SeriesChapter",
     "SeriesPage",
     "SeriesSuggestion",
+    "ServiceSeen",
     "ImageRecord",
     "ImageProvenance",
     "ImageRegion",

backend/app/models/service_seen.py

@@ -0,0 +1,88 @@
+"""service_seen — the learned roster of FabledCurator's own moving parts.
+
+Nothing else in this application knows what is SUPPOSED to be running.
+`celery inspect` reports the workers that answer, so a stopped worker is a
+shorter list rather than a red light, and Postgres and Redis have no
+representation at all. That is why the only place an operator could see a
+dead service was Portainer, which knows the intended set (milestone 365).
+
+This table is the memory that makes an absence observable: every part that
+has ever checked in, and when it last did. A row that stops advancing is a
+part that stopped.
+
+## Why the key is not the hostname
+
+`_read_workers_sync()` returns celery's worker names, which here are
+`celery@<container id>`. Those are minted fresh on every deploy. Keyed on
+them, this table would record a death and a birth every time the stack is
+updated — and a status page that goes red on every deploy is a status page
+nobody reads, which is worse than not having one.
+
+So a celery role is keyed on its **queue set**, which is assigned per role in
+docker-compose.yml (`CELERY_QUEUES`) and survives container replacement:
+
+    default,import,thumbnail,download  ->  worker
+    maintenance,scan                   ->  scheduler   (celery worker --beat)
+    ml                                 ->  ml-worker
+
+Two replicas of one role share a queue set and are therefore ONE row — which
+is right, because the question being answered is "is that role being served",
+not "how many containers exist". The replica count and their hostnames go in
+`details`, where they can change without the identity changing.
+
+The GPU agent is keyed on its `agent_id`, the identity its lease protocol
+already uses (`api/gpu.py`).
+
+## What is NOT in here
+
+Postgres and Redis. They are always expected and never learned, and a
+last-seen for them would be actively misleading — that one answered thirty
+seconds ago says nothing about now. They are probed live at request time.
+
+## kind
+
+Plain `String`, not a Postgres ENUM and not CHECK-gated, matching
+`gpu_job.status` and `backup_run.status`. The value set here is expected to
+grow as parts are added, and a constraint swap per new kind (rule 36) would
+be cost with no invariant behind it.
+
+    celery  — a worker role, keyed on its queue set
+    agent   — a GPU agent, keyed on its agent_id
+"""
+
+from datetime import datetime
+
+from sqlalchemy import JSON, DateTime, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .base import Base
+
+
+class ServiceSeen(Base):
+    __tablename__ = "service_seen"
+
+    # No indexes beyond the primary key, deliberately. This table holds one row
+    # per moving part — a handful, forever — so every query against it is a
+    # full read of a few rows and an index would be write cost buying nothing
+    # (the lesson of #3301, which removed seven redundant ones).
+    key: Mapped[str] = mapped_column(String(128), primary_key=True)
+    kind: Mapped[str] = mapped_column(String(16), nullable=False)
+
+    # What to call it in the UI. Derived from the queue set where it is
+    # recognised, and falling back to the raw queue list where it is not — a
+    # deployment that slices its queues differently should still show something
+    # true rather than a name this code invented for it.
+    display_name: Mapped[str] = mapped_column(String(64), nullable=False)
+
+    first_seen_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(),
+    )
+    last_seen_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(),
+    )
+
+    # The parts that change without changing identity: replica hostnames,
+    # active task counts, the queues actually being served. Kept as a blob
+    # because it is displayed and never queried — giving it columns would
+    # invite filtering on it, which is what the activity endpoints are for.
+    details: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)

backend/app/services/service_roster.py

@@ -0,0 +1,175 @@
+"""The learned roster: which of FabledCurator's parts have checked in, and when.
+
+Milestone 365. `celery inspect` answers "who is here"; this answers "who is
+missing", which nothing in the application could do before — see
+`models/service_seen.py` for why the identity is a queue set and not a
+worker hostname.
+
+## Who does the observing, and why it is the web process
+
+Three candidates, and the choice matters more than the code:
+
+* **A celery beat sweep.** Rejected. If the scheduler dies, the sweep stops,
+  every row goes stale, and the page reports that everything is down when one
+  thing is. An alarm that cannot distinguish "one part died" from "the
+  observer died" is worse than no alarm.
+* **A background task in web.** Rejected on a detail of how this deploys:
+  hypercorn runs `--workers 4`, so a `before_serving` loop would be FOUR
+  concurrent inspect loops hammering the broker, forever, per container.
+* **Refresh on demand, rate-limited by the data itself.** Taken. Whichever web
+  process happens to serve a health request refreshes the roster if it is
+  older than REFRESH_TTL, and otherwise reads what is already there.
+
+The third has the property the other two lack: **the observer is the thing
+serving the page.** If web is down you get a browser error rather than a
+confidently green page, which is the honest failure. It also self-limits
+without coordination — the TTL lives in the row everybody can see.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from sqlalchemy import func, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..models import ServiceSeen
+
+log = logging.getLogger(__name__)
+
+# How stale the roster may be before a health request refreshes it. Comfortably
+# under the staleness thresholds that decide a service is missing, so the
+# verdict is never limited by how often anyone looked.
+REFRESH_TTL_SECONDS = 20.0
+
+# celery inspect is a broker round trip and this sits on a request path, so it
+# gets a deadline (rule 156). A broker that has stopped answering must make the
+# roster stale — which is a true statement about the system — not hang the one
+# page that exists to explain it.
+INSPECT_TIMEOUT_SECONDS = 2.0
+
+# Queue set -> the name an operator recognises. Sorted-tuple keys, because the
+# order celery reports them in is not guaranteed.
+#
+# A deployment that slices CELERY_QUEUES differently falls through to the raw
+# queue list rather than being given a name this table invented for it: a
+# wrong-but-confident label on a status page is worse than an ugly true one.
+ROLE_NAMES: dict[tuple[str, ...], str] = {
+    ("default", "download", "import", "thumbnail"): "Worker",
+    ("maintenance", "scan"): "Scheduler",
+    ("ml",): "ML worker",
+}
+
+
+def role_display_name(queues: tuple[str, ...]) -> str:
+    known = ROLE_NAMES.get(queues)
+    if known:
+        return known
+    return "Worker (" + ", ".join(queues) + ")"
+
+
+def _inspect_celery_sync() -> dict[tuple[str, ...], dict]:
+    """celery inspect, grouped by queue set rather than by worker.
+
+    Returns {queue_set: {"hostnames": [...], "active": int}}. Two replicas of
+    one role collapse into one entry on purpose — the question is whether the
+    role is being served, not how many containers exist.
+    """
+    from ..celery_app import celery as celery_app
+
+    insp = celery_app.control.inspect(timeout=INSPECT_TIMEOUT_SECONDS)
+    active_queues = insp.active_queues() or {}
+    active_tasks = insp.active() or {}
+
+    grouped: dict[tuple[str, ...], dict] = {}
+    for hostname, queues in active_queues.items():
+        key = tuple(sorted({q["name"] for q in queues}))
+        entry = grouped.setdefault(key, {"hostnames": [], "active": 0})
+        entry["hostnames"].append(hostname)
+        entry["active"] += len(active_tasks.get(hostname, []))
+    for entry in grouped.values():
+        entry["hostnames"].sort()
+    return grouped
+
+
+async def touch_service(
+    session: AsyncSession, *, key: str, kind: str, display_name: str, details: dict
+) -> None:
+    """Record that a part checked in just now.
+
+    Upsert rather than read-modify-write: several web processes and several
+    agents can be doing this at once, and the last writer is simply the most
+    recent sighting. `first_seen_at` is deliberately NOT updated — it is the
+    one field that answers "has this ever run", which the learned-roster design
+    depends on.
+    """
+    stmt = pg_insert(ServiceSeen).values(
+        key=key, kind=kind, display_name=display_name, details=details,
+    )
+    stmt = stmt.on_conflict_do_update(
+        index_elements=[ServiceSeen.key],
+        set_={
+            "kind": stmt.excluded.kind,
+            "display_name": stmt.excluded.display_name,
+            "details": stmt.excluded.details,
+            "last_seen_at": func.now(),
+        },
+    )
+    await session.execute(stmt)
+
+
+async def refresh_celery_roster(session: AsyncSession) -> None:
+    """Inspect the broker and record what answered. Never raises.
+
+    A failure here means the roster does not advance, and the rows going stale
+    is then a TRUE report about a broker nobody can reach. Letting the
+    exception out would instead break the health endpoint, which is the one
+    thing that must keep answering when the stack is unwell.
+    """
+    try:
+        grouped = await asyncio.wait_for(
+            asyncio.to_thread(_inspect_celery_sync),
+            timeout=INSPECT_TIMEOUT_SECONDS * 2,
+        )
+    except Exception:
+        log.warning("service roster: celery inspect failed; roster not refreshed", exc_info=True)
+        return
+
+    for queues, entry in grouped.items():
+        await touch_service(
+            session,
+            key="celery:" + ",".join(queues),
+            kind="celery",
+            display_name=role_display_name(queues),
+            details={
+                "queues": list(queues),
+                "hostnames": entry["hostnames"],
+                "replicas": len(entry["hostnames"]),
+                "active": entry["active"],
+            },
+        )
+
+
+async def refresh_if_stale(session: AsyncSession) -> None:
+    """Refresh the celery roster if nobody has for REFRESH_TTL_SECONDS.
+
+    Rate-limited by the data rather than by a lock: the gate is the newest
+    last_seen_at across the celery rows, which every web process can see. Two
+    processes racing through the gate costs one redundant inspect and writes
+    the same values twice, so the benign outcome needs no coordination to
+    prevent.
+    """
+    newest = (
+        await session.execute(
+            select(func.max(ServiceSeen.last_seen_at)).where(ServiceSeen.kind == "celery")
+        )
+    ).scalar_one_or_none()
+
+    if newest is not None:
+        age = (await session.execute(select(func.now()))).scalar_one() - newest
+        if age.total_seconds() < REFRESH_TTL_SECONDS:
+            return
+
+    await refresh_celery_roster(session)