diff --git a/backend/app/frontend/miner_detail_view.py b/backend/app/frontend/miner_detail_view.py
new file mode 100644
index 0000000..00ae1f9
--- /dev/null
+++ b/backend/app/frontend/miner_detail_view.py
@@ -0,0 +1,76 @@
+"""Semantic HTML for the Internal Alpha Miner Detail page."""
+
+
+def miner_detail_content() -> str:
+ return """
+
+ Loading miner
+ Requesting current identity and operational context.
+
+
+ Miner detail unavailable
+ The miner could not be loaded.
+
+
+
+
+
+
+
+ Latest facts
+
Current state
+
No observation
+
+
+
+ Physical context
+
Location
+
+ Work context
+
Repair context
+
+
+
+
+ Identity history
+ Open to load Asset ID and network identity history.
+
+
+ Recent observations
+ Open to load up to 100 recent factual observations.
+
+
+
diff --git a/backend/app/inventory/repository.py b/backend/app/inventory/repository.py
index 991883d..1005cec 100644
--- a/backend/app/inventory/repository.py
+++ b/backend/app/inventory/repository.py
@@ -59,6 +59,66 @@ MINER_SELECT = f"""
) tag_data ON true
"""
+MINER_CURRENT_SELECT = f"""
+ SELECT
+ miners.id,
+ miners.site_id,
+ miners.room_id,
+ miners.asset_tag,
+ miners.current_mac,
+ host(miners.current_ip) AS current_ip,
+ miners.hostname,
+ miners.miner_sn,
+ miners.control_board_sn,
+ miners.power_sn,
+ miners.model,
+ miners.firmware_version,
+ miners.lifecycle_status,
+ miners.status,
+ miners.notes,
+ miners.first_seen_at,
+ miners.last_seen_at,
+ miners.created_at,
+ miners.updated_at,
+ tag_data.tags,
+ latest.observed_at,
+ latest.hashrate_ths,
+ latest.power_w,
+ latest.chip_temp_max_c,
+ latest.hash_stable
+ FROM miners
+ LEFT JOIN LATERAL (
+ SELECT COALESCE(
+ jsonb_agg(
+ jsonb_build_object(
+ 'id', tags.id,
+ 'name', tags.name,
+ 'created_at', tags.created_at,
+ 'updated_at', tags.updated_at
+ )
+ ORDER BY lower({TAG_TRIMMED_NAME_SQL}), tags.id
+ ),
+ '[]'::jsonb
+ ) AS tags
+ FROM miner_tags
+ JOIN tags ON tags.id = miner_tags.tag_id
+ WHERE miner_tags.miner_id = miners.id
+ ) tag_data ON true
+ LEFT JOIN LATERAL (
+ SELECT snapshots.observed_at,
+ ROUND((snapshots.mhs_av / 1000000.0)::numeric, 2)::float
+ AS hashrate_ths,
+ ROUND(snapshots.power_w::numeric, 2)::float AS power_w,
+ ROUND(snapshots.chip_temp_max_c::numeric, 2)::float
+ AS chip_temp_max_c,
+ snapshots.hash_stable
+ FROM miner_status_snapshots AS snapshots
+ WHERE snapshots.miner_id = miners.id
+ ORDER BY snapshots.observed_at DESC, snapshots.id DESC
+ LIMIT 1
+ ) latest ON true
+"""
+
ACTIVE_REPAIR_STATUSES = (
"OPEN",
"IN_PROGRESS",
@@ -260,6 +320,17 @@ class InventoryRepository:
with connection.cursor(row_factory=dict_row) as cur:
return self._fetch_miner(cur, "miners.id = %s", (miner_id,))
+ def get_current_miner_by_id(self, miner_id: UUID) -> dict | None:
+ """Return Inventory identity with its latest factual telemetry."""
+ with self._connection_scope() as connection:
+ with connection.cursor(row_factory=dict_row) as cur:
+ cur.execute(
+ f"{MINER_CURRENT_SELECT} WHERE miners.id = %s",
+ (miner_id,),
+ )
+ row = cur.fetchone()
+ return dict(row) if row else None
+
def get_miner_by_id_for_update(self, miner_id: UUID) -> dict | None:
with self._connection_scope() as connection:
with connection.cursor(row_factory=dict_row) as cur:
diff --git a/backend/app/inventory/service.py b/backend/app/inventory/service.py
index c70175e..db2d718 100644
--- a/backend/app/inventory/service.py
+++ b/backend/app/inventory/service.py
@@ -137,6 +137,13 @@ class InventoryService:
raise MinerNotFound(miner_id)
return self._detail_model(row)
+ def get_current_miner(self, miner_id: UUID) -> InventoryMinerDetail:
+ """Return current Inventory identity and the latest observation."""
+ row = self.repository.get_current_miner_by_id(miner_id)
+ if not row:
+ raise MinerNotFound(miner_id)
+ return self._detail_model(row)
+
def get_asset_id_history(
self,
miner_id: UUID,
diff --git a/backend/app/locations/models.py b/backend/app/locations/models.py
index b078f89..5986509 100644
--- a/backend/app/locations/models.py
+++ b/backend/app/locations/models.py
@@ -118,6 +118,18 @@ class MineMapResponse(BaseModel):
error: None = None
+class MinerLocationDetail(BaseModel):
+ site_id: UUID | None = None
+ site_name: str | None = None
+ room_id: UUID | None = None
+ room_name: str | None = None
+ position_id: UUID
+ rack_label: str | None = None
+ shelf_label: str | None = None
+ position_label: str | None = None
+ position_index: int | None = None
+
+
class LocationSummary(BaseModel):
sites: int
rooms: int
diff --git a/backend/app/locations/repository.py b/backend/app/locations/repository.py
index 20d17b4..10d0550 100644
--- a/backend/app/locations/repository.py
+++ b/backend/app/locations/repository.py
@@ -229,3 +229,29 @@ class LocationMapRepository:
"""
)
return dict(cur.fetchone())
+
+ def get_miner_location(self, miner_id: UUID) -> dict[str, Any] | None:
+ """Return the one current physical position assigned to a miner."""
+ with self._connection_scope() as connection:
+ with connection.cursor(row_factory=dict_row) as cur:
+ cur.execute(
+ """
+ SELECT sites.id AS site_id, sites.name AS site_name,
+ rooms.id AS room_id,
+ COALESCE(rooms.name, position.room_name) AS room_name,
+ position.id AS position_id,
+ position.rack_name AS rack_label,
+ position.shelf_name AS shelf_label,
+ position.position_label,
+ position.position_index
+ FROM location_positions AS position
+ LEFT JOIN rooms ON rooms.id = position.room_id
+ LEFT JOIN sites ON sites.id = rooms.site_id
+ WHERE position.miner_id = %s
+ ORDER BY position.id
+ LIMIT 1
+ """,
+ (miner_id,),
+ )
+ row = cur.fetchone()
+ return dict(row) if row else None
diff --git a/backend/app/locations/service.py b/backend/app/locations/service.py
index 9858242..3e002d1 100644
--- a/backend/app/locations/service.py
+++ b/backend/app/locations/service.py
@@ -14,6 +14,7 @@ from app.locations.models import (
MineMapData,
MineMapMeta,
MineMapResponse,
+ MinerLocationDetail,
)
from app.locations.repository import LocationMapRepository
@@ -87,6 +88,12 @@ class LocationMapService:
),
)
+ def get_miner_location(self, miner_id: UUID) -> MinerLocationDetail | None:
+ row = self.repository.get_miner_location(miner_id)
+ if row is None:
+ return None
+ return MinerLocationDetail.model_validate(row)
+
def summarize_locations(self, *, include_miner_counts: bool) -> LocationSummary:
return LocationSummary.model_validate(
self.repository.summarize_locations(
diff --git a/backend/app/main.py b/backend/app/main.py
index b6bc3bd..e8d2b13 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -23,11 +23,13 @@ from app.identity import (
)
from app.frontend.router import (
dashboard_page as internal_alpha_dashboard_page,
+ miner_detail_page as internal_alpha_miner_detail_page,
router as frontend_router,
)
from app.inventory.router import router as inventory_router
from app.locations.router import router as locations_router
from app.components.router import router as components_router
+from app.miner_detail.router import router as miner_detail_router
from app.repairs.exceptions import (
ControlBoardReplacementActorRequired,
InvalidMacAddress,
@@ -60,6 +62,7 @@ app.include_router(inventory_router)
app.include_router(components_router)
app.include_router(locations_router)
app.include_router(dashboard_router)
+app.include_router(miner_detail_router)
install_identity_middleware(app)
@@ -1814,7 +1817,7 @@ def get_miner(miner_id: str, current_user: dict = Depends(get_current_user)):
@app.get("/api/miners/{miner_id}/network-identities")
-def list_miner_network_identities(miner_id: str, current_user: dict = Depends(get_current_user)):
+def list_miner_network_identities(miner_id: str, current_user: dict = Depends(require_permission("miners.view"))):
conn = None
try:
conn = get_db_connection()
@@ -29087,6 +29090,9 @@ async def minerops_inject_detail_extra_v2(request, call_next):
response = await call_next(request)
try:
+ if request.scope.get("endpoint") is internal_alpha_miner_detail_page:
+ return response
+
path_value = request.url.path or ""
if not (path_value.startswith("/miners/") and path_value.endswith("/detail")):
@@ -29683,6 +29689,9 @@ async def minerops_inject_clean_detail_v3(request, call_next):
response = await call_next(request)
try:
+ if request.scope.get("endpoint") is internal_alpha_miner_detail_page:
+ return response
+
path_value = request.url.path or ""
if not (path_value.startswith("/miners/") and path_value.endswith("/detail")):
@@ -30921,6 +30930,9 @@ async def minerops_inject_location_management(request, call_next):
response = await call_next(request)
try:
+ if request.scope.get("endpoint") is internal_alpha_miner_detail_page:
+ return response
+
path = request.url.path or ""
if not (path.startswith("/miners/") and path.endswith("/detail")):
diff --git a/backend/app/miner_detail/__init__.py b/backend/app/miner_detail/__init__.py
new file mode 100644
index 0000000..42902e3
--- /dev/null
+++ b/backend/app/miner_detail/__init__.py
@@ -0,0 +1 @@
+"""Permission-aware Miner Detail read model."""
diff --git a/backend/app/miner_detail/dependencies.py b/backend/app/miner_detail/dependencies.py
new file mode 100644
index 0000000..bfcf3d3
--- /dev/null
+++ b/backend/app/miner_detail/dependencies.py
@@ -0,0 +1,14 @@
+"""Dependency construction for the Miner Detail composition service."""
+
+from app.inventory.dependencies import get_inventory_service
+from app.locations.dependencies import get_location_map_service
+from app.miner_detail.service import MinerDetailService
+from app.repairs.dependencies import get_repair_center_service
+
+
+def get_miner_detail_service() -> MinerDetailService:
+ return MinerDetailService(
+ inventory_service=get_inventory_service(),
+ location_service=get_location_map_service(),
+ repair_service=get_repair_center_service(),
+ )
diff --git a/backend/app/miner_detail/models.py b/backend/app/miner_detail/models.py
new file mode 100644
index 0000000..6fd8662
--- /dev/null
+++ b/backend/app/miner_detail/models.py
@@ -0,0 +1,76 @@
+"""Typed contracts for the Internal Alpha Miner Detail overview."""
+
+from datetime import datetime
+from typing import Literal
+from uuid import UUID
+
+from pydantic import BaseModel, Field
+
+from app.inventory.models import InventoryMinerDetail
+from app.locations.models import MinerLocationDetail
+
+
+OperationalState = Literal[
+ "online", "offline", "unknown", "unstable", "repair", "spare", "retired"
+]
+
+
+class MinerDetailAccess(BaseModel):
+ can_view_locations: bool = False
+ can_read_repairs: bool = False
+ can_view_repairs: bool = False
+ can_read_components: bool = False
+
+
+class MinerDetailLocationSection(BaseModel):
+ available: bool
+ restricted: bool
+ location: MinerLocationDetail | None = None
+
+
+class MinerDetailRepairEpisode(BaseModel):
+ repair_id: UUID | None = None
+ source: Literal["modular", "legacy"]
+ active: bool
+ status: str
+ reason: str | None = None
+ opened_at: datetime | None = None
+ closed_at: datetime | None = None
+ latest_context: str | None = None
+ latest_context_at: datetime | None = None
+ parts_count: int | None = None
+ parts_summary: list[str] = Field(default_factory=list)
+
+
+class MinerDetailRepairSection(BaseModel):
+ available: bool
+ restricted: bool
+ active_repair: MinerDetailRepairEpisode | None = None
+ episodes: list[MinerDetailRepairEpisode] = Field(default_factory=list)
+
+
+class MinerDetailCapabilities(BaseModel):
+ can_view_map: bool = False
+ can_open_repair_center: bool = False
+ can_read_repair_history: bool = False
+ can_read_components: bool = False
+
+
+class MinerDetailOverviewData(BaseModel):
+ miner: InventoryMinerDetail
+ operational_state: OperationalState
+ location: MinerDetailLocationSection
+ repair: MinerDetailRepairSection
+ capabilities: MinerDetailCapabilities
+
+
+class MinerDetailOverviewMeta(BaseModel):
+ generated_at: datetime
+ database_query_count: int = Field(ge=1, le=3)
+
+
+class MinerDetailOverviewResponse(BaseModel):
+ data: MinerDetailOverviewData
+ meta: MinerDetailOverviewMeta
+ error: None = None
+
diff --git a/backend/app/miner_detail/router.py b/backend/app/miner_detail/router.py
new file mode 100644
index 0000000..729028d
--- /dev/null
+++ b/backend/app/miner_detail/router.py
@@ -0,0 +1,59 @@
+"""HTTP boundary for the Internal Alpha Miner Detail overview."""
+
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Response, status
+
+from app.identity import require_permission, user_has_permission
+from app.inventory.exceptions import MinerNotFound
+from app.miner_detail.dependencies import get_miner_detail_service
+from app.miner_detail.models import (
+ MinerDetailAccess,
+ MinerDetailOverviewResponse,
+)
+from app.miner_detail.service import MinerDetailService
+
+
+router = APIRouter(prefix="/api/miners", tags=["miners"])
+MinerDetailUser = Annotated[dict, Depends(require_permission("miners.view"))]
+MinerDetailServiceDependency = Annotated[
+ MinerDetailService,
+ Depends(get_miner_detail_service),
+]
+
+
+@router.get(
+ "/{miner_id}/overview",
+ response_model=MinerDetailOverviewResponse,
+)
+def get_miner_detail_overview(
+ miner_id: UUID,
+ current_user: MinerDetailUser,
+ service: MinerDetailServiceDependency,
+ response: Response,
+) -> MinerDetailOverviewResponse:
+ response.headers["Cache-Control"] = "no-store"
+ try:
+ return service.get_overview(
+ miner_id,
+ MinerDetailAccess(
+ can_view_locations=user_has_permission(
+ current_user, "locations.view"
+ ),
+ can_read_repairs=user_has_permission(
+ current_user, "repairs.read"
+ ),
+ can_view_repairs=user_has_permission(
+ current_user, "repairs.view"
+ ),
+ can_read_components=user_has_permission(
+ current_user, "components.read"
+ ),
+ ),
+ )
+ except MinerNotFound as exc:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail="Miner not found",
+ ) from exc
diff --git a/backend/app/miner_detail/service.py b/backend/app/miner_detail/service.py
new file mode 100644
index 0000000..f2e098a
--- /dev/null
+++ b/backend/app/miner_detail/service.py
@@ -0,0 +1,152 @@
+"""Composition service for the Internal Alpha Miner Detail overview."""
+
+from datetime import datetime, timezone
+from math import isfinite
+from uuid import UUID
+
+from app.inventory.models import InventoryMinerDetail
+from app.inventory.service import InventoryService
+from app.locations.service import LocationMapService
+from app.miner_detail.models import (
+ MinerDetailAccess,
+ MinerDetailCapabilities,
+ MinerDetailLocationSection,
+ MinerDetailOverviewData,
+ MinerDetailOverviewMeta,
+ MinerDetailOverviewResponse,
+ MinerDetailRepairEpisode,
+ MinerDetailRepairSection,
+ OperationalState,
+)
+from app.repairs.center_service import RepairCenterService
+from app.repairs.models import RepairCenterAccess, RepairCenterItem
+
+
+class MinerDetailService:
+ """Coordinate domain-owned current projections without transport logic."""
+
+ def __init__(
+ self,
+ inventory_service: InventoryService,
+ location_service: LocationMapService,
+ repair_service: RepairCenterService,
+ ):
+ self.inventory_service = inventory_service
+ self.location_service = location_service
+ self.repair_service = repair_service
+
+ @staticmethod
+ def _episode(item: RepairCenterItem) -> MinerDetailRepairEpisode:
+ return MinerDetailRepairEpisode(
+ repair_id=item.repair_id,
+ source=item.source,
+ active=item.active,
+ status=item.repair_status,
+ reason=item.reason,
+ opened_at=item.opened_at,
+ closed_at=item.closed_at,
+ latest_context=item.latest_context,
+ latest_context_at=item.latest_context_at,
+ parts_count=item.parts_count,
+ parts_summary=item.parts_summary,
+ )
+
+ @staticmethod
+ def _operational_state(
+ miner: InventoryMinerDetail,
+ active_repair: MinerDetailRepairEpisode | None,
+ ) -> OperationalState:
+ if active_repair is not None or miner.lifecycle_status == "in_repair":
+ return "repair"
+ if miner.lifecycle_status == "retired":
+ return "retired"
+ if miner.lifecycle_status == "spare":
+ return "spare"
+ runtime = (miner.runtime_status or "").strip().lower()
+ if runtime == "offline":
+ return "offline"
+ if miner.hash_stable is False:
+ return "unstable"
+ if runtime == "online":
+ return "online"
+ return "unknown"
+
+ def get_overview(
+ self,
+ miner_id: UUID,
+ access: MinerDetailAccess,
+ ) -> MinerDetailOverviewResponse:
+ miner = self.inventory_service.get_current_miner(miner_id)
+ # Location is owned by the permission-aware section below. Keep the
+ # shared projection from leaking location identifiers. Non-finite
+ # telemetry is not a factual display value and remains unavailable.
+ redacted = {
+ field: None
+ for field in (
+ "site_id", "site_name", "room_id", "room_name",
+ "location_position_id", "rack_label", "shelf_label",
+ "position_label", "position_index",
+ )
+ }
+ for field in ("hashrate_ths", "power_w", "chip_temp_max_c"):
+ value = getattr(miner, field)
+ if value is not None and not isfinite(value):
+ redacted[field] = None
+ miner = miner.model_copy(
+ update=redacted
+ )
+ location = (
+ self.location_service.get_miner_location(miner_id)
+ if access.can_view_locations
+ else None
+ )
+ repair_items = (
+ self.repair_service.list_miner_repairs(
+ miner_id,
+ RepairCenterAccess(
+ can_read_repairs=True,
+ can_view_miners=True,
+ ),
+ )
+ if access.can_read_repairs
+ else []
+ )
+ episodes = [self._episode(item) for item in repair_items]
+ active_repair = next(
+ (
+ episode for episode in episodes
+ if episode.active and episode.source == "modular"
+ ),
+ None,
+ )
+ query_count = 1 + int(access.can_view_locations) + int(access.can_read_repairs)
+ return MinerDetailOverviewResponse(
+ data=MinerDetailOverviewData(
+ miner=miner,
+ operational_state=self._operational_state(miner, active_repair),
+ location=MinerDetailLocationSection(
+ available=access.can_view_locations,
+ restricted=not access.can_view_locations,
+ location=location,
+ ),
+ repair=MinerDetailRepairSection(
+ available=access.can_read_repairs,
+ restricted=not access.can_read_repairs,
+ active_repair=active_repair,
+ episodes=episodes,
+ ),
+ capabilities=MinerDetailCapabilities(
+ can_view_map=bool(
+ access.can_view_locations and location is not None
+ and location.site_id and location.room_id
+ ),
+ can_open_repair_center=access.can_view_repairs,
+ can_read_repair_history=access.can_read_repairs,
+ can_read_components=access.can_read_components,
+ ),
+ ),
+ meta=MinerDetailOverviewMeta(
+ generated_at=datetime.now(timezone.utc),
+ database_query_count=query_count,
+ ),
+ )
diff --git a/backend/app/repairs/center_repository.py b/backend/app/repairs/center_repository.py
index 3b49dd2..3f287cd 100644
--- a/backend/app/repairs/center_repository.py
+++ b/backend/app/repairs/center_repository.py
@@ -315,6 +315,28 @@ class RepairCenterRepository:
connection.close()
return {"items": items, "summary": summary, "location_options": options}
+ def list_miner_repairs(self, miner_id) -> list[dict[str, Any]]:
+ """Return a bounded, exact repair history for one durable miner."""
+ ctes = self._ctes(show_miners=False, show_locations=False)
+ connection = self._connection_factory()
+ try:
+ with connection.cursor(row_factory=dict_row) as cur:
+ cur.execute(
+ f"""
+ {ctes}
+ SELECT candidate.* FROM candidates AS candidate
+ WHERE candidate.miner_id = %s
+ ORDER BY candidate.active DESC,
+ candidate.opened_at DESC NULLS LAST,
+ candidate.repair_id NULLS LAST
+ LIMIT 50
+ """,
+ (miner_id,),
+ )
+ return [dict(row) for row in cur.fetchall()]
+ finally:
+ connection.close()
+
def summarize_active_repairs(self) -> dict[str, Any]:
"""Return exact active counts without loading Repair Center rows."""
ctes = self._ctes(show_miners=False, show_locations=False)
diff --git a/backend/app/repairs/center_service.py b/backend/app/repairs/center_service.py
index d49ba64..f8fd90c 100644
--- a/backend/app/repairs/center_service.py
+++ b/backend/app/repairs/center_service.py
@@ -96,6 +96,28 @@ class RepairCenterService:
),
)
+ def list_miner_repairs(
+ self, miner_id, access: RepairCenterAccess
+ ) -> list[RepairCenterItem]:
+ items = []
+ for row in self.repository.list_miner_repairs(miner_id):
+ mapped = {
+ **row,
+ "miner_status": row.get("runtime_status"),
+ "latest_event_status": row.get("legacy_event_status"),
+ "latest_event_description": row.get("latest_context"),
+ "latest_event_at": row.get("latest_context_at"),
+ "asset_state": row.get("legacy_asset_state"),
+ "state_notes": row.get("latest_context"),
+ "reported_error": row.get("reported_problem"),
+ }
+ item = RepairCenterItem.model_validate(
+ {**mapped, "capabilities": RepairCenterCapabilities()}
+ )
+ item.capabilities = self._capabilities(item, access)
+ items.append(item)
+ return items
+
def summarize_active_repairs(self) -> ActiveRepairSummary:
return ActiveRepairSummary.model_validate(
self.repository.summarize_active_repairs()
diff --git a/backend/tests/test_miner_detail.py b/backend/tests/test_miner_detail.py
new file mode 100644
index 0000000..d6679d3
--- /dev/null
+++ b/backend/tests/test_miner_detail.py
@@ -0,0 +1,636 @@
+import ast
+import re
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import patch
+from uuid import UUID, uuid4
+
+import psycopg
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from app.config import settings
+from app.frontend.router import miner_detail_page, router as frontend_router
+from app.identity import get_current_user
+from app.inventory.exceptions import MinerNotFound
+from app.inventory.models import InventoryMinerDetail
+from app.inventory.repository import InventoryRepository
+from app.inventory.service import InventoryService
+from app.locations.models import MinerLocationDetail
+from app.locations.repository import LocationMapRepository
+from app.miner_detail.dependencies import get_miner_detail_service
+from app.miner_detail.models import MinerDetailAccess
+from app.miner_detail.router import router as detail_router
+from app.miner_detail.service import MinerDetailService
+from app.repairs.center_repository import RepairCenterRepository
+from app.repairs.models import RepairCenterCapabilities, RepairCenterItem
+
+
+NOW = datetime(2026, 8, 17, 12, 0, tzinfo=timezone.utc)
+APP_ROOT = Path(__file__).parents[1] / "app"
+STATIC_ROOT = APP_ROOT / "frontend" / "static"
+
+
+def current_miner(**updates):
+ values = {
+ "id": uuid4(),
+ "asset_id": "MOP-001",
+ "current_mac": "aa:bb:cc:dd:ee:ff",
+ "current_ip": "192.0.2.10",
+ "model": "M50",
+ "miner_sn": "MINER-SN",
+ "lifecycle_status": "active",
+ "runtime_status": "online",
+ "site_id": uuid4(),
+ "site_name": "Restricted Site",
+ "room_id": uuid4(),
+ "room_name": "Restricted Room",
+ "location_position_id": uuid4(),
+ "rack_label": "Rack A",
+ "shelf_label": "Shelf 1",
+ "position_label": "A-01",
+ "position_index": 1,
+ "observed_at": NOW,
+ "hashrate_ths": 0.0,
+ "power_w": 0.0,
+ "chip_temp_max_c": 0.0,
+ "hash_stable": None,
+ "tags": [],
+ "created_at": NOW - timedelta(days=10),
+ "updated_at": NOW,
+ "last_seen_at": NOW,
+ "hostname": "miner-001",
+ "control_board_sn": "CB-SN",
+ "psu_serial": "PSU-SN",
+ "firmware_version": "1.2.3",
+ "notes": None,
+ "first_seen_at": NOW - timedelta(days=9),
+ }
+ values.update(updates)
+ return InventoryMinerDetail(**values)
+
+
+def location():
+ return MinerLocationDetail(
+ site_id=uuid4(),
+ site_name="North Site",
+ room_id=uuid4(),
+ room_name="Room 1",
+ position_id=uuid4(),
+ rack_label="Rack A",
+ shelf_label="Shelf 1",
+ position_label="A-01",
+ position_index=1,
+ )
+
+
+def repair_item(*, source="modular", active=True):
+ return RepairCenterItem(
+ repair_id=uuid4() if source == "modular" else None,
+ source=source,
+ active=active,
+ opened_at=NOW - timedelta(hours=2),
+ repair_status="OPEN" if active else "COMPLETED",
+ reason="CONTROL_BOARD" if source == "modular" else None,
+ latest_context="Board diagnostics",
+ latest_context_at=NOW,
+ miner_id=uuid4(),
+ lifecycle_status="in_repair" if active else "active",
+ parts_count=1 if source == "modular" else None,
+ parts_summary=["control board"] if source == "modular" else [],
+ capabilities=RepairCenterCapabilities(),
+ )
+
+
+class FakeInventoryService:
+ def __init__(self, miner=None, *, missing=False):
+ self.miner = miner or current_miner()
+ self.missing = missing
+ self.calls = []
+
+ def get_current_miner(self, miner_id):
+ self.calls.append(miner_id)
+ if self.missing:
+ raise MinerNotFound(miner_id)
+ return self.miner
+
+
+class FakeLocationService:
+ def __init__(self, value=None):
+ self.value = value
+ self.calls = []
+
+ def get_miner_location(self, miner_id):
+ self.calls.append(miner_id)
+ return self.value
+
+
+class FakeRepairService:
+ def __init__(self, values=None):
+ self.values = list(values or [])
+ self.calls = []
+
+ def list_miner_repairs(self, miner_id, access):
+ self.calls.append((miner_id, access))
+ return self.values
+
+
+class MinerDetailServiceTests(unittest.TestCase):
+ def build(self, *, miner=None, mapped=None, repairs=None, missing=False):
+ self.inventory = FakeInventoryService(miner, missing=missing)
+ self.locations = FakeLocationService(mapped)
+ self.repairs = FakeRepairService(repairs)
+ return MinerDetailService(self.inventory, self.locations, self.repairs)
+
+ def test_full_access_uses_three_domain_queries_and_preserves_zero(self):
+ miner_id = uuid4()
+ mapped = location()
+ active = repair_item()
+ service = self.build(mapped=mapped, repairs=[active])
+ response = service.get_overview(
+ miner_id,
+ MinerDetailAccess(
+ can_view_locations=True,
+ can_read_repairs=True,
+ can_view_repairs=True,
+ can_read_components=True,
+ ),
+ )
+ self.assertEqual(self.inventory.calls, [miner_id])
+ self.assertEqual(self.locations.calls, [miner_id])
+ self.assertEqual(len(self.repairs.calls), 1)
+ self.assertEqual(response.meta.database_query_count, 3)
+ self.assertEqual(response.data.miner.hashrate_ths, 0)
+ self.assertEqual(response.data.miner.power_w, 0)
+ self.assertEqual(response.data.miner.chip_temp_max_c, 0)
+ self.assertIsNone(response.data.miner.hash_stable)
+ self.assertEqual(response.data.location.location, mapped)
+ self.assertEqual(response.data.repair.active_repair.repair_id, active.repair_id)
+ self.assertEqual(response.data.operational_state, "repair")
+ self.assertTrue(response.data.capabilities.can_view_map)
+ self.assertTrue(response.data.capabilities.can_open_repair_center)
+
+ def test_restricted_domains_are_not_queried_and_location_never_leaks(self):
+ service = self.build(mapped=location(), repairs=[repair_item()])
+ response = service.get_overview(uuid4(), MinerDetailAccess())
+ self.assertEqual(response.meta.database_query_count, 1)
+ self.assertEqual(self.locations.calls, [])
+ self.assertEqual(self.repairs.calls, [])
+ self.assertTrue(response.data.location.restricted)
+ self.assertIsNone(response.data.location.location)
+ self.assertTrue(response.data.repair.restricted)
+ self.assertEqual(response.data.repair.episodes, [])
+ miner = response.data.miner
+ for field in (
+ "site_id", "site_name", "room_id", "room_name",
+ "location_position_id", "rack_label", "shelf_label",
+ "position_label", "position_index",
+ ):
+ self.assertIsNone(getattr(miner, field), field)
+ serialized = response.model_dump_json()
+ self.assertNotIn("Restricted Site", serialized)
+ self.assertNotIn("Restricted Room", serialized)
+
+ def test_no_repair_and_legacy_repair_are_truthfully_distinct(self):
+ no_repair = self.build().get_overview(
+ uuid4(), MinerDetailAccess(can_read_repairs=True)
+ )
+ self.assertTrue(no_repair.data.repair.available)
+ self.assertFalse(no_repair.data.repair.restricted)
+ self.assertIsNone(no_repair.data.repair.active_repair)
+ self.assertEqual(no_repair.data.repair.episodes, [])
+
+ legacy = repair_item(source="legacy")
+ legacy_response = self.build(repairs=[legacy]).get_overview(
+ uuid4(), MinerDetailAccess(can_read_repairs=True)
+ )
+ self.assertIsNone(legacy_response.data.repair.active_repair)
+ self.assertEqual(legacy_response.data.repair.episodes[0].source, "legacy")
+ self.assertIsNone(legacy_response.data.repair.episodes[0].repair_id)
+
+ def test_unscoped_legacy_position_is_truthful_but_not_map_linkable(self):
+ unscoped = location().model_copy(
+ update={"site_id": None, "site_name": None, "room_id": None}
+ )
+ response = self.build(mapped=unscoped).get_overview(
+ uuid4(), MinerDetailAccess(can_view_locations=True)
+ )
+ self.assertTrue(response.data.location.available)
+ self.assertEqual(response.data.location.location.position_label, "A-01")
+ self.assertFalse(response.data.capabilities.can_view_map)
+
+ def test_factual_state_uses_no_temperature_or_age_threshold(self):
+ cases = (
+ ({"runtime_status": "offline", "chip_temp_max_c": 99.0}, "offline"),
+ ({"runtime_status": "online", "hash_stable": False}, "unstable"),
+ ({"runtime_status": "online", "observed_at": NOW - timedelta(days=90)}, "online"),
+ ({"runtime_status": "unknown", "chip_temp_max_c": 1000.0}, "unknown"),
+ ({"runtime_status": "online", "lifecycle_status": "spare"}, "spare"),
+ ({"runtime_status": "online", "lifecycle_status": "retired"}, "retired"),
+ )
+ for updates, expected in cases:
+ with self.subTest(expected=expected):
+ response = self.build(miner=current_miner(**updates)).get_overview(
+ uuid4(), MinerDetailAccess()
+ )
+ self.assertEqual(response.data.operational_state, expected)
+
+ response = self.build(
+ miner=current_miner(hashrate_ths=float("nan"))
+ ).get_overview(uuid4(), MinerDetailAccess())
+ self.assertIsNone(response.data.miner.hashrate_ths)
+ self.assertEqual(response.data.operational_state, "online")
+
+ def test_unknown_miner_propagates_domain_not_found_without_other_queries(self):
+ service = self.build(missing=True, mapped=location(), repairs=[repair_item()])
+ with self.assertRaises(MinerNotFound):
+ service.get_overview(
+ uuid4(),
+ MinerDetailAccess(can_view_locations=True, can_read_repairs=True),
+ )
+ self.assertEqual(self.locations.calls, [])
+ self.assertEqual(self.repairs.calls, [])
+
+
+class StubMinerDetailService:
+ def __init__(self, *, missing=False):
+ self.missing = missing
+ self.calls = []
+ self.delegate = MinerDetailService(
+ FakeInventoryService(),
+ FakeLocationService(location()),
+ FakeRepairService([repair_item()]),
+ )
+
+ def get_overview(self, miner_id, access):
+ self.calls.append((miner_id, access))
+ if self.missing:
+ raise MinerNotFound(miner_id)
+ return self.delegate.get_overview(miner_id, access)
+
+
+class MinerDetailRouterTests(unittest.TestCase):
+ def setUp(self):
+ self.service = StubMinerDetailService()
+ self.app = FastAPI()
+ self.app.include_router(detail_router)
+ self.app.dependency_overrides[get_miner_detail_service] = lambda: self.service
+ self.client = TestClient(self.app)
+
+ def tearDown(self):
+ self.client.close()
+
+ def authenticate(self, permissions):
+ self.app.dependency_overrides[get_current_user] = lambda: {
+ "id": str(uuid4()),
+ "role": "Custom",
+ "permissions": list(permissions),
+ }
+
+ def test_requires_authentication_and_miners_view(self):
+ path = f"/api/miners/{uuid4()}/overview"
+ self.assertEqual(self.client.get(path).status_code, 401)
+ self.authenticate(("locations.view", "repairs.read"))
+ self.assertEqual(self.client.get(path).status_code, 403)
+
+ def test_returns_typed_permission_aware_no_store_envelope(self):
+ miner_id = uuid4()
+ self.authenticate((
+ "miners.view", "locations.view", "repairs.read",
+ "repairs.view", "components.read",
+ ))
+ response = self.client.get(f"/api/miners/{miner_id}/overview")
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.headers["cache-control"], "no-store")
+ payload = response.json()
+ self.assertEqual(payload["data"]["miner"]["id"], str(self.service.delegate.inventory_service.miner.id))
+ self.assertEqual(payload["meta"]["database_query_count"], 3)
+ self.assertIsNone(payload["error"])
+ _, access = self.service.calls[-1]
+ self.assertTrue(access.can_view_locations)
+ self.assertTrue(access.can_read_repairs)
+ self.assertTrue(access.can_view_repairs)
+ self.assertTrue(access.can_read_components)
+
+ def test_malformed_and_unknown_uuid_use_existing_validation_and_404(self):
+ self.authenticate(("miners.view",))
+ self.assertEqual(self.client.get("/api/miners/not-a-uuid/overview").status_code, 422)
+ self.service.missing = True
+ response = self.client.get(f"/api/miners/{uuid4()}/overview")
+ self.assertEqual(response.status_code, 404)
+ self.assertEqual(response.json()["detail"], "Miner not found")
+
+
+class MinerDetailRepositoryIntegrationTests(unittest.TestCase):
+ def setUp(self):
+ self.connection = psycopg.connect(
+ host=settings.postgres_host,
+ port=settings.postgres_port,
+ dbname=settings.postgres_db,
+ user=settings.postgres_user,
+ password=settings.postgres_password,
+ )
+
+ def tearDown(self):
+ self.connection.rollback()
+ self.connection.close()
+
+ def test_inventory_and_location_projections_use_latest_factual_rows(self):
+ miner_id, site_id, room_id = uuid4(), uuid4(), uuid4()
+ earlier = NOW - timedelta(hours=1)
+ with self.connection.cursor() as cur:
+ cur.execute("INSERT INTO sites (id, name) VALUES (%s, 'Detail Site')", (site_id,))
+ cur.execute(
+ "INSERT INTO rooms (id, site_id, name) VALUES (%s, %s, 'Detail Room')",
+ (room_id, site_id),
+ )
+ cur.execute(
+ "INSERT INTO miners (id, asset_tag, current_mac, current_ip, model, "
+ "status, lifecycle_status, first_seen_at, last_seen_at) "
+ "VALUES (%s, %s, 'aa:bb:cc:00:00:01', '192.0.2.22', 'M50', "
+ "'online', 'active', NULL, NULL)",
+ (miner_id, "DETAIL-" + uuid4().hex[:8]),
+ )
+ cur.execute(
+ "INSERT INTO location_positions (room_id, room_name, rack_name, "
+ "shelf_name, position_index, position_label, miner_id) "
+ "VALUES (%s, 'Detail Room', 'Rack A', 'Shelf 1', 2, 'A-02', %s)",
+ (room_id, miner_id),
+ )
+ cur.execute(
+ "INSERT INTO miner_status_snapshots "
+ "(id, miner_id, observed_at, mhs_av, power_w, chip_temp_max_c, hash_stable) "
+ "VALUES (%s, %s, %s, 120000000, 3300, 70, true), "
+ "(%s, %s, %s, 0, 0, 0, NULL)",
+ (uuid4(), miner_id, earlier, uuid4(), miner_id, NOW),
+ )
+ inventory = InventoryService(
+ InventoryRepository(connection=self.connection)
+ ).get_current_miner(miner_id)
+ self.assertEqual(inventory.hashrate_ths, 0)
+ self.assertEqual(inventory.power_w, 0)
+ self.assertEqual(inventory.chip_temp_max_c, 0)
+ self.assertIsNone(inventory.hash_stable)
+ self.assertEqual(inventory.observed_at, NOW)
+ placement = LocationMapRepository(
+ connection=self.connection
+ ).get_miner_location(miner_id)
+ self.assertEqual(placement["site_id"], site_id)
+ self.assertEqual(placement["room_id"], room_id)
+ self.assertEqual(placement["rack_label"], "Rack A")
+ self.assertIsInstance(placement["position_id"], UUID)
+ self.assertEqual(RepairCenterRepository().list_miner_repairs(miner_id), [])
+
+
+class DummyCursor:
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *args):
+ return False
+
+ def execute(self, *args):
+ return None
+
+ def fetchall(self):
+ return []
+
+
+class DummyConnection:
+ def cursor(self, **kwargs):
+ return DummyCursor()
+
+ def close(self):
+ return None
+
+
+class NetworkIdentityPermissionTests(unittest.TestCase):
+ def setUp(self):
+ from app.main import list_miner_network_identities
+
+ self.app = FastAPI()
+ self.app.get("/api/miners/{miner_id}/network-identities")(
+ list_miner_network_identities
+ )
+ self.client = TestClient(self.app)
+
+ def tearDown(self):
+ self.client.close()
+
+ def test_network_history_requires_miners_view(self):
+ path = f"/api/miners/{uuid4()}/network-identities"
+ self.assertEqual(self.client.get(path).status_code, 401)
+ self.app.dependency_overrides[get_current_user] = lambda: {
+ "id": str(uuid4()), "role": "Custom", "permissions": ["locations.view"]
+ }
+ self.assertEqual(self.client.get(path).status_code, 403)
+ self.app.dependency_overrides[get_current_user] = lambda: {
+ "id": str(uuid4()), "role": "Custom", "permissions": ["miners.view"]
+ }
+ with patch("app.main.get_db_connection", return_value=DummyConnection()):
+ response = self.client.get(path)
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.json()["data"], [])
+
+
+class MinerDetailFrontendTests(unittest.TestCase):
+ def setUp(self):
+ self.app = FastAPI()
+ self.app.include_router(frontend_router)
+ self.app.dependency_overrides[get_current_user] = lambda: {
+ "id": str(uuid4()),
+ "username": "reader",
+ "display_name": "Detail Reader",
+ "role": "Read Only",
+ "permissions": [
+ "miners.view", "locations.view", "repairs.read", "repairs.view"
+ ],
+ }
+ self.client = TestClient(self.app)
+ self.miner_id = uuid4()
+
+ def tearDown(self):
+ self.client.close()
+
+ def test_page_requires_permission_and_renders_only_internal_alpha_shell(self):
+ path = f"/miners/{self.miner_id}/detail"
+ response = self.client.get(path)
+ self.assertEqual(response.status_code, 200)
+ for expected in (
+ "Miner Detail | MinerOps",
+ 'data-view="miner_detail"',
+ f'data-miner-id="{self.miner_id}"',
+ "/assets/minerops-miner-detail.js",
+ 'id="minerIdentityDetails"',
+ 'id="minerRepairEpisodes"',
+ 'id="minerObservations"',
+ ):
+ self.assertIn(expected, response.text)
+ for legacy in (
+ "minerops-miner-detail-extra", "minerops-miner-detail-clean-v3",
+ "minerops-location-management", "Manual Edit Missing Details",
+ "Reboot", "Performance Mode",
+ ):
+ self.assertNotIn(legacy, response.text)
+ self.app.dependency_overrides[get_current_user] = lambda: {
+ "id": str(uuid4()), "role": "Custom", "permissions": []
+ }
+ self.assertEqual(self.client.get(path).status_code, 403)
+
+ def test_return_to_accepts_protected_local_context_and_rejects_external(self):
+ path = f"/miners/{self.miner_id}/detail"
+ local = "/monitoring/map?site_id=" + str(uuid4()) + "&miner_id=" + str(self.miner_id)
+ response = self.client.get(path, params={"return_to": local})
+ self.assertIn("data-return-to=", response.text)
+ self.assertIn("/monitoring/map?site_id=", response.text)
+ response = self.client.get(
+ path, params={"return_to": "https://example.invalid/steal"}
+ )
+ self.assertNotIn("data-return-to=", response.text)
+ self.assertNotIn("example.invalid", response.text)
+
+ def test_refresh_action_has_visible_name_and_keeps_existing_behavior(self):
+ response = self.client.get(f"/miners/{self.miner_id}/detail")
+ self.assertRegex(
+ response.text,
+ r'',
+ )
+ source = (STATIC_ROOT / "miner-detail.js").read_text(encoding="utf-8")
+ self.assertIn(
+ 'el.refresh.addEventListener("click", loadOverview);',
+ source,
+ )
+ css = (STATIC_ROOT / "app.css").read_text(encoding="utf-8")
+ self.assertIn(".miner-detail-actions .quiet-button { color: var(--ink); }", css)
+
+ def test_live_route_precedence_and_middleware_isolation(self):
+ from app.main import app
+
+ detail_handlers = [
+ route.endpoint for route in app.routes
+ if route.path == "/miners/{miner_id}/detail"
+ and "GET" in (route.methods or set())
+ ]
+ self.assertGreaterEqual(len(detail_handlers), 2)
+ self.assertIs(detail_handlers[0], miner_detail_page)
+ current_user = {
+ "id": str(uuid4()), "username": "reader",
+ "display_name": "Detail Reader", "role": "Read Only",
+ "permissions": ["miners.view"],
+ }
+ app.dependency_overrides[get_current_user] = lambda: current_user
+ try:
+ with patch("app.identity.request_user_or_none", return_value=current_user):
+ with TestClient(app) as client:
+ response = client.get(f"/miners/{self.miner_id}/detail")
+ finally:
+ app.dependency_overrides.pop(get_current_user, None)
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.text.count('data-view="miner_detail"'), 1)
+ self.assertIn("/assets/minerops-miner-detail.js", response.text)
+ for legacy in (
+ "/assets/minerops-miner-detail-extra.js",
+ "/assets/minerops-miner-detail-extra-v2.js",
+ "/assets/minerops-miner-detail-clean-v3.js",
+ "/assets/minerops-location-management.js",
+ ):
+ self.assertNotIn(legacy, response.text)
+ paths = {route.path for route in app.routes}
+ for legacy_api in (
+ "/api/miners/{miner_id}",
+ "/api/miners/{miner_id}/detail-extra",
+ "/api/miners/{miner_id}/detail-clean-v3",
+ "/api/miners/{miner_id}/network-identities",
+ "/api/miners/{miner_id}/status-snapshots",
+ ):
+ self.assertIn(legacy_api, paths)
+
+ def test_frontend_is_lazy_read_only_bounded_and_xss_safe(self):
+ source = (STATIC_ROOT / "miner-detail.js").read_text(encoding="utf-8")
+ self.assertEqual(
+ source.count(
+ 'api("/api/miners/" + encodeURIComponent(minerId) + "/overview")'
+ ),
+ 1,
+ )
+ self.assertEqual(source.count("Promise.all(["), 2)
+ self.assertIn("/asset-id-history", source)
+ self.assertIn("/network-identities", source)
+ self.assertIn("/status-snapshots", source)
+ self.assertIn("/events", source)
+ self.assertIn("/parts", source)
+ self.assertNotIn("setInterval", source)
+ for mutation in (
+ 'method: "POST"', 'method: "PATCH"', 'method: "PUT"',
+ 'method: "DELETE"', "reboot", "change-mac", "manual-details",
+ ):
+ self.assertNotIn(mutation, source)
+ for unsafe in ("innerHTML", "outerHTML", "insertAdjacentHTML", "document.write"):
+ self.assertNotIn(unsafe, source)
+ self.assertIn("textContent", source)
+ self.assertIn("new URLSearchParams", source)
+ self.assertIn("encodeURIComponent", source)
+ self.assertNotRegex(
+ source.lower(),
+ r"(temperature|hashrate|stale).{0,40}(threshold|>=\s*90|>\s*90)",
+ )
+ main_source = (APP_ROOT / "main.py").read_text(encoding="utf-8")
+ snapshot = re.search(
+ r'def list_miner_status_snapshots\(.*?LIMIT 100;',
+ main_source,
+ re.DOTALL,
+ )
+ self.assertIsNotNone(snapshot)
+
+ def test_cross_navigation_preserves_context_with_durable_uuid(self):
+ fleet = (STATIC_ROOT / "fleet.js").read_text(encoding="utf-8")
+ repairs = (STATIC_ROOT / "repair-center.js").read_text(encoding="utf-8")
+ for source in (fleet, repairs):
+ self.assertIn('return_to', source)
+ self.assertIn("window.location.pathname + window.location.search", source)
+ self.assertIn("encodeURIComponent", source)
+ detail = (STATIC_ROOT / "miner-detail.js").read_text(encoding="utf-8")
+ self.assertIn('new URLSearchParams({ miner_id: minerIdValue })', detail)
+ self.assertIn('params.set("site_id", location.site_id)', detail)
+ self.assertIn('params.set("room_id", location.room_id)', detail)
+ self.assertIn('new URLSearchParams({ scope: "all", q: data.miner.id })', detail)
+
+
+class MinerDetailArchitectureTests(unittest.TestCase):
+ def test_router_and_service_preserve_dependency_boundaries(self):
+ for relative in ("miner_detail/router.py", "miner_detail/service.py"):
+ source = (APP_ROOT / relative).read_text(encoding="utf-8")
+ modules = []
+ for node in ast.walk(ast.parse(source)):
+ if isinstance(node, ast.Import):
+ modules.extend(alias.name for alias in node.names)
+ elif isinstance(node, ast.ImportFrom) and node.module:
+ modules.append(node.module)
+ self.assertFalse(
+ any(name.startswith(("psycopg", "sqlalchemy", "asyncpg"))
+ for name in modules)
+ )
+ for keyword in ("SELECT ", "INSERT ", "UPDATE ", "DELETE "):
+ self.assertNotIn(keyword, source.upper())
+ service_source = (APP_ROOT / "miner_detail" / "service.py").read_text(
+ encoding="utf-8"
+ )
+ self.assertNotIn("fastapi", service_source.lower())
+
+ def test_exactly_one_new_api_route_and_no_state_changing_page_get(self):
+ routes = [
+ (route.path, route.methods)
+ for route in detail_router.routes
+ ]
+ self.assertEqual(routes, [("/api/miners/{miner_id}/overview", {"GET"})])
+ page = next(
+ route for route in frontend_router.routes
+ if route.path == "/miners/{miner_id}/detail"
+ )
+ self.assertEqual(page.methods, {"GET"})
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/scripts/test-miner-detail.sh b/scripts/test-miner-detail.sh
new file mode 100755
index 0000000..7bebc4a
--- /dev/null
+++ b/scripts/test-miner-detail.sh
@@ -0,0 +1,102 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "${repo_root}"
+
+readonly source_files=(
+ backend/app/frontend/miner_detail_view.py
+ backend/app/frontend/router.py
+ backend/app/frontend/static/app.css
+ backend/app/frontend/static/fleet.js
+ backend/app/frontend/static/miner-detail.js
+ backend/app/frontend/static/repair-center.js
+ backend/app/frontend/views.py
+ backend/app/inventory/repository.py
+ backend/app/inventory/service.py
+ backend/app/locations/models.py
+ backend/app/locations/repository.py
+ backend/app/locations/service.py
+ backend/app/main.py
+ backend/app/miner_detail/__init__.py
+ backend/app/miner_detail/dependencies.py
+ backend/app/miner_detail/models.py
+ backend/app/miner_detail/router.py
+ backend/app/miner_detail/service.py
+ backend/app/repairs/center_repository.py
+ backend/app/repairs/center_service.py
+ backend/tests/test_miner_detail.py
+)
+
+for file in "${source_files[@]}"; do
+ if [[ ! -r "${file}" ]]; then
+ echo "Miner Detail test input is not readable: ${file}" >&2
+ exit 1
+ fi
+ if [[ "$(stat -c '%a' "${file}")" != "644" ]]; then
+ echo "Miner Detail source/test mode must be 0644: ${file}" >&2
+ exit 1
+ fi
+done
+if [[ ! -x scripts/test-miner-detail.sh ]] \
+ || [[ "$(stat -c '%a' scripts/test-miner-detail.sh)" != "755" ]]; then
+ echo "scripts/test-miner-detail.sh must be executable with mode 0755." >&2
+ exit 1
+fi
+echo "Miner Detail file-mode preflight passed."
+
+project_name="${MINEROPS_MINER_DETAIL_TEST_PROJECT:-minerops-miner-detail-$$}"
+compose=(
+ docker compose
+ --project-name "${project_name}"
+ --env-file .env.dev
+ -f compose.dev.yaml
+)
+
+cleanup() {
+ status=$?
+ if (( status != 0 )); then
+ echo
+ echo "Miner Detail tests failed. Recent PostgreSQL logs:"
+ "${compose[@]}" logs --no-color --tail 120 postgres || true
+ fi
+ "${compose[@]}" down --volumes --remove-orphans >/dev/null 2>&1 || true
+ exit "${status}"
+}
+trap cleanup EXIT
+
+if [[ ! -f .env.dev ]]; then
+ echo "Missing .env.dev. Run ./scripts/dev-env-init.sh first." >&2
+ exit 1
+fi
+
+echo "Starting disposable Miner Detail database..."
+"${compose[@]}" up --detach postgres
+for attempt in $(seq 1 60); do
+ if "${compose[@]}" exec -T postgres sh -ec \
+ 'pg_isready -U "$POSTGRES_USER" -d "$POSTGRES_DB"' >/dev/null 2>&1; then
+ break
+ fi
+ if (( attempt == 60 )); then
+ echo "PostgreSQL did not become ready." >&2
+ exit 1
+ fi
+ sleep 1
+done
+
+echo "Building Miner Detail test image..."
+"${compose[@]}" build backend
+"${compose[@]}" run --rm --no-deps backend alembic upgrade head
+
+echo "Running Miner Detail and affected-domain focused tests..."
+"${compose[@]}" run --rm --no-deps \
+ -v "${repo_root}/backend/tests:/app/tests:ro" \
+ backend python -m unittest -v \
+ tests.test_miner_detail \
+ tests.test_frontend_shell \
+ tests.test_inventory_domain \
+ tests.test_inventory_router \
+ tests.test_mine_map \
+ tests.test_repair_center
+
+echo "Miner Detail focused tests passed."