M8-T11: add WarmteLink discovery and history API

app/api/routes/api/meter_sources.py

@@ -12,10 +12,12 @@ from app.api.routes.api.deps import require_csrf, require_session
 from app.dependencies import get_db
 from app.integrations.meter_sources import SourceProfileError, list_source_profiles, sanitize_source_config
 from app.models.energy import DsmrReading, Meter
-from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel
+from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel, WarmteLinkReading
 from app.schemas.meter_source import (
-    BindingCreate, BindingListResponse, BindingPatch, BindingResponse, ChannelReadingResponse,
+    BindingCreate, BindingListResponse, BindingPatch, BindingResponse, ChannelBindingSummaryResponse,
+    ChannelReadingResponse,
     ChannelReadingsResponse, CommoditiesResponse, CommodityResponse, DiscoverResponse,
+    DiscoverChannelResponse,
     MeterSourceChannelListResponse, MeterSourceChannelResponse, MeterSourceCreate,
     MeterSourceListResponse, MeterSourcePatch, MeterSourceResponse, SourceConfigFieldResponse,
     SourceProfileResponse, SourceProfilesResponse,
@@ -181,8 +183,25 @@ def discover_source(source_uuid: str, db: Session = Depends(get_db),
                     _auth: AuthenticatedSession = Depends(require_session), _csrf: None = Depends(require_csrf)) -> DiscoverResponse:
     source = _source_or_404(db, source_uuid)
     if source.kind == "warmtelink_serial":
-        return DiscoverResponse(requested=False, supported=False, status="not_implemented",
-                                detail="Serial discovery is available after the WarmteLink worker is installed.")
+        if not source.enabled:
+            return DiscoverResponse(
+                requested=False, supported=True, status="error",
+                detail="The WarmteLink source is disabled.", channels=_discover_channels(db, source),
+            )
+        # This merely schedules lifecycle convergence.  It never opens a serial
+        # descriptor or waits for a frame in the request thread; the one managed
+        # worker remains the sole owner of serial I/O and can keep reconnecting.
+        request = warmtelink_worker_manager.request_discovery(source.id)
+        if request.completed.is_set():
+            # A worker may have accepted a frame during the bounded wait.
+            # Refresh only durable accepted metadata, never candidates/raw data.
+            db.expire_all()
+            source = _source_or_404(db, source_uuid)
+        return DiscoverResponse(
+            requested=request.status != "error", supported=True, status=request.status,
+            request_id=request.request_id or None, detail=request.detail,
+            channels=_discover_channels(db, source),
+        )
     return DiscoverResponse(requested=False, supported=True, status="managed_by_runtime",
                             detail="This source is discovered by its runtime subscription; no connection was opened.")
 
@@ -195,32 +214,61 @@ def source_channels(source_uuid: str, db: Session = Depends(get_db),
     items = []
     for channel in channels:
         bindings = list_bindings(db, channel_id=channel.id)
+        meter_ids = [binding.meter_id for binding in bindings]
         items.append(MeterSourceChannelResponse(
             uuid=channel.uuid, label=channel.label, suggested_commodity=channel.suggested_commodity,
             unit=channel.unit, device_type=channel.device_type, latest_value=channel.latest_value,
             latest_at=channel.latest_at, latest_quality=channel.latest_quality, binding_count=len(bindings),
-            bound_meter_ids=[binding.meter_id for binding in bindings],
+            bound_meter_ids=meter_ids,
+            binding_summary=ChannelBindingSummaryResponse(count=len(bindings), meter_ids=meter_ids),
         ))
-    return MeterSourceChannelListResponse(items=items, total=len(items))
+    return MeterSourceChannelListResponse(items=items, total=len(items), source_status=source.status)
 
 
 @router.get("/sources/{source_uuid}/channels/{channel_uuid}/readings", response_model=ChannelReadingsResponse)
 def channel_readings(source_uuid: str, channel_uuid: str, limit: int = Query(default=100, ge=1, le=1000),
-                     start: datetime | None = None, end: datetime | None = None, db: Session = Depends(get_db),
+                     from_: datetime | None = Query(default=None, alias="from"),
+                     to: datetime | None = Query(default=None), db: Session = Depends(get_db),
                      _auth: AuthenticatedSession = Depends(require_session)) -> ChannelReadingsResponse:
     source = _source_or_404(db, source_uuid)
-    _channel_or_404(db, source, channel_uuid)
-    if source.kind != "dsmr_mqtt":
-        return ChannelReadingsResponse(items=[], total=0)
+    channel = _channel_or_404(db, source, channel_uuid)
+    if from_ is not None and to is not None and _as_utc(from_) >= _as_utc(to):
+        raise HTTPException(status_code=422, detail="'from' must be earlier than 'to'.")
+    if source.kind == "warmtelink_serial":
+        statement = select(WarmteLinkReading).where(WarmteLinkReading.channel_id == channel.id)
+        model = WarmteLinkReading
+    else:
+        # DSMR remains a source-level protocol history.  Its channel is the
+        # public electricity identity, while payload/telegram diagnostics stay
+        # private to ingestion and the legacy latest endpoint.
         statement = select(DsmrReading).where(DsmrReading.meter_source_id == source.id)
-    if start is not None:
-        statement = statement.where(DsmrReading.recorded_at >= _as_utc(start))
-    if end is not None:
-        statement = statement.where(DsmrReading.recorded_at < _as_utc(end))
-    rows = list(db.execute(statement.order_by(DsmrReading.recorded_at.desc()).limit(limit)).scalars())
-    # The DSMR payload remains available only from its legacy compatibility endpoint;
-    # this generic endpoint intentionally exposes no telegram/equipment identifiers.
-    return ChannelReadingsResponse(items=[ChannelReadingResponse(recorded_at=row.recorded_at) for row in rows], total=len(rows))
+        model = DsmrReading
+    if from_ is not None:
+        statement = statement.where(model.recorded_at >= _as_utc(from_))
+    if to is not None:
+        statement = statement.where(model.recorded_at < _as_utc(to))
+    rows = list(db.execute(statement.order_by(model.recorded_at.asc()).limit(limit)).scalars())
+    return ChannelReadingsResponse(
+        items=[ChannelReadingResponse(
+            recorded_at=row.recorded_at,
+            value=getattr(row, "value", None), quality=getattr(row, "quality", None),
+        ) for row in rows],
+        total=len(rows),
+    )
+
+
+def _discover_channels(db: Session, source: MeterSource) -> list[DiscoverChannelResponse]:
+    """Return only public, accepted channel metadata for discover responses."""
+    return [
+        DiscoverChannelResponse(
+            uuid=channel.uuid, label=channel.label, unit=channel.unit,
+            latest_value=channel.latest_value, latest_at=channel.latest_at,
+            latest_quality=channel.latest_quality,
+        )
+        for channel in db.execute(
+            select(MeterSourceChannel).where(MeterSourceChannel.source_id == source.id)
+        ).scalars()
+    ]
 
 
 @router.get("/meters/{meter_id}/bindings", response_model=BindingListResponse)

app/schemas/meter_source.py

@@ -64,7 +64,23 @@ class DiscoverResponse(BaseModel):
     requested: bool
     supported: bool
     status: str
+    request_id: int | None = None
     detail: str | None = None
+    channels: list["DiscoverChannelResponse"] = Field(default_factory=list)
+
+
+class DiscoverChannelResponse(BaseModel):
+    uuid: str
+    label: str
+    unit: str
+    latest_value: Decimal | None
+    latest_at: datetime | None
+    latest_quality: str | None
+
+
+class ChannelBindingSummaryResponse(BaseModel):
+    count: int
+    meter_ids: list[int]
 
 
 class CommodityResponse(BaseModel):
@@ -88,11 +104,13 @@ class MeterSourceChannelResponse(BaseModel):
     latest_quality: str | None
     binding_count: int
     bound_meter_ids: list[int]
+    binding_summary: ChannelBindingSummaryResponse
 
 
 class MeterSourceChannelListResponse(BaseModel):
     items: list[MeterSourceChannelResponse]
     total: int
+    source_status: str
 
 
 class ChannelReadingResponse(BaseModel):

app/services/warmtelink_worker.py

@@ -8,8 +8,8 @@ from __future__ import annotations
 
 from collections.abc import Callable
 from contextlib import suppress
-from dataclasses import dataclass
-from datetime import UTC, datetime
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta
 import logging
 import threading
 from typing import Protocol
@@ -27,6 +27,9 @@ logger = logging.getLogger(__name__)
 
 _BACKOFF_SECONDS = (1, 2, 4, 8, 16, 32, 60)
 _JOIN_TIMEOUT_SECONDS = 5
+_DISCOVERY_LOCK_TIMEOUT_SECONDS = 0.05
+_DISCOVERY_WAIT_SECONDS = 0.1
+_DISCOVERY_TIMEOUT_SECONDS = 5
 
 
 class ReadOnlySerial(Protocol):
@@ -69,6 +72,18 @@ class _WorkerConfig:
     config: dict
 
 
+@dataclass
+class DiscoveryRequest:
+    """One source-scoped request, completed only by its serial owner."""
+
+    request_id: int
+    source_id: int
+    deadline: datetime
+    status: str = "pending"
+    detail: str | None = None
+    completed: threading.Event = field(default_factory=threading.Event)
+
+
 class WarmteLinkWorker:
     """One interruptible, read-only serial loop for one meter source."""
 
@@ -88,6 +103,8 @@ class WarmteLinkWorker:
         self._clock = clock or _EventClock()
         self._serial: ReadOnlySerial | None = None
         self._serial_lock = threading.Lock()
+        self._discovery_lock = threading.Lock()
+        self._discoveries: list[DiscoveryRequest] = []
         # Never inherit a daemon flag from a caller's background thread: a serial
         # descriptor and its orderly shutdown must remain visible to the process.
         self._thread = threading.Thread(
@@ -105,6 +122,35 @@ class WarmteLinkWorker:
         self._stop_event.set()
         self._close_serial()
 
+    def request_discovery(self, request: DiscoveryRequest) -> None:
+        """Queue a read request; this worker remains the sole serial owner."""
+        with self._discovery_lock:
+            self._discoveries.append(request)
+
+    def _finish_discoveries(self, status: str, detail: str | None = None) -> None:
+        now = datetime.now(UTC)
+        with self._discovery_lock:
+            pending, self._discoveries = self._discoveries, []
+        for request in pending:
+            if request.completed.is_set():
+                continue
+            if request.deadline <= now and status == "completed":
+                request.status, request.detail = "error", "Discovery timed out."
+            else:
+                request.status, request.detail = status, detail
+            request.completed.set()
+
+    def _expire_discoveries(self) -> None:
+        now = datetime.now(UTC)
+        with self._discovery_lock:
+            expired = [request for request in self._discoveries if request.deadline <= now]
+            self._discoveries = [request for request in self._discoveries if request.deadline > now]
+        for request in expired:
+            if request.completed.is_set():
+                continue
+            request.status, request.detail = "error", "Discovery timed out."
+            request.completed.set()
+
     def join(self, timeout: float = _JOIN_TIMEOUT_SECONDS) -> bool:
         self._thread.join(timeout)
         return not self._thread.is_alive()
@@ -142,6 +188,7 @@ class WarmteLinkWorker:
                         return
                     self._serial = device
                 while not self._stop_event.is_set():
+                    self._expire_discoveries()
                     chunk = device.read(1024)
                     if not chunk:
                         # ``timeout`` reads are normal, but still yield so a bad
@@ -152,13 +199,16 @@ class WarmteLinkWorker:
                     for frame in frames:
                         if self._stop_event.is_set():
                             break
-                        self._ingestor.handle_frame(
+                        admitted = self._ingestor.handle_frame(
                             self.source_id, frame, session_factory=self._session_factory
                         )
+                        if admitted:
+                            self._finish_discoveries("completed")
                         # A complete frame proves transport recovery even if its
                         # contents are rejected by the privacy/admission layer.
                         backoff_index = 0
             except Exception:
+                self._finish_discoveries("error", "WarmteLink discovery failed.")
                 self._record_error("WarmteLink serial connection failed")
                 delay = _BACKOFF_SECONDS[min(backoff_index, len(_BACKOFF_SECONDS) - 1)]
                 backoff_index += 1
@@ -181,6 +231,7 @@ class WarmteLinkWorkerManager:
         self._workers: dict[int, tuple[_WorkerConfig, WarmteLinkWorker]] = {}
         self._lock = threading.Lock()
         self._reapers: set[int] = set()
+        self._next_discovery_id = 0
         self._shutting_down = False
 
     @property
@@ -203,6 +254,48 @@ class WarmteLinkWorkerManager:
             self._shutting_down = False
         self.reconcile()
 
+    def request_discovery(self, source_id: int) -> DiscoveryRequest:
+        """Ask the current source worker for one bounded read/discovery attempt.
+
+        This intentionally does not reconcile or open a descriptor.  Lifecycle
+        convergence remains separate; a request can neither replace nor stop a
+        worker when an HTTP client times out or disconnects.
+        """
+        now = datetime.now(UTC)
+        request = DiscoveryRequest(0, source_id, now)
+        if not self._lock.acquire(timeout=_DISCOVERY_LOCK_TIMEOUT_SECONDS):
+            request.status, request.detail = "error", "Discovery queue is busy."
+            request.completed.set()
+            return request
+        try:
+            self._next_discovery_id += 1
+            request.request_id = self._next_discovery_id
+            request.deadline = now + timedelta(seconds=_DISCOVERY_TIMEOUT_SECONDS)
+            if self._shutting_down:
+                request.status, request.detail = "error", "WarmteLink manager is stopped."
+                request.completed.set()
+            elif (entry := self._workers.get(source_id)) is None:
+                request.status, request.detail = "error", "WarmteLink worker is not running."
+                request.completed.set()
+            else:
+                entry[1].request_discovery(request)
+                timer = threading.Timer(_DISCOVERY_TIMEOUT_SECONDS, self._timeout_discovery, args=(request,))
+                timer.daemon = True
+                timer.start()
+        finally:
+            self._lock.release()
+        # A tiny bounded wait makes an immediately available frame observable,
+        # without turning an HTTP call into serial I/O or an unbounded wait.
+        request.completed.wait(_DISCOVERY_WAIT_SECONDS)
+        return request
+
+    @staticmethod
+    def _timeout_discovery(request: DiscoveryRequest) -> None:
+        """Resolve a stale HTTP request without touching its healthy worker."""
+        if not request.completed.is_set():
+            request.status, request.detail = "error", "Discovery timed out."
+            request.completed.set()
+
     def _read_desired(self) -> dict[int, _WorkerConfig]:
         with self._session_factory() as session:
             return {

docs/design/m8-warmtelink-energy.md

@@ -689,7 +689,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接
 
 ### M8-T11 — WarmteLink discover、latest 与 history API
 
-- **Status**: `todo`
+- **Status**: `done`
 - **Depends**: M8-T10
 - **Context**: worker 链路稳定后，把一次发现、状态和规范化历史接到已建立的 source API。
 

frontend/src/api/schema.d.ts

@@ -1453,6 +1453,13 @@ export interface components {
             /** Enabled */
             enabled: boolean;
         };
+        /** ChannelBindingSummaryResponse */
+        ChannelBindingSummaryResponse: {
+            /** Count */
+            count: number;
+            /** Meter Ids */
+            meter_ids: number[];
+        };
         /** ChannelReadingResponse */
         ChannelReadingResponse: {
             /**
@@ -1753,6 +1760,21 @@ export interface components {
             /** Name */
             name: string;
         };
+        /** DiscoverChannelResponse */
+        DiscoverChannelResponse: {
+            /** Uuid */
+            uuid: string;
+            /** Label */
+            label: string;
+            /** Unit */
+            unit: string;
+            /** Latest Value */
+            latest_value: string | null;
+            /** Latest At */
+            latest_at: string | null;
+            /** Latest Quality */
+            latest_quality: string | null;
+        };
         /** DiscoverResponse */
         DiscoverResponse: {
             /** Requested */
@@ -1761,8 +1783,12 @@ export interface components {
             supported: boolean;
             /** Status */
             status: string;
+            /** Request Id */
+            request_id?: number | null;
             /** Detail */
             detail?: string | null;
+            /** Channels */
+            channels?: components["schemas"]["DiscoverChannelResponse"][];
         };
         /**
          * DsmrLatestResponse
@@ -2051,6 +2077,8 @@ export interface components {
             items: components["schemas"]["MeterSourceChannelResponse"][];
             /** Total */
             total: number;
+            /** Source Status */
+            source_status: string;
         };
         /** MeterSourceChannelResponse */
         MeterSourceChannelResponse: {
@@ -2074,6 +2102,7 @@ export interface components {
             binding_count: number;
             /** Bound Meter Ids */
             bound_meter_ids: number[];
+            binding_summary: components["schemas"]["ChannelBindingSummaryResponse"];
         };
         /** MeterSourceCreate */
         MeterSourceCreate: {
@@ -3977,8 +4006,8 @@ export interface operations {
         parameters: {
             query?: {
                 limit?: number;
-                start?: string | null;
-                end?: string | null;
+                from?: string | null;
+                to?: string | null;
             };
             header?: never;
             path: {

openapi/openapi.json

@@ -1945,7 +1945,7 @@
             }
           },
           {
-            "name": "start",
+            "name": "from",
             "in": "query",
             "required": false,
             "schema": {
@@ -1958,11 +1958,11 @@
                   "type": "null"
                 }
               ],
-              "title": "Start"
+              "title": "From"
             }
           },
           {
-            "name": "end",
+            "name": "to",
             "in": "query",
             "required": false,
             "schema": {
@@ -1975,7 +1975,7 @@
                   "type": "null"
                 }
               ],
-              "title": "End"
+              "title": "To"
             }
           }
         ],
@@ -3462,6 +3462,27 @@
         "title": "CatalogEntrySchema",
         "description": "An entity from the catalog with its current toggle state."
       },
+      "ChannelBindingSummaryResponse": {
+        "properties": {
+          "count": {
+            "type": "integer",
+            "title": "Count"
+          },
+          "meter_ids": {
+            "items": {
+              "type": "integer"
+            },
+            "type": "array",
+            "title": "Meter Ids"
+          }
+        },
+        "type": "object",
+        "required": [
+          "count",
+          "meter_ids"
+        ],
+        "title": "ChannelBindingSummaryResponse"
+      },
       "ChannelReadingResponse": {
         "properties": {
           "recorded_at": {
@@ -4055,6 +4076,67 @@
         "title": "DeviceInfoSchema",
         "description": "HA device grouping info for an exposable entity."
       },
+      "DiscoverChannelResponse": {
+        "properties": {
+          "uuid": {
+            "type": "string",
+            "title": "Uuid"
+          },
+          "label": {
+            "type": "string",
+            "title": "Label"
+          },
+          "unit": {
+            "type": "string",
+            "title": "Unit"
+          },
+          "latest_value": {
+            "anyOf": [
+              {
+                "type": "string",
+                "pattern": "^(?!^[-+.]*$)[+-]?0*\\d*\\.?\\d*$"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Latest Value"
+          },
+          "latest_at": {
+            "anyOf": [
+              {
+                "type": "string",
+                "format": "date-time"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Latest At"
+          },
+          "latest_quality": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Latest Quality"
+          }
+        },
+        "type": "object",
+        "required": [
+          "uuid",
+          "label",
+          "unit",
+          "latest_value",
+          "latest_at",
+          "latest_quality"
+        ],
+        "title": "DiscoverChannelResponse"
+      },
       "DiscoverResponse": {
         "properties": {
           "requested": {
@@ -4069,6 +4151,17 @@
             "type": "string",
             "title": "Status"
           },
+          "request_id": {
+            "anyOf": [
+              {
+                "type": "integer"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Request Id"
+          },
           "detail": {
             "anyOf": [
               {
@@ -4079,6 +4172,13 @@
               }
             ],
             "title": "Detail"
+          },
+          "channels": {
+            "items": {
+              "$ref": "#/components/schemas/DiscoverChannelResponse"
+            },
+            "type": "array",
+            "title": "Channels"
           }
         },
         "type": "object",
@@ -4691,12 +4791,17 @@
           "total": {
             "type": "integer",
             "title": "Total"
+          },
+          "source_status": {
+            "type": "string",
+            "title": "Source Status"
           }
         },
         "type": "object",
         "required": [
           "items",
-          "total"
+          "total",
+          "source_status"
         ],
         "title": "MeterSourceChannelListResponse"
       },
@@ -4781,6 +4886,9 @@
             },
             "type": "array",
             "title": "Bound Meter Ids"
+          },
+          "binding_summary": {
+            "$ref": "#/components/schemas/ChannelBindingSummaryResponse"
           }
         },
         "type": "object",
@@ -4794,7 +4902,8 @@
           "latest_at",
           "latest_quality",
           "binding_count",
-          "bound_meter_ids"
+          "bound_meter_ids",
+          "binding_summary"
         ],
         "title": "MeterSourceChannelResponse"
       },

openapi/openapi.yaml

@@ -1467,7 +1467,7 @@ paths:
           minimum: 1
           default: 100
           title: Limit
-      - name: start
+      - name: from
         in: query
         required: false
         schema:
@@ -1475,8 +1475,8 @@ paths:
           - type: string
             format: date-time
           - type: 'null'
-          title: Start
-      - name: end
+          title: From
+      - name: to
         in: query
         required: false
         schema:
@@ -1484,7 +1484,7 @@ paths:
           - type: string
             format: date-time
           - type: 'null'
-          title: End
+          title: To
       responses:
         '200':
           description: Successful Response
@@ -2597,6 +2597,21 @@ components:
       - enabled
       title: CatalogEntrySchema
       description: An entity from the catalog with its current toggle state.
+    ChannelBindingSummaryResponse:
+      properties:
+        count:
+          type: integer
+          title: Count
+        meter_ids:
+          items:
+            type: integer
+          type: array
+          title: Meter Ids
+      type: object
+      required:
+      - count
+      - meter_ids
+      title: ChannelBindingSummaryResponse
     ChannelReadingResponse:
       properties:
         recorded_at:
@@ -3050,6 +3065,43 @@ components:
       - name
       title: DeviceInfoSchema
       description: HA device grouping info for an exposable entity.
+    DiscoverChannelResponse:
+      properties:
+        uuid:
+          type: string
+          title: Uuid
+        label:
+          type: string
+          title: Label
+        unit:
+          type: string
+          title: Unit
+        latest_value:
+          anyOf:
+          - type: string
+            pattern: ^(?!^[-+.]*$)[+-]?0*\d*\.?\d*$
+          - type: 'null'
+          title: Latest Value
+        latest_at:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Latest At
+        latest_quality:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Latest Quality
+      type: object
+      required:
+      - uuid
+      - label
+      - unit
+      - latest_value
+      - latest_at
+      - latest_quality
+      title: DiscoverChannelResponse
     DiscoverResponse:
       properties:
         requested:
@@ -3061,11 +3113,21 @@ components:
         status:
           type: string
           title: Status
+        request_id:
+          anyOf:
+          - type: integer
+          - type: 'null'
+          title: Request Id
         detail:
           anyOf:
           - type: string
           - type: 'null'
           title: Detail
+        channels:
+          items:
+            $ref: '#/components/schemas/DiscoverChannelResponse'
+          type: array
+          title: Channels
       type: object
       required:
       - requested
@@ -3542,10 +3604,14 @@ components:
         total:
           type: integer
           title: Total
+        source_status:
+          type: string
+          title: Source Status
       type: object
       required:
       - items
       - total
+      - source_status
       title: MeterSourceChannelListResponse
     MeterSourceChannelResponse:
       properties:
@@ -3593,6 +3659,8 @@ components:
             type: integer
           type: array
           title: Bound Meter Ids
+        binding_summary:
+          $ref: '#/components/schemas/ChannelBindingSummaryResponse'
       type: object
       required:
       - uuid
@@ -3605,6 +3673,7 @@ components:
       - latest_quality
       - binding_count
       - bound_meter_ids
+      - binding_summary
       title: MeterSourceChannelResponse
     MeterSourceCreate:
       properties:

tests/test_meter_source_api.py

@@ -2,7 +2,12 @@
 
 from __future__ import annotations
 
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+from queue import Empty, Queue
+from types import SimpleNamespace
+import threading
+import time
 from unittest.mock import patch
 
 from fastapi.testclient import TestClient
@@ -10,7 +15,7 @@ from sqlalchemy import create_engine, select
 from sqlalchemy.orm import Session
 
 from app.models.energy import DsmrReading, Meter
-from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel
+from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel, WarmteLinkReading
 
 _CSRF = "test-csrf-token"
 
@@ -212,6 +217,7 @@ def test_source_and_binding_error_contracts_csrf_timezone_and_dsmr_compatibility
         readings = client.get(f"/api/energy/sources/{source_uuid}/channels/{channel_uuid}/readings")
         assert readings.status_code == 200
         assert readings.json()["total"] == 1
+        assert readings.json()["items"] == [{"recorded_at": now.isoformat().replace("+00:00", ""), "value": None, "quality": None}]
         assert "telegram_id" not in readings.text
         latest = client.get("/api/energy/dsmr/latest")
         assert latest.status_code == 200
@@ -314,7 +320,9 @@ def test_source_channel_binding_response_contract_and_discover_capabilities(auth
         assert discovered.status_code == 200
         assert discovered.json() == {
             "requested": False, "supported": True, "status": "managed_by_runtime",
+            "request_id": None,
             "detail": "This source is discovered by its runtime subscription; no connection was opened.",
+            "channels": [],
         }
         channels = client.get(f"/api/energy/sources/{source['uuid']}/channels")
         assert channels.status_code == 200
@@ -322,8 +330,9 @@ def test_source_channel_binding_response_contract_and_discover_capabilities(auth
         assert channel["uuid"] == channel_uuid
         assert set(channel) == {
             "uuid", "label", "suggested_commodity", "unit", "device_type", "latest_value",
-            "latest_at", "latest_quality", "binding_count", "bound_meter_ids",
+            "latest_at", "latest_quality", "binding_count", "bound_meter_ids", "binding_summary",
         }
+        assert channels.json()["source_status"] == "unknown"
 
         meter = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json={
             "label": "Contract meter", "started_at": "2030-01-01T00:00:00Z", "reason": "initial",
@@ -381,3 +390,261 @@ def test_binding_patch_omitted_null_and_adjacent_half_open_boundaries(auth_datab
         })
         assert adjacent.status_code == 201
     engine.dispose()
+
+
+def test_warmtelink_discover_and_minute_history_are_bounded_and_private(auth_database, monkeypatch):
+    """Discover delegates to the manager; readings expose accepted minute samples only."""
+    from app.api.routes.api import meter_sources
+
+    requested: list[int] = []
+    monkeypatch.setattr(
+        meter_sources.warmtelink_worker_manager, "request_discovery",
+        lambda source_id: requested.append(source_id) or SimpleNamespace(
+            status="completed", request_id=1, detail=None, completed=SimpleNamespace(is_set=lambda: False),
+        ),
+    )
+    monkeypatch.setattr(meter_sources.warmtelink_worker_manager, "reconcile", lambda: None)
+    client, engine = _client(auth_database)
+    with client:
+        _login(client)
+        created = client.post("/api/energy/sources", headers={"X-CSRF-Token": _CSRF}, json={
+            "name": "WarmteLink", "kind": "warmtelink_serial", "config": {"path": "/dev/fake"},
+        })
+        assert created.status_code == 201
+        source_uuid = created.json()["uuid"]
+        now = datetime(2030, 1, 1, 12, 0, 30, tzinfo=UTC)
+        with Session(engine) as session:
+            source = session.execute(select(MeterSource).where(MeterSource.uuid == source_uuid)).scalar_one()
+            source.status = "online"
+            channel = MeterSourceChannel(
+                source_id=source.id, channel_key="heating", label="Heating", unit="GJ",
+                latest_value=Decimal("7.002"), latest_at=now, latest_quality="unverifiable",
+                created_at=now, updated_at=now,
+            )
+            session.add(channel)
+            session.flush()
+            session.add_all([
+                WarmteLinkReading(
+                    channel_id=channel.id, recorded_at=now - timedelta(minutes=1), received_at=now,
+                    value=Decimal("7.001"), unit="GJ", quality="unverifiable", equipment_fingerprint="masked",
+                ),
+                WarmteLinkReading(
+                    channel_id=channel.id, recorded_at=now, received_at=now,
+                    value=Decimal("7.002"), unit="GJ", quality="unverifiable", equipment_fingerprint="masked",
+                ),
+            ])
+            session.commit()
+            channel_uuid = channel.uuid
+
+        discover = client.post(f"/api/energy/sources/{source_uuid}/discover", headers={"X-CSRF-Token": _CSRF})
+        assert discover.status_code == 200
+        assert discover.json()["status"] == "completed"
+        assert requested and "fingerprint" not in discover.text and "channel_key" not in discover.text
+        assert discover.json()["channels"][0]["latest_quality"] == "unverifiable"
+
+        history = client.get(
+            f"/api/energy/sources/{source_uuid}/channels/{channel_uuid}/readings",
+            params={"from": "2030-01-01T11:59:00Z", "to": "2030-01-01T12:01:00Z", "limit": 1},
+        )
+        assert history.status_code == 200
+        assert history.json()["total"] == 1
+        assert history.json()["items"] == [{
+            "recorded_at": "2030-01-01T11:59:30", "value": "7.001", "quality": "unverifiable",
+        }]
+        assert client.get(
+            f"/api/energy/sources/{source_uuid}/channels/{channel_uuid}/readings",
+            params={"from": "2030-01-01T12:01:00Z", "to": "2030-01-01T12:00:00Z"},
+        ).status_code == 422
+        assert client.get(
+            f"/api/energy/sources/{source_uuid}/channels/{channel_uuid}/readings", params={"limit": 0}
+        ).status_code == 422
+    engine.dispose()
+
+
+def test_warmtelink_discover_auth_csrf_disabled_and_source_ownership(auth_database, monkeypatch):
+    from app.api.routes.api import meter_sources
+
+    monkeypatch.setattr(
+        meter_sources.warmtelink_worker_manager, "request_discovery",
+        lambda _source_id: SimpleNamespace(
+            status="pending", request_id=1, detail=None, completed=SimpleNamespace(is_set=lambda: False),
+        ),
+    )
+    monkeypatch.setattr(meter_sources.warmtelink_worker_manager, "reconcile", lambda: None)
+    client, engine = _client(auth_database)
+    with client:
+        serial = client.post("/api/energy/sources", headers={"X-CSRF-Token": _CSRF}, json={
+            "name": "Serial", "kind": "warmtelink_serial", "enabled": False, "config": {"path": "/dev/fake"},
+        })
+        assert serial.status_code == 401  # no session yet
+        _login(client)
+        serial = client.post("/api/energy/sources", headers={"X-CSRF-Token": _CSRF}, json={
+            "name": "Serial", "kind": "warmtelink_serial", "enabled": False, "config": {"path": "/dev/fake"},
+        })
+        other = _create_source(client)
+        channel_uuid = _add_channel(engine, other["uuid"])
+        assert client.post(f"/api/energy/sources/{serial.json()['uuid']}/discover").status_code == 403
+        disabled = client.post(
+            f"/api/energy/sources/{serial.json()['uuid']}/discover", headers={"X-CSRF-Token": _CSRF}
+        )
+        assert disabled.status_code == 200 and disabled.json()["status"] == "error"
+        assert client.get(
+            f"/api/energy/sources/{serial.json()['uuid']}/channels/{channel_uuid}/readings"
+        ).status_code == 404
+    engine.dispose()
+
+
+def test_discovery_manager_is_source_scoped_and_never_replaces_a_worker(auth_database):
+    """The real manager queues requests on one fake read-only serial owner."""
+    from app.services.warmtelink_worker import WarmteLinkWorkerManager
+
+    engine = create_engine(auth_database["app_url"], connect_args={"check_same_thread": False})
+    with Session(engine) as session:
+        now = datetime.now(UTC)
+        source = MeterSource(
+            name="Serial", kind="warmtelink_serial", enabled=True, config={"path": "/dev/fake"},
+            created_at=now, updated_at=now,
+        )
+        session.add(source)
+        session.commit()
+        source_id = source.id
+
+    class FakeReadOnlyWorker:
+        instances: list["FakeReadOnlyWorker"] = []
+
+        def __init__(self, _source_id, _config, **_kwargs):
+            self.requests = []
+            self.thread = SimpleNamespace(is_alive=lambda: True)
+            self.__class__.instances.append(self)
+
+        def start(self):
+            return None
+
+        def stop(self):
+            return None
+
+        def join(self, timeout=5):
+            return True
+
+        def request_discovery(self, request):
+            self.requests.append(request)
+
+    manager = WarmteLinkWorkerManager(
+        session_factory=lambda: Session(engine), worker_factory=FakeReadOnlyWorker,
+    )
+    manager.reconcile()
+    assert manager.worker_count == 1
+    first = manager.request_discovery(source_id)
+    assert first.status == "pending"
+    worker = FakeReadOnlyWorker.instances[0]
+    assert len(worker.requests) == 1
+    # A client timing out/cancelling leaves the queued request and worker alone;
+    # completing it later cannot open another descriptor or create bindings.
+    worker.requests[0].status = "completed"
+    worker.requests[0].completed.set()
+    results = []
+    threads = [threading.Thread(target=lambda: results.append(manager.request_discovery(source_id))) for _ in range(2)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+    assert manager.worker_count == 1
+    assert len(FakeReadOnlyWorker.instances) == 1
+    assert len(worker.requests) == 3
+    with Session(engine) as session:
+        assert session.query(Meter).count() == 0
+        assert session.query(MeterSourceBinding).count() == 0
+    manager.shutdown()
+    engine.dispose()
+
+
+def test_real_warmtelink_discovery_waits_for_admission_and_keeps_its_serial_owner(
+    auth_database, monkeypatch,
+):
+    """A rejected candidate is neither a discovery success nor exposed metadata."""
+    from app.integrations.p1 import dsmr_crc16
+    from app.services import warmtelink_worker
+    from app.services.warmtelink_worker import WarmteLinkWorkerManager
+
+    engine = create_engine(auth_database["app_url"], connect_args={"check_same_thread": False})
+    with Session(engine) as session:
+        now = datetime.now(UTC)
+        source = MeterSource(
+            name="Serial", kind="warmtelink_serial", enabled=True, config={"path": "/dev/fake"},
+            created_at=now, updated_at=now,
+        )
+        session.add(source)
+        session.commit()
+        source_id = source.id
+
+    def frame(second: int, *, crc: bool = False) -> bytes:
+        body = (
+            b"/WARMTE\r\n"
+            + f"0-0:1.0.0(2608221200{second:02d}S)\r\n".encode()
+            + b"0-0:96.1.1(REDACTED)\r\n"
+            + b"0-1:24.1.0(006)\r\n"
+            + b"0-1:96.1.0(REDACTED)\r\n"
+            + f"0-1:24.2.1(2608221200{second:02d}S)(5.900*m3)\r\n".encode()
+            + b"0-2:24.1.0(012)\r\n"
+            + b"0-2:96.1.0(REDACTED)\r\n"
+            + f"0-2:24.2.1(2608221200{second:02d}S)(0.017*GJ)\r\n".encode()
+        )
+        payload = body + b"!"
+        return payload + (f"{dsmr_crc16(payload):04X}".encode() if crc else b"") + b"\r\n"
+
+    class FakeReadOnlySerial:
+        instances: list["FakeReadOnlySerial"] = []
+
+        def __init__(self):
+            self.frames: Queue[bytes] = Queue()
+            self.closed = False
+            self.__class__.instances.append(self)
+
+        def read(self, _size: int = 1) -> bytes:
+            try:
+                return self.frames.get(timeout=0.01)
+            except Empty:
+                return b""
+
+        def close(self) -> None:
+            self.closed = True
+
+    monkeypatch.setattr(warmtelink_worker, "_DISCOVERY_TIMEOUT_SECONDS", 0.15)
+    monkeypatch.setattr(warmtelink_worker, "_DISCOVERY_WAIT_SECONDS", 0.02)
+    manager = WarmteLinkWorkerManager(
+        session_factory=lambda: Session(engine), serial_factory=lambda _config: FakeReadOnlySerial(),
+    )
+    manager.reconcile()
+    serial = FakeReadOnlySerial.instances[0]
+
+    request = manager.request_discovery(source_id)
+    assert request.status == "pending"
+    serial.frames.put(frame(0))  # First unverifiable candidate is not admitted.
+    time.sleep(0.04)
+    assert not request.completed.is_set()
+    with Session(engine) as session:
+        assert session.query(MeterSourceChannel).filter_by(source_id=source_id).count() == 0
+        assert session.query(MeterSourceBinding).count() == 0
+
+    serial.frames.put(frame(10))  # Strictly continuous successor admits both channels.
+    assert request.completed.wait(1)
+    assert request.status == "completed"
+    with Session(engine) as session:
+        assert session.query(MeterSourceChannel).filter_by(source_id=source_id).count() == 2
+        assert session.query(MeterSourceBinding).count() == 0
+
+    rejected = manager.request_discovery(source_id)
+    assert rejected.status == "pending"
+    serial.frames.put(b"/malformed!\r\n")
+    assert rejected.completed.wait(1)
+    assert rejected.status == "error" and rejected.detail == "Discovery timed out."
+    assert manager.worker_count == 1 and len(FakeReadOnlySerial.instances) == 1
+
+    recovered = manager.request_discovery(source_id)
+    serial.frames.put(frame(20, crc=True))
+    assert recovered.completed.wait(1)
+    assert recovered.status == "completed"
+    assert manager.worker_count == 1 and len(FakeReadOnlySerial.instances) == 1
+    manager.shutdown()
+    assert serial.closed
+    engine.dispose()