fix(system): report the stack whose manifest matches the running images

docs/data-plane/source-flow.md

@@ -129,7 +129,7 @@ Every app rolls on every config change, so nothing here depends on which app can
 
 ## Asking what the deployment is, and what a source is doing
 
-Two reads carry the facts a console would otherwise have to guess at. `GET /api/v1/system/deployment` answers what this deployment IS: the profile the deployer injected as `DFE_ENGINE_PROFILE`, the one transport it carries a source on - which is both the only one a source may name and the one a source that names none takes - whether the stage pools sit behind mesh listeners and where, whether compiled routing reaches its running apps (`applies_routing`), and the stack, engine, dfe-ui and per-app pinned versions (`apps`, name to version tag, from the deploy repo's pins). `GET /api/v1/system/version` serves the version half from the same function, so the footer and the deployment card cannot disagree, and `GET /api/v1/apps` judges each app's `offered` against the deployment's own profile when the caller names none.
+Two reads carry the facts a console would otherwise have to guess at. `GET /api/v1/system/deployment` answers what this deployment IS: the profile the deployer injected as `DFE_ENGINE_PROFILE`, the one transport it carries a source on - which is both the only one a source may name and the one a source that names none takes - whether the stage pools sit behind mesh listeners and where, whether compiled routing reaches its running apps (`applies_routing`), and the stack, engine, dfe-ui and per-app pinned versions (`apps`, name to version tag, from the deploy repo's pins). The stack comes from the deploy repo's pins, else from the published stack whose image digests match the `DFE_ENGINE_STACK_IMAGES` this deployment runs, else from the `DFE_ENGINE_STACK_VERSION` label, and `source` names which (`deploy-repo`, `registry`, `deployment`, `engine`). `DFE_ENGINE_STACK_RESOLVE=false` skips the registry read for an air-gapped site, and `DFE_ENGINE_STACK_MANIFEST_REPOSITORY` points it at a mirror. `GET /api/v1/system/version` serves the version half from the same function, so the footer and the deployment card cannot disagree, and `GET /api/v1/apps` judges each app's `offered` against the deployment's own profile when the caller names none.
 
 `GET /api/v1/sources/{name}/signals` answers what one source is doing: `records_per_min` and `last_seen` over a bounded five-minute window, read from the loader's `loader_messages_by_table_total` counter in the otel database, plus `landed_in_default`. The counter is per TABLE, so `landed_in_default` says the number covers the whole shared landing table rather than this source alone - which is the case for a fetcher pushing to the main topic. Either reading is null where no series answers, and the route still returns 200: a source only just defined has nothing to report, and a zero would claim a measurement nobody took.
 

openapi/openapi.json

@@ -39431,11 +39431,12 @@
             "type": "string",
             "enum": [
               "deploy-repo",
+              "registry",
               "deployment",
               "engine"
             ],
             "title": "Source",
-            "description": "deploy-repo when the stack version came from pins.yaml, deployment when it came from what deployed this pod, else engine."
+            "description": "deploy-repo when the stack version came from pins.yaml, registry when the published stack manifests matched the images this deployment runs, deployment when it came from what deployed this pod, else engine."
           },
           "backing": {
             "additionalProperties": {
@@ -56247,11 +56248,12 @@
             "type": "string",
             "enum": [
               "deploy-repo",
+              "registry",
               "deployment",
               "engine"
             ],
             "title": "Source",
-            "description": "deploy-repo when the stack version came from pins.yaml, deployment when it came from what deployed this pod, else engine."
+            "description": "deploy-repo when the stack version came from pins.yaml, registry when the published stack manifests matched the images this deployment runs, deployment when it came from what deployed this pod, else engine."
           },
           "python_version": {
             "type": "string",

src/dfe_engine/api/app.py

@@ -656,11 +656,33 @@ def reconcile_service_roles_now() -> ReconcileResult:
             name="gitops-consolidation",
         )
 
+    # In the background: the registry is the internet's, so startup and readiness never wait on it.
+    from dfe_engine.gitops.pins import load_pins, stack_version
+    from dfe_engine.stack_resolve import resolve_into
+
+    app.state.resolved_stack = None
+    read_pin = None
+    if gitcrud is not None:
+        deploy_repo = gitcrud
+
+        def read_pin() -> str | None:
+            # The pins are read as deployment_facts reads them, and they win over the registry.
+            with deploy_repo.reading():
+                return stack_version(load_pins(deploy_repo.repo_path))
+
+    app.state.stack_resolution = asyncio.create_task(
+        resolve_into(app.state, settings, read_pin), name="stack-resolution"
+    )
+
     health.set_started()
     health.set_ready()
     logger.info(f"DFE Engine API started (port={settings.api.port})")
     yield
     health.set_ready(False)
+    stack_resolution = getattr(app.state, "stack_resolution", None)
+    if stack_resolution is not None:
+        stack_resolution.cancel()
+        await asyncio.gather(stack_resolution, return_exceptions=True)
     oidc_group_sync = getattr(app.state, "oidc_group_sync", None)
     if oidc_group_sync is not None:
         # A sync writes group files and asks for a reconcile, so it stops before both close.

src/dfe_engine/api/v1/system.py

@@ -145,10 +145,11 @@ class VersionResponse(BaseModel):
             "dfe-ui, which then matches the ui field above."
         )
     )
-    source: Literal["deploy-repo", "deployment", "engine"] = Field(
+    source: Literal["deploy-repo", "registry", "deployment", "engine"] = Field(
         description=(
-            "deploy-repo when the stack version came from pins.yaml, deployment when it "
-            "came from what deployed this pod, else engine."
+            "deploy-repo when the stack version came from pins.yaml, registry when the "
+            "published stack manifests matched the images this deployment runs, deployment "
+            "when it came from what deployed this pod, else engine."
         )
     )
     python_version: str = Field(description="Python interpreter version")
@@ -285,10 +286,11 @@ class DeploymentResponse(BaseModel):
             "dfe-ui, which then matches the ui field above."
         )
     )
-    source: Literal["deploy-repo", "deployment", "engine"] = Field(
+    source: Literal["deploy-repo", "registry", "deployment", "engine"] = Field(
         description=(
-            "deploy-repo when the stack version came from pins.yaml, deployment when it "
-            "came from what deployed this pod, else engine."
+            "deploy-repo when the stack version came from pins.yaml, registry when the "
+            "published stack manifests matched the images this deployment runs, deployment "
+            "when it came from what deployed this pod, else engine."
         )
     )
     backing: dict[str, BackingFacts] = Field(
@@ -357,17 +359,30 @@ class SettingsSummary(BaseModel):
 # -- Endpoints ------------------------------------------------
 
 
+def _stack_source(
+    pinned: str | None, resolved: str | None, stack: str | None
+) -> Literal["deploy-repo", "registry", "deployment", "engine"]:
+    """Name where ``stack`` came from, in the order deployment_facts prefers them."""
+    if pinned:
+        return "deploy-repo"
+    if resolved:
+        return "registry"
+    return "deployment" if stack else "engine"
+
+
 def deployment_facts(request: Request, settings: Any) -> DeploymentResponse:
     """Everything this deployment can state about itself, read once.
 
     Both /system/deployment and /system/version answer from here, so the versions
     the console footer shows and the versions its deployment card shows cannot
     disagree.
 
-    The pins win where there are any: they are what the operator chose. A deploy
-    with no pins base still knows what stood it up, because the chart passes that
-    in, so the footer shows a stack version rather than nothing. With neither, the
-    engine's own version is the whole answer.
+    The pins win where there are any: they are what the operator chose. Next comes
+    the published stack whose image digests match what this deployment runs, which
+    the engine resolves in the background after startup. A deploy with neither still
+    knows what stood it up, because the chart or compose passes that in, so the footer
+    shows a stack version rather than nothing. With none of them, the engine's own
+    version is the whole answer.
 
     Blocking: it reads the deploy repo, so a route calls it off the event loop.
     """
@@ -380,7 +395,8 @@ def deployment_facts(request: Request, settings: Any) -> DeploymentResponse:
         with gc.reading():
             pins = load_pins(gc.repo_path)
     pinned = stack_version(pins)
-    stack = pinned or settings.stack_version or None
+    resolved = getattr(request.app.state, "resolved_stack", None)
+    stack = pinned or resolved or settings.stack_version or None
     apps = component_overrides(pins)
     transport = settings.transport
     return DeploymentResponse(
@@ -402,7 +418,7 @@ def deployment_facts(request: Request, settings: Any) -> DeploymentResponse:
         engine=__version__,
         ui=apps.get(UI_COMPONENT) or settings.ui_version or None,
         apps=apps,
-        source="deploy-repo" if pinned else ("deployment" if stack else "engine"),
+        source=_stack_source(pinned, resolved, stack),
         backing=backing_facts(request),
     )
 

src/dfe_engine/settings.py

@@ -2159,6 +2159,49 @@ class DFESettings(BaseModel):
             "repo pins no stack. DFE_ENGINE_STACK_VERSION."
         ),
     )
+    stack_images: list[str] = Field(
+        default_factory=list,
+        description=(
+            "The DFE images this deployment runs, each as ref@sha256:digest, comma-separated "
+            "in DFE_ENGINE_STACK_IMAGES. When set, the engine reads the published stack "
+            "manifests at startup and reports the newest stack whose digests agree with "
+            "every image listed, in place of the stack_version label. List DFE apps only: "
+            "an image no stack manifest names stops any stack matching."
+        ),
+    )
+    stack_manifest_repository: str = Field(
+        default="ghcr.io/hyperi-io/dfe-stack-manifest",
+        description=(
+            "host/path of the registry repository the stack manifests are published to, "
+            "read anonymously. Point it at a mirror for another organisation. "
+            "DFE_ENGINE_STACK_MANIFEST_REPOSITORY."
+        ),
+    )
+    stack_resolve: bool = Field(
+        default=True,
+        description=(
+            "Resolve the running stack from stack_images against the registry. Set false "
+            "on an air-gapped site, which then reports the stack_version label. "
+            "DFE_ENGINE_STACK_RESOLVE."
+        ),
+    )
+    stack_resolve_request_timeout_seconds: float = Field(
+        default=5.0,
+        gt=0,
+        description=(
+            "Longest one registry request in the stack resolution may take. "
+            "DFE_ENGINE_STACK_RESOLVE_REQUEST_TIMEOUT_SECONDS."
+        ),
+    )
+    stack_resolve_budget_seconds: float = Field(
+        default=30.0,
+        gt=0,
+        description=(
+            "Longest the whole stack resolution may take, after which the stack_version "
+            "label answers. It runs in the background and never holds startup or readiness. "
+            "DFE_ENGINE_STACK_RESOLVE_BUDGET_SECONDS."
+        ),
+    )
     ui_version: str = Field(
         default="",
         description=(
@@ -2377,6 +2420,7 @@ def _strip_lower(value: str) -> str:
             "DFE_ENGINE_SECRETS_RECOVERY_WINDOW_DAYS": "secrets.recovery_window_days",
             "DFE_ENGINE_SECRETS_TAGS": "secrets.tags",
             "DFE_ENGINE_STACK_VERSION": "stack_version",
+            "DFE_ENGINE_STACK_MANIFEST_REPOSITORY": "stack_manifest_repository",
             "DFE_ENGINE_UI_VERSION": "ui_version",
             "DFE_ENGINE_ENV": "env",
         },
@@ -2446,6 +2490,8 @@ def _strip_lower(value: str) -> str:
         {
             "DFE_ENGINE_CLICKHOUSE_BOOTSTRAP_WAIT_SECONDS": "clickhouse.bootstrap_wait_seconds",
             "DFE_ENGINE_DETECTION_GUARD_PREVIEW_TIMEOUT_SECONDS": "hunts.detection_guard.preview_timeout_seconds",
+            "DFE_ENGINE_STACK_RESOLVE_REQUEST_TIMEOUT_SECONDS": "stack_resolve_request_timeout_seconds",
+            "DFE_ENGINE_STACK_RESOLVE_BUDGET_SECONDS": "stack_resolve_budget_seconds",
             "DFE_ENGINE_HUNTS_PHASE_SPREAD_FRACTION": "hunts.phase_spread_fraction",
             "DFE_ENGINE_HUNTS_FAILURE_BACKOFF_INITIAL_SECONDS": "hunts.failure_backoff_initial_seconds",
             "DFE_ENGINE_HUNTS_FAILURE_BACKOFF_MAX_SECONDS": "hunts.failure_backoff_max_seconds",
@@ -2497,6 +2543,7 @@ def _strip_lower(value: str) -> str:
             "DFE_ENGINE_GITOPS_PUSH": "gitops.push",
             "DFE_ENGINE_API_DOCS_ENABLED": "api.docs_enabled",
             "DFE_ENGINE_E2E_SERVER": "e2e_server",
+            "DFE_ENGINE_STACK_RESOLVE": "stack_resolve",
         },
     ),
     (
@@ -2541,6 +2588,7 @@ def _strip_lower(value: str) -> str:
             "DFE_ENGINE_HUNTS_ALERT_CHANNELS": "hunts.alert_channels",
             "DFE_ENGINE_HUNTS_LOOKUP_DATABASES": "hunts.lookup_databases",
             "DFE_ENGINE_API_CORS_ORIGINS": "api.cors_origins",
+            "DFE_ENGINE_STACK_IMAGES": "stack_images",
         },
     ),
 )
@@ -2607,7 +2655,12 @@ def _get_env_overrides() -> dict:
             if key not in overrides.get(section, {}):
                 _set_path(overrides, path, str(Path(config_dir) / subdir))
 
-    return {section: value for section, value in overrides.items() if value}
+    # Only an empty section goes: a top-level false or zero is the operator's answer.
+    return {
+        section: value
+        for section, value in overrides.items()
+        if value or not isinstance(value, dict)
+    }
 
 
 def _deep_merge(base: dict, override: dict) -> dict: