Ingestion: Home Assistant WebSocket push path

HomeAssistantWebSocketWorker holds a persistent state_changed subscription per HA endpoint that opts in via the connector's WebSocket toggle (HaEndpointConfig.UseWebSocket): auth handshake, subscribe, ingest in real time, capped-backoff reconnect. The REST poll worker skips WS endpoints so each is served once. HaWebSocketProtocol holds the pure handshake/parse logic. Verified by 11 protocol unit tests + a live integration test against an in-process fake HA server. CLAUDE.md updated.

Claude-Session: https://claude.ai/code/session_01Kib2MniVFbD95fkgLgBBnB

CLAUDE.md

@@ -6,7 +6,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 
 MeterVault is a self-hosted, local-first energy & utility metering platform: it ingests meter data from Home Assistant, Tasmota and MQTT on a schedule, stores every reading timestamped and immutable, normalizes it into consumption, and turns it into cost dashboards. Energy types (electricity, water, heating oil, gas, …) and meters are **user-defined, never hardcoded**.
 
-**Status: implemented (M0–M7) + SDD §8 panels.** The full solution is built and green — five projects, ~108 tests, working Docker deploy. `docs/SDD.md` remains the design reference; the milestone map (§12) matches the git history (M0…M7 commits). The dedicated **PV/Solar** (`/solar`), **Oil/consumable** (`/consumables`) and **meter-detail** (`/meters/{id}`) views (SDD §8.4–§8.6) are implemented as read models in `Infrastructure/Dashboard` (`SolarService`, `ConsumableService`, `MeterDetailService`) — PV meters are found by `Mode == GenerationCounter` and grid/load meters by a `role` tag in `Meter.Meta` (`MeterRoles`/`MeterMeta`), so nothing is hardcoded by name. **Admin write-CRUD** (SDD §8.7) is implemented as MudBlazor inline-dialog pages: energy types, meters (+ recompute on mode/baseline change), a meter's ingest sources (meter-detail Sources tab), tariffs, cost categories + members, and connectors (`ingestion_endpoint`, secrets by env-var reference only). `/admin/settings` is a read-only effective-config view (settings are env-driven and reproducible, not DB-stored). **Home Assistant reading** is configured here: an HA connector (`BaseUrl` + `TokenEnv`) + an HA source (entity id) drives `HomeAssistantWorker`'s REST poll; `HaConnectionTester` powers the connector "Test connection" button. **Meter topology & flow**: `MeterLink` (a directed `from→to` edge; a downstream meter is a *subsection* of an upstream one, multi-parent allowed) drives a per-energy-type page `/energy/{id}` with a hand-rolled SVG **Sankey** (`SankeyChart.razor`, since ApexCharts has no Sankey type) computed by `FlowService` (link value = downstream consumption, split proportionally across multiple parents; unaccounted remainder → an "Other" node). Upstream meters are wired cycle-safely in the meter editor; the nav lists a link per energy type. Remaining refinements (HA WebSocket *push* — REST poll works today; commit-arbitrary-CSV-from-UI needs a meter-mapping wizard; full de-DE UI-string localization) are noted at their commits. Set `MeterVault__SeedReferenceData=true` (compose: `METERVAULT_SEED=true`) for a one-command populated demo.
+**Status: implemented (M0–M7) + SDD §8 panels.** The full solution is built and green — five projects, ~108 tests, working Docker deploy. `docs/SDD.md` remains the design reference; the milestone map (§12) matches the git history (M0…M7 commits). The dedicated **PV/Solar** (`/solar`), **Oil/consumable** (`/consumables`) and **meter-detail** (`/meters/{id}`) views (SDD §8.4–§8.6) are implemented as read models in `Infrastructure/Dashboard` (`SolarService`, `ConsumableService`, `MeterDetailService`) — PV meters are found by `Mode == GenerationCounter` and grid/load meters by a `role` tag in `Meter.Meta` (`MeterRoles`/`MeterMeta`), so nothing is hardcoded by name. **Admin write-CRUD** (SDD §8.7) is implemented as MudBlazor inline-dialog pages: energy types, meters (+ recompute on mode/baseline change), a meter's ingest sources (meter-detail Sources tab), tariffs, cost categories + members, and connectors (`ingestion_endpoint`, secrets by env-var reference only). `/admin/settings` is a read-only effective-config view (settings are env-driven and reproducible, not DB-stored). **Home Assistant reading** is configured here: an HA connector (`BaseUrl` + `TokenEnv`) + an HA source (entity id) drives `HomeAssistantWorker`'s REST poll, or — with the connector's **WebSocket push** toggle (`HaEndpointConfig.UseWebSocket`) — `HomeAssistantWebSocketWorker` holds a persistent `state_changed` subscription and ingests in real time (the poll worker skips WS endpoints, so each is served once; `HaWebSocketProtocol` is the pure, unit-tested handshake/parse logic). `HaConnectionTester` powers the connector "Test connection" button. **Meter topology & flow**: `MeterLink` (a directed `from→to` edge; a downstream meter is a *subsection* of an upstream one, multi-parent allowed) drives a per-energy-type page `/energy/{id}` with a hand-rolled SVG **Sankey** (`SankeyChart.razor`, since ApexCharts has no Sankey type) computed by `FlowService` (link value = downstream consumption, split proportionally across multiple parents; unaccounted remainder → an "Other" node). Upstream meters are wired cycle-safely in the meter editor; the nav lists a link per energy type. **CSV mapping wizard** (`/import/wizard`): upload an arbitrary CSV, map columns → meters/roles, dry-run preview, then commit as a revertible `import_batch` (the `/import` page lists batches with one-click revert). The `instant_rate` mode is normalized (`InstantRateNormalizer`: rate integrated over time, trapezoidal). Remaining refinement: full de-DE UI-string localization (data parsing is already de-DE) — noted at its commit. Set `MeterVault__SeedReferenceData=true` (compose: `METERVAULT_SEED=true`) for a one-command populated demo.
 
 ## Source of truth
 

README.md

@@ -41,9 +41,9 @@ full design.
 
 ```bash
 docker compose -f deploy/docker-compose.yml up -d
-# open http://localhost:8080  → Import → "Load reference data" for a populated demo
+# open http://localhost:8760  → Import → "Load reference data" for a populated demo
 # ...or start pre-populated:  METERVAULT_SEED=true docker compose -f deploy/docker-compose.yml up -d
-# API docs at http://localhost:8080/swagger
+# API docs at http://localhost:8760/swagger
 ```
 
 Configuration is via environment variables (`Section__Key` double-underscore mapping), e.g.:
@@ -67,7 +67,7 @@ of an environment variable, resolved at runtime.
 ## Pushing readings (Home Assistant)
 
 ```bash
-curl -X POST http://localhost:8080/api/v1/readings \
+curl -X POST http://localhost:8760/api/v1/readings \
   -H "X-Api-Key: $METERVAULT_API_KEY" -H "Content-Type: application/json" \
   -d '[{"meterId": 1, "time": "2026-01-01T12:00:00Z", "value": 47200}]'
 ```

deploy/Dockerfile

@@ -25,8 +25,8 @@ RUN apt-get update \
     && apt-get install -y --no-install-recommends curl \
     && rm -rf /var/lib/apt/lists/*
 COPY --from=build /app/publish ./
-ENV ASPNETCORE_URLS=http://+:8080 \
+ENV ASPNETCORE_URLS=http://+:8760 \
     ASPNETCORE_ENVIRONMENT=Production \
     DOTNET_SYSTEM_GLOBALIZATION_INVARIANT=false
-EXPOSE 8080
+EXPOSE 8760
 ENTRYPOINT ["dotnet", "MeterVault.App.dll"]

deploy/docker-compose.yml

@@ -40,9 +40,9 @@ services:
       #   MeterVault__ApiKeys__0: your-secret-key
       #   MeterVault__AllowAnonymousApi: "true"
     ports:
-      - "${METERVAULT_PORT:-8080}:8080"
+      - "${METERVAULT_PORT:-8760}:8760"
     healthcheck:
-      test: ["CMD-SHELL", "curl -fsS http://localhost:8080/healthz || exit 1"]
+      test: ["CMD-SHELL", "curl -fsS http://localhost:8760/healthz || exit 1"]
       interval: 15s
       timeout: 5s
       retries: 5

deploy/unraid-template.xml

@@ -11,10 +11,10 @@
   <Privileged>false</Privileged>
   <Overview>Self-hosted energy &amp; utility metering: ingest from Home Assistant/Tasmota/MQTT, normalize to consumption, and produce cost dashboards. Needs a TimescaleDB instance.</Overview>
   <Category>HomeAutomation: Tools: Productivity:</Category>
-  <WebUI>http://[IP]:[PORT:8080]/</WebUI>
+  <WebUI>http://[IP]:[PORT:8760]/</WebUI>
   <Icon>https://git.finalfactory.de/FinalFactory/MeterVault/raw/branch/master/docs/icon.png</Icon>
 
-  <Config Name="WebUI Port" Target="8080" Default="8080" Mode="tcp" Description="HTTP port" Type="Port" Display="always" Required="true">8080</Config>
+  <Config Name="WebUI Port" Target="8760" Default="8760" Mode="tcp" Description="HTTP port" Type="Port" Display="always" Required="true">8760</Config>
 
   <Config Name="Database connection" Target="ConnectionStrings__Default" Default="Host=timescaledb;Port=5432;Database=metervault;Username=metervault;Password=changeme" Mode="" Description="PostgreSQL/TimescaleDB connection string" Type="Variable" Display="always" Required="true">Host=timescaledb;Port=5432;Database=metervault;Username=metervault;Password=changeme</Config>
 

src/App/Components/Pages/Admin/Connectors.razor

@@ -71,6 +71,10 @@ else
         {
             <MudTextField @bind-Value="_working.BaseUrl" Label="Base URL (e.g. http://homeassistant.local:8123)" Class="mb-2" />
             <MudTextField @bind-Value="_working.TokenEnv" Label="Token env-var name (e.g. HA_TOKEN)" Class="mb-2" />
+            <MudSwitch T="bool" @bind-Value="_working.UseWebSocket" Label="Real-time WebSocket push" Color="Color.Primary" Class="mb-1" />
+            <MudText Typo="Typo.caption" Color="Color.Secondary" Class="mb-2 d-block">
+                On: subscribe to state changes and ingest in real time. Off: REST poll on each source's interval.
+            </MudText>
             <MudTextField @bind-Value="_working.TestEntityId" Label="Test entity id (optional, e.g. sensor.house_power)" Class="mb-2" />
             <MudButton Variant="Variant.Outlined" StartIcon="@Icons.Material.Filled.NetworkCheck" OnClick="TestHaAsync" Disabled="_testing" Class="mb-2">
                 Test connection
@@ -130,7 +134,7 @@ else
             _working = new EditModel
             {
                 Id = endpoint.Id, Type = endpoint.Type, Name = endpoint.Name, IsEnabled = endpoint.IsEnabled,
-                BaseUrl = ha.BaseUrl, TokenEnv = ha.TokenEnv,
+                BaseUrl = ha.BaseUrl, TokenEnv = ha.TokenEnv, UseWebSocket = ha.UseWebSocket,
             };
         }
         else
@@ -170,7 +174,7 @@ else
         }
 
         var config = _working.Type == EndpointType.HomeAssistant
-            ? new HaEndpointConfig { BaseUrl = Trim(_working.BaseUrl), TokenEnv = Trim(_working.TokenEnv) }.ToJson()
+            ? new HaEndpointConfig { BaseUrl = Trim(_working.BaseUrl), TokenEnv = Trim(_working.TokenEnv), UseWebSocket = _working.UseWebSocket }.ToJson()
             : new EndpointConfig
             {
                 Host = string.IsNullOrWhiteSpace(_working.Host) ? "localhost" : _working.Host.Trim(),
@@ -234,6 +238,7 @@ else
         // Home Assistant
         public string? BaseUrl { get; set; }
         public string? TokenEnv { get; set; }
+        public bool UseWebSocket { get; set; }
         public string? TestEntityId { get; set; }
 
         // MQTT broker

src/App/Components/Pages/Import.razor

@@ -1,8 +1,13 @@
 @page "/import"
 @inject MeterVault.Infrastructure.Import.ReferenceDataImporter ReferenceImporter
 @inject MeterVault.Infrastructure.Import.CsvImporter CsvImporter
+@inject MeterVault.Infrastructure.Import.ImportService ImportService
+@inject Microsoft.EntityFrameworkCore.IDbContextFactory<MeterVault.Infrastructure.Persistence.MeterVaultDbContext> DbFactory
+@inject IDialogService Dialogs
 @inject ISnackbar Snackbar
 @using MeterVault.Infrastructure.Import
+@using MeterVault.Infrastructure.Persistence
+@using Microsoft.EntityFrameworkCore
 
 <PageTitle>MeterVault — Import</PageTitle>
 
@@ -29,18 +34,23 @@
 
     <MudItem xs="12" md="6">
         <MudPaper Class="pa-4" Elevation="2" Style="height:100%">
-            <MudText Typo="Typo.h6">Dry-run a CSV</MudText>
-            <MudText Typo="Typo.body2" Color="Color.Secondary" Class="mb-3">
-                Upload a sheet and preview what would be staged (no changes are made).
+            <div class="d-flex align-center justify-space-between">
+                <MudText Typo="Typo.h6">Your own CSV</MudText>
+                <MudButton Variant="Variant.Filled" Color="Color.Secondary" Href="/import/wizard"
+                           StartIcon="@Icons.Material.Filled.AutoFixHigh">Mapping wizard</MudButton>
+            </div>
+            <MudText Typo="Typo.body2" Color="Color.Secondary" Class="mb-3 mt-1">
+                Map an arbitrary sheet's columns to your meters/categories, preview and commit it as a
+                revertible import. Or dry-run against one of the built-in reference profiles below.
             </MudText>
-            <MudSelect T="string" @bind-Value="_profileName" Label="Sheet type" Dense="true" Class="mb-2">
+            <MudSelect T="string" @bind-Value="_profileName" Label="Reference profile" Dense="true" Class="mb-2">
                 <MudSelectItem T="string" Value="@("Strom")">Electricity (Strom)</MudSelectItem>
                 <MudSelectItem T="string" Value="@("Wasser")">Water (Wasser)</MudSelectItem>
                 <MudSelectItem T="string" Value="@("Heizöl")">Heating oil (Heizöl)</MudSelectItem>
                 <MudSelectItem T="string" Value="@("Kosten")">Costs (Kosten)</MudSelectItem>
             </MudSelect>
             <MudButton HtmlTag="label" Variant="Variant.Outlined" StartIcon="@Icons.Material.Filled.UploadFile" for="csvUpload">
-                Choose CSV
+                Dry-run a reference sheet
             </MudButton>
             <InputFile id="csvUpload" OnChange="PreviewAsync" accept=".csv" style="display:none" />
         </MudPaper>
@@ -71,6 +81,49 @@
             </MudPaper>
         </MudItem>
     }
+
+    <MudItem xs="12">
+        <MudPaper Class="pa-4" Elevation="2">
+            <MudText Typo="Typo.h6" Class="mb-2">Recent imports</MudText>
+            @if (_batches.Count == 0)
+            {
+                <MudText Typo="Typo.body2" Color="Color.Secondary">No imports yet.</MudText>
+            }
+            else
+            {
+                <MudSimpleTable Dense="true" Hover="true">
+                    <thead><tr><th>#</th><th>Source</th><th style="text-align:right">Rows</th><th>Imported</th><th>Status</th><th></th></tr></thead>
+                    <tbody>
+                        @foreach (var batch in _batches)
+                        {
+                            <tr>
+                                <td>@batch.Id</td>
+                                <td>@(batch.SourceName ?? "—")</td>
+                                <td style="text-align:right">@batch.RowCount</td>
+                                <td>@batch.CreatedAt.LocalDateTime.ToString("yyyy-MM-dd HH:mm")</td>
+                                <td>
+                                    @if (batch.RevertedAt is not null)
+                                    {
+                                        <MudChip T="string" Size="Size.Small" Color="Color.Default">reverted</MudChip>
+                                    }
+                                    else
+                                    {
+                                        <MudChip T="string" Size="Size.Small" Color="Color.Success">active</MudChip>
+                                    }
+                                </td>
+                                <td style="text-align:right">
+                                    <MudButton Size="Size.Small" Variant="Variant.Outlined" Color="Color.Error"
+                                               StartIcon="@Icons.Material.Filled.Undo"
+                                               Disabled="@(batch.RevertedAt is not null || _reverting == batch.Id)"
+                                               OnClick="@(() => RevertAsync(batch))">Revert</MudButton>
+                                </td>
+                            </tr>
+                        }
+                    </tbody>
+                </MudSimpleTable>
+            }
+        </MudPaper>
+    </MudItem>
 </MudGrid>
 
 @code {
@@ -78,9 +131,49 @@
     private bool _referenceLoaded;
     private string _profileName = "Strom";
     private StagedImport? _preview;
+    private List<ImportBatch> _batches = [];
+    private int? _reverting;
 
-    protected override async Task OnInitializedAsync() =>
+    protected override async Task OnInitializedAsync()
+    {
         _referenceLoaded = await ReferenceImporter.IsLoadedAsync();
+        await LoadBatchesAsync();
+    }
+
+    private async Task LoadBatchesAsync()
+    {
+        await using var db = await DbFactory.CreateDbContextAsync();
+        _batches = await db.ImportBatches.AsNoTracking()
+            .OrderByDescending(b => b.Id)
+            .Take(25)
+            .ToListAsync();
+    }
+
+    private async Task RevertAsync(ImportBatch batch)
+    {
+        if (!await Confirm.ConfirmAsync(Dialogs, "Revert import?",
+                $"Delete all {batch.RowCount} rows from import #{batch.Id} ({batch.SourceName ?? "unnamed"}) and recompute the affected meters?",
+                "Revert"))
+        {
+            return;
+        }
+
+        _reverting = batch.Id;
+        try
+        {
+            await ImportService.RevertAsync(batch.Id);
+            Snackbar.Add($"Import #{batch.Id} reverted.", Severity.Success);
+            await LoadBatchesAsync();
+        }
+        catch (Exception ex)
+        {
+            Snackbar.Add($"Revert failed: {ex.Message}", Severity.Error);
+        }
+        finally
+        {
+            _reverting = null;
+        }
+    }
 
     private async Task LoadReferenceAsync()
     {
@@ -91,6 +184,7 @@
             await ReferenceImporter.LoadAsync(dir);
             _referenceLoaded = true;
             Snackbar.Add("Reference data loaded.", Severity.Success);
+            await LoadBatchesAsync();
         }
         catch (Exception ex)
         {

src/App/Components/Pages/ImportWizard.razor

@@ -0,0 +1,393 @@
+@page "/import/wizard"
+@inject MeterVault.Infrastructure.Import.CsvImporter CsvImporter
+@inject MeterVault.Infrastructure.Import.ImportService ImportService
+@inject Microsoft.EntityFrameworkCore.IDbContextFactory<MeterVault.Infrastructure.Persistence.MeterVaultDbContext> DbFactory
+@inject ISnackbar Snackbar
+@inject NavigationManager Nav
+@using System.Text.Json
+@using MeterVault.Core.Domain
+@using MeterVault.Infrastructure.Import
+@using MeterVault.Infrastructure.Persistence
+@using Microsoft.EntityFrameworkCore
+
+<PageTitle>MeterVault — Import wizard</PageTitle>
+
+<div class="d-flex align-center mb-4" style="gap:1rem">
+    <MudIconButton Icon="@Icons.Material.Filled.ArrowBack" Href="/import" aria-label="Back to Import" />
+    <MudText Typo="Typo.h4">Import wizard</MudText>
+</div>
+
+<MudText Typo="Typo.body2" Color="Color.Secondary" Class="mb-4">
+    Upload any CSV, map its columns to your meters and categories, preview what would be staged, then
+    commit it as a revertible import. Values may use the German dialect (decimal comma, unit suffixes,
+    <code>Monat JJJJ</code> or <code>TT.MM.JJJJ</code> dates) — the same parser the reference sheets use.
+</MudText>
+
+<MudPaper Class="pa-4 mb-4" Elevation="2">
+    <div class="d-flex align-center" style="gap:1rem; flex-wrap:wrap">
+        <MudButton HtmlTag="label" Variant="Variant.Filled" Color="Color.Primary"
+                   StartIcon="@Icons.Material.Filled.UploadFile" for="wizardUpload">
+            Choose CSV
+        </MudButton>
+        <InputFile id="wizardUpload" OnChange="OnFileAsync" accept=".csv" style="display:none" />
+        @if (_fileName is not null)
+        {
+            <MudText><b>@_fileName</b> — @_rows.Count rows, @_colCount columns</MudText>
+        }
+    </div>
+</MudPaper>
+
+@if (_colCount > 0)
+{
+    <MudPaper Class="pa-4 mb-4" Elevation="2">
+        <MudText Typo="Typo.h6" Class="mb-2">1. Parsing options</MudText>
+        <MudGrid>
+            <MudItem xs="12" sm="6" md="3">
+                <MudSelect T="int" @bind-Value="_dateColumn" Label="Date column" Dense="true">
+                    @foreach (var i in Enumerable.Range(0, _colCount))
+                    {
+                        <MudSelectItem T="int" Value="i">@ColLabel(i)</MudSelectItem>
+                    }
+                </MudSelect>
+            </MudItem>
+            <MudItem xs="12" sm="6" md="3">
+                <MudSelect T="DateKind" @bind-Value="_dateKind" Label="Date format" Dense="true">
+                    <MudSelectItem T="DateKind" Value="DateKind.Auto">Auto-detect</MudSelectItem>
+                    <MudSelectItem T="DateKind" Value="DateKind.MonthName">Month name (Januar 2024)</MudSelectItem>
+                    <MudSelectItem T="DateKind" Value="DateKind.DayDotMonthYear">Day (31.12.2024)</MudSelectItem>
+                </MudSelect>
+            </MudItem>
+            <MudItem xs="6" sm="6" md="2">
+                <MudNumericField T="int" @bind-Value="_headerRow" Label="Header row" Min="0" Margin="Margin.Dense" />
+            </MudItem>
+            <MudItem xs="6" sm="6" md="2">
+                <MudNumericField T="int" @bind-Value="_firstDataRow" Label="First data row" Min="0" Margin="Margin.Dense" />
+            </MudItem>
+            <MudItem xs="12" md="2" Class="d-flex flex-column">
+                <MudSwitch T="bool" @bind-Value="_skipAllZero" Label="Skip zero rows" Color="Color.Primary" />
+                <MudSwitch T="bool" @bind-Value="_detectSwaps" Label="Detect swaps" Color="Color.Primary" />
+            </MudItem>
+        </MudGrid>
+    </MudPaper>
+
+    <MudPaper Class="pa-4 mb-4" Elevation="2">
+        <MudText Typo="Typo.h6" Class="mb-2">2. Column preview</MudText>
+        <div style="overflow-x:auto">
+            <MudSimpleTable Dense="true" Bordered="true" Style="min-width:100%">
+                <thead>
+                    <tr>
+                        @foreach (var i in Enumerable.Range(0, _colCount))
+                        {
+                            <th style="@(i == _dateColumn ? "background:var(--mud-palette-primary-hover)" : "")">
+                                Col @i@(i == _dateColumn ? " 📅" : "")
+                            </th>
+                        }
+                    </tr>
+                </thead>
+                <tbody>
+                    @foreach (var r in Enumerable.Range(0, Math.Min(_rows.Count, _firstDataRow + 8)))
+                    {
+                        <tr style="@(r < _firstDataRow ? "opacity:0.5" : "")">
+                            @foreach (var i in Enumerable.Range(0, _colCount))
+                            {
+                                <td style="font-size:0.78rem; white-space:nowrap">@Cell(r, i)</td>
+                            }
+                        </tr>
+                    }
+                </tbody>
+            </MudSimpleTable>
+        </div>
+        <MudText Typo="Typo.caption" Color="Color.Secondary">
+            Faded rows are before the first data row. The 📅 column supplies the date.
+        </MudText>
+    </MudPaper>
+
+    <MudPaper Class="pa-4 mb-4" Elevation="2">
+        <MudText Typo="Typo.h6" Class="mb-2">3. Map columns</MudText>
+        <div style="overflow-x:auto">
+            <MudSimpleTable Dense="true" Style="min-width:100%">
+                <thead>
+                    <tr><th>Column</th><th>Sample</th><th style="min-width:160px">Role</th><th style="min-width:220px">Target</th><th style="min-width:120px">Unit</th></tr>
+                </thead>
+                <tbody>
+                    @foreach (var i in Enumerable.Range(0, _colCount))
+                    {
+                        <tr>
+                            <td>
+                                <b>Col @i</b>
+                                @if (!string.IsNullOrWhiteSpace(Header(i)))
+                                {
+                                    <br /><span style="font-size:0.72rem; opacity:0.7">@Header(i)</span>
+                                }
+                            </td>
+                            <td style="font-size:0.8rem; max-width:140px; overflow:hidden; text-overflow:ellipsis">@Sample(i)</td>
+                            <td>
+                                <MudSelect T="MappingRole" @bind-Value="_columns[i].Role" Dense="true" Margin="Margin.Dense">
+                                    @foreach (var role in Enum.GetValues<MappingRole>())
+                                    {
+                                        <MudSelectItem T="MappingRole" Value="role">@role</MudSelectItem>
+                                    }
+                                </MudSelect>
+                            </td>
+                            <td>
+                                @if (_columns[i].Role is MappingRole.Reading or MappingRole.Delivery or MappingRole.TankLevel)
+                                {
+                                    <MudSelect T="int?" @bind-Value="_columns[i].MeterId" Dense="true" Margin="Margin.Dense"
+                                               Placeholder="Select meter" Clearable="true">
+                                        @foreach (var m in _meters)
+                                        {
+                                            <MudSelectItem T="int?" Value="m.Id">@m.Name (@m.Unit)</MudSelectItem>
+                                        }
+                                    </MudSelect>
+                                }
+                                else if (_columns[i].Role == MappingRole.ManualCost)
+                                {
+                                    <MudSelect T="int?" @bind-Value="_columns[i].CategoryId" Dense="true" Margin="Margin.Dense"
+                                               Placeholder="Select category" Clearable="true">
+                                        @foreach (var c in _categories)
+                                        {
+                                            <MudSelectItem T="int?" Value="c.Id">@c.Name</MudSelectItem>
+                                        }
+                                    </MudSelect>
+                                }
+                            </td>
+                            <td>
+                                @if (_columns[i].Role is MappingRole.Reading or MappingRole.Delivery or MappingRole.TankLevel)
+                                {
+                                    <MudTextField @bind-Value="_columns[i].Unit" Placeholder="e.g. kWh" Margin="Margin.Dense" />
+                                }
+                            </td>
+                        </tr>
+                    }
+                </tbody>
+            </MudSimpleTable>
+        </div>
+    </MudPaper>
+
+    @if (_validationErrors.Count > 0)
+    {
+        <MudAlert Severity="Severity.Warning" Class="mb-4">
+            @foreach (var err in _validationErrors)
+            {
+                <div>@err</div>
+            }
+        </MudAlert>
+    }
+
+    <MudPaper Class="pa-4 mb-4" Elevation="2">
+        <div class="d-flex align-center mb-2" style="gap:1rem; flex-wrap:wrap">
+            <MudText Typo="Typo.h6">4. Preview &amp; commit</MudText>
+            <MudButton Variant="Variant.Outlined" Color="Color.Primary" StartIcon="@Icons.Material.Filled.Visibility"
+                       OnClick="Preview">Dry-run preview</MudButton>
+            <MudButton Variant="Variant.Filled" Color="Color.Success" StartIcon="@Icons.Material.Filled.Save"
+                       OnClick="CommitAsync" Disabled="_staged is null || _staged.TotalRows == 0 || _committing">
+                Commit import
+            </MudButton>
+            @if (_committing)
+            {
+                <MudProgressCircular Size="Size.Small" Indeterminate="true" Color="Color.Primary" />
+            }
+        </div>
+
+        @if (_staged is not null)
+        {
+            <div class="d-flex mt-2" style="gap:2rem; flex-wrap:wrap">
+                <MudText>Readings: <b>@_staged.Readings.Count</b></MudText>
+                <MudText>Events: <b>@_staged.Events.Count</b></MudText>
+                <MudText>Manual costs: <b>@_staged.ManualCosts.Count</b></MudText>
+                <MudText>Skipped rows: <b>@_staged.SkippedRows</b></MudText>
+            </div>
+            @if (_staged.TotalRows == 0)
+            {
+                <MudAlert Severity="Severity.Info" Dense="true" Class="mt-2">
+                    Nothing staged. Check the first-data-row, date column and column mappings above.
+                </MudAlert>
+            }
+            @if (_staged.Warnings.Count > 0)
+            {
+                <MudExpansionPanels Class="mt-3">
+                    <MudExpansionPanel Text="@($"{_staged.Warnings.Count} warnings")">
+                        @foreach (var warning in _staged.Warnings.Take(100))
+                        {
+                            <MudText Typo="Typo.body2">@warning</MudText>
+                        }
+                    </MudExpansionPanel>
+                </MudExpansionPanels>
+            }
+        }
+    </MudPaper>
+}
+
+@code {
+    private sealed class ColumnState
+    {
+        public MappingRole Role { get; set; } = MappingRole.Ignore;
+        public int? MeterId { get; set; }
+        public int? CategoryId { get; set; }
+        public string? Unit { get; set; }
+    }
+
+    private string? _fileName;
+    private string? _csvText;
+    private List<string[]> _rows = [];
+    private int _colCount;
+    private ColumnState[] _columns = [];
+
+    private int _dateColumn;
+    private DateKind _dateKind = DateKind.Auto;
+    private int _headerRow;
+    private int _firstDataRow = 1;
+    private bool _skipAllZero = true;
+    private bool _detectSwaps;
+
+    private List<Meter> _meters = [];
+    private List<CostCategory> _categories = [];
+    private StagedImport? _staged;
+    private List<string> _validationErrors = [];
+    private bool _committing;
+
+    private async Task OnFileAsync(InputFileChangeEventArgs args)
+    {
+        var file = args.File;
+        if (file is null)
+        {
+            return;
+        }
+
+        _fileName = file.Name;
+        using (var reader = new StreamReader(file.OpenReadStream(maxAllowedSize: 10 * 1024 * 1024)))
+        {
+            _csvText = await reader.ReadToEndAsync();
+        }
+
+        _rows = CsvImporter.ReadRawRows(new StringReader(_csvText)).ToList();
+        _colCount = _rows.Count == 0 ? 0 : _rows.Max(r => r.Length);
+        _columns = Enumerable.Range(0, _colCount).Select(_ => new ColumnState()).ToArray();
+        _dateColumn = 0;
+        _staged = null;
+        _validationErrors = [];
+
+        await using var db = await DbFactory.CreateDbContextAsync();
+        _meters = await db.Meters.AsNoTracking().OrderBy(m => m.Name).ToListAsync();
+        _categories = await db.CostCategories.AsNoTracking().OrderBy(c => c.Sort).ThenBy(c => c.Name).ToListAsync();
+    }
+
+    private void Preview()
+    {
+        _validationErrors = Validate();
+        if (_validationErrors.Count > 0)
+        {
+            _staged = null;
+            return;
+        }
+
+        using var reader = new StringReader(_csvText ?? string.Empty);
+        _staged = CsvImporter.Stage(BuildProfile(), reader);
+    }
+
+    private async Task CommitAsync()
+    {
+        if (_staged is null || _staged.TotalRows == 0)
+        {
+            return;
+        }
+
+        _committing = true;
+        try
+        {
+            var mappingJson = JsonSerializer.Serialize(new
+            {
+                dateColumn = _dateColumn,
+                dateKind = _dateKind.ToString(),
+                firstDataRow = _firstDataRow,
+                columns = _columns
+                    .Select((c, i) => new { index = i, role = c.Role.ToString(), c.MeterId, c.CategoryId, c.Unit })
+                    .Where(c => c.role != nameof(MappingRole.Ignore)),
+            });
+
+            var batchId = await ImportService.CommitAsync(_staged, _fileName, mappingJson);
+            Snackbar.Add($"Imported batch #{batchId}: {_staged.TotalRows} rows staged. Consumption recomputed.", Severity.Success);
+            Nav.NavigateTo("/import");
+        }
+        catch (Exception ex)
+        {
+            Snackbar.Add($"Commit failed: {ex.Message}", Severity.Error);
+        }
+        finally
+        {
+            _committing = false;
+        }
+    }
+
+    private List<string> Validate()
+    {
+        var errors = new List<string>();
+        if (_columns.Count(c => c.Role != MappingRole.Ignore) == 0)
+        {
+            errors.Add("Map at least one column to a role other than Ignore.");
+        }
+
+        for (var i = 0; i < _columns.Length; i++)
+        {
+            var c = _columns[i];
+            if (c.Role is MappingRole.Reading or MappingRole.Delivery or MappingRole.TankLevel && c.MeterId is null)
+            {
+                errors.Add($"Col {i} ({c.Role}) needs a target meter.");
+            }
+
+            if (c.Role == MappingRole.ManualCost && c.CategoryId is null)
+            {
+                errors.Add($"Col {i} (ManualCost) needs a target category.");
+            }
+        }
+
+        return errors;
+    }
+
+    private MappingProfile BuildProfile() => new()
+    {
+        Name = _fileName ?? "Custom CSV",
+        HeaderRowIndex = _headerRow,
+        FirstDataRowIndex = _firstDataRow,
+        DateColumn = _dateColumn,
+        DateKind = _dateKind,
+        SkipAllZeroRows = _skipAllZero,
+        DetectCumulativeSwaps = _detectSwaps,
+        Columns = Enumerable.Range(0, _colCount)
+            .Where(i => _columns[i].Role != MappingRole.Ignore)
+            .Select(i => new ColumnMapping
+            {
+                Index = i,
+                Role = _columns[i].Role,
+                MeterId = _columns[i].MeterId,
+                CategoryId = _columns[i].CategoryId,
+                Unit = string.IsNullOrWhiteSpace(_columns[i].Unit) ? null : _columns[i].Unit,
+            })
+            .ToList(),
+    };
+
+    private string Cell(int row, int col) =>
+        row >= 0 && row < _rows.Count && col < _rows[row].Length ? _rows[row][col] : string.Empty;
+
+    private string Header(int col) => Cell(_headerRow, col);
+
+    private string Sample(int col)
+    {
+        for (var r = _firstDataRow; r < _rows.Count && r < _firstDataRow + 20; r++)
+        {
+            var value = Cell(r, col);
+            if (!string.IsNullOrWhiteSpace(value))
+            {
+                return value;
+            }
+        }
+
+        return string.Empty;
+    }
+
+    private string ColLabel(int col)
+    {
+        var header = Header(col);
+        return string.IsNullOrWhiteSpace(header) ? $"Col {col}" : $"Col {col}: {header}";
+    }
+}

src/App/Components/Pages/Meters.razor

@@ -85,6 +85,14 @@ else
                 upstream meters you select below (e.g. Sum Solar = Solar 1 + Solar 2).
             </MudAlert>
         }
+        else if (_working.Mode == MeterMode.InstantRate)
+        {
+            <MudAlert Severity="Severity.Info" Dense="true" Class="mb-2">
+                Power/flow sensor — readings are an instantaneous rate, integrated over time into consumption.
+                Store the value as a <b>per-hour</b> rate in this meter's unit (e.g. kW for kWh, L/h for L): a
+                source reporting W or L/min should carry a scale factor to convert it first.
+            </MudAlert>
+        }
         <MudTextField @bind-Value="_working.Unit" Label="Unit" Required="true" Class="mb-2" />
         <MudNumericField T="double" @bind-Value="_working.InitialBaseline" Label="Initial register baseline" Class="mb-2" />
         <MudSelect T="string" @bind-Value="_working.Role" Label="PV role (optional)" Class="mb-2">

src/App/Confirm.cs

@@ -11,4 +11,12 @@ public static class Confirm
             .ConfigureAwait(false);
         return result == true;
     }
+
+    /// <summary>A generic yes/cancel confirmation with a caller-supplied confirm-button label.</summary>
+    public static async Task<bool> ConfirmAsync(IDialogService dialog, string title, string message, string confirmText)
+    {
+        var result = await dialog.ShowMessageBoxAsync(title, message, yesText: confirmText, cancelText: "Cancel")
+            .ConfigureAwait(false);
+        return result == true;
+    }
 }

src/App/Program.cs

@@ -60,7 +60,7 @@ try
     app.UseStatusCodePagesWithReExecute("/not-found", createScopeForStatusCodePages: true);
     app.UseSerilogRequestLogging();
     app.UseReverseProxyTrust();
-    // No HTTPS redirection: the app serves plain HTTP (port 8080) behind a reverse proxy
+    // No HTTPS redirection: the app serves plain HTTP (port 8760) behind a reverse proxy
     // that terminates TLS (SDD §10). HTTPS redirection here would break the container and proxy.
     app.UseAntiforgery();
 

src/Core/Domain/Enums.cs

@@ -21,7 +21,7 @@ public enum MeterMode
     /// <summary>Source already reports increments. The value is the increment.</summary>
     DirectDelta,
 
-    /// <summary>Power/flow sensor (schema-supported; worker deferred post-v1). Integrate rate over time.</summary>
+    /// <summary>Power/flow sensor. Consumption = rate integrated over time (trapezoidal).</summary>
     InstantRate,
 
     /// <summary>Computed from other meters via a user-defined expression.</summary>

src/Core/Normalization/NormalizationEngine.cs

@@ -24,6 +24,7 @@ public sealed class NormalizationEngine : INormalizationEngine
         new Normalizers.RuntimeCounterNormalizer(),
         new Normalizers.ConsumableBalanceNormalizer(),
         new Normalizers.DirectDeltaNormalizer(),
+        new Normalizers.InstantRateNormalizer(),
         new Normalizers.VirtualNormalizer(),
     ]);
 

src/Core/Normalization/Normalizers/InstantRateNormalizer.cs

@@ -0,0 +1,55 @@
+using MeterVault.Core.Domain;
+
+namespace MeterVault.Core.Normalization.Normalizers;
+
+/// <summary>
+/// Power/flow sensor (SDD §5.2 <c>instant_rate</c>): the reading value is an instantaneous rate, not
+/// a register. Consumption is the rate integrated over time (trapezoidal rule between consecutive
+/// samples), attributed to the interval's end reading — so the first reading only seeds the integral
+/// and N readings produce N−1 consumption rows.
+/// <para>
+/// The value is treated as a rate expressed <em>per hour</em> in the meter's own unit, i.e.
+/// <c>rate × Δhours = consumption</c>. Power in kW integrated over hours yields kWh; a flow in L/h
+/// yields L; m³/h yields m³. A source that reports a native unit (W, L/min) should carry a
+/// scale/offset that converts it to this canonical per-hour rate before it is stored as a reading.
+/// Signs are preserved (a bidirectional power sensor may go negative on export).
+/// </para>
+/// </summary>
+public sealed class InstantRateNormalizer : IMeterNormalizer
+{
+    public MeterMode Mode => MeterMode.InstantRate;
+
+    public IEnumerable<Consumption> Normalize(NormalizationContext context)
+    {
+        ArgumentNullException.ThrowIfNull(context);
+
+        var readings = context.Readings.OrderBy(r => r.Time).ToList();
+
+        Reading? previous = null;
+        foreach (var reading in readings)
+        {
+            if (previous is not null)
+            {
+                var hours = (reading.Time - previous.Time).TotalHours;
+                if (hours > 0)
+                {
+                    // Trapezoidal integral of the rate over [previous, reading]; the linear mean of
+                    // the two samples is exact for a rate that varies linearly between them.
+                    var amount = (previous.Value + reading.Value) / 2d * hours;
+
+                    yield return new Consumption
+                    {
+                        MeterId = context.Meter.MeterId,
+                        Time = reading.Time,
+                        Amount = amount,
+                        Kind = ConsumptionKind.Consumption,
+                        Quality = reading.Quality,
+                        ImportBatchId = reading.ImportBatchId,
+                    };
+                }
+            }
+
+            previous = reading;
+        }
+    }
+}

src/Infrastructure/DependencyInjection.cs

@@ -57,6 +57,7 @@ public static class DependencyInjection
         services.AddHttpClient();
         services.AddHostedService<MqttIngestionWorker>();
         services.AddHostedService<HomeAssistantWorker>();
+        services.AddHostedService<HomeAssistantWebSocketWorker>();
         return services;
     }
 }

src/Infrastructure/Import/CsvImporter.cs

@@ -222,6 +222,12 @@ public sealed class CsvImporter
         return true;
     }
 
+    /// <summary>
+    /// Reads the raw CSV grid (no profile applied) so the import wizard can show a column preview
+    /// and let the user map columns before staging. Rows are ragged (their own field counts).
+    /// </summary>
+    public static IReadOnlyList<string[]> ReadRawRows(TextReader reader) => ReadRows(reader);
+
     internal static List<string[]> ReadRows(TextReader reader)
     {
         var config = new CsvConfiguration(CultureInfo.InvariantCulture)

src/Infrastructure/Ingestion/HaEndpointConfig.cs

@@ -22,6 +22,13 @@ public sealed record HaEndpointConfig
     /// <summary>Name of the environment variable holding the long-lived access token.</summary>
     public string? TokenEnv { get; init; }
 
+    /// <summary>
+    /// When true, a persistent WebSocket subscription pushes state changes in real time
+    /// (<see cref="HomeAssistantWebSocketWorker"/>); when false (default) the REST poll worker
+    /// samples on each source's interval. An endpoint is handled by exactly one of the two.
+    /// </summary>
+    public bool UseWebSocket { get; init; }
+
     public static HaEndpointConfig Parse(string? json)
     {
         if (string.IsNullOrWhiteSpace(json))

src/Infrastructure/Ingestion/HaStateClient.cs

@@ -31,10 +31,16 @@ public sealed class HaStateClient(HttpClient httpClient)
         return ParseState(await JsonDocument.ParseAsync(stream, cancellationToken: cancellationToken).ConfigureAwait(false), attribute);
     }
 
-    internal static HaState? ParseState(JsonDocument document, string? attribute)
-    {
-        var root = document.RootElement;
+    internal static HaState? ParseState(JsonDocument document, string? attribute) =>
+        ParseStateElement(document.RootElement, attribute);
 
+    /// <summary>
+    /// Extracts a numeric value + last-updated time from a Home Assistant state object (the REST
+    /// entity payload, or a <c>new_state</c> from a WebSocket <c>state_changed</c> event — they share
+    /// the same shape). Returns null for <c>unavailable</c>/<c>unknown</c>/non-numeric states.
+    /// </summary>
+    internal static HaState? ParseStateElement(JsonElement root, string? attribute)
+    {
         double value;
         if (string.IsNullOrWhiteSpace(attribute))
         {

src/Infrastructure/Ingestion/HaWebSocketProtocol.cs

@@ -0,0 +1,70 @@
+using System.Text.Json;
+
+namespace MeterVault.Infrastructure.Ingestion;
+
+/// <summary>
+/// Pure helpers for the Home Assistant WebSocket API (<c>/api/websocket</c>): the auth handshake
+/// (<c>auth_required</c> → <c>auth</c> → <c>auth_ok</c>), subscribing to <c>state_changed</c> events,
+/// and reading the entity id + <c>new_state</c> out of an event frame. No I/O — the transport lives
+/// in <see cref="HomeAssistantWebSocketWorker"/>, which keeps this unit-testable.
+/// </summary>
+internal static class HaWebSocketProtocol
+{
+    /// <summary>Derives the WebSocket endpoint from an HTTP base URL (http→ws, https→wss, path <c>/api/websocket</c>).</summary>
+    public static Uri WebSocketUri(string baseUrl)
+    {
+        var http = new Uri(baseUrl.TrimEnd('/') + "/api/websocket", UriKind.Absolute);
+        var scheme = http.Scheme.Equals("https", StringComparison.OrdinalIgnoreCase) ? "wss" : "ws";
+        return new UriBuilder(http) { Scheme = scheme }.Uri;
+    }
+
+    public static string AuthMessage(string token) =>
+        JsonSerializer.Serialize(new { type = "auth", access_token = token });
+
+    public static string SubscribeStateChanged(int id) =>
+        JsonSerializer.Serialize(new { id, type = "subscribe_events", event_type = "state_changed" });
+
+    public static string? MessageType(JsonElement root) =>
+        root.TryGetProperty("type", out var type) && type.ValueKind == JsonValueKind.String ? type.GetString() : null;
+
+    public static bool IsAuthRequired(JsonElement root) => MessageType(root) == "auth_required";
+
+    public static bool IsAuthOk(JsonElement root) => MessageType(root) == "auth_ok";
+
+    public static bool IsAuthInvalid(JsonElement root) => MessageType(root) == "auth_invalid";
+
+    /// <summary>
+    /// If <paramref name="root"/> is a <c>state_changed</c> event with a numeric-capable
+    /// <c>new_state</c> object, yields the entity id and that state object. The caller extracts the
+    /// numeric value (state or a named attribute) per source via
+    /// <see cref="HaStateClient.ParseStateElement"/>.
+    /// </summary>
+    public static bool TryReadStateChanged(JsonElement root, out string entityId, out JsonElement newState)
+    {
+        entityId = string.Empty;
+        newState = default;
+
+        if (MessageType(root) != "event"
+            || !root.TryGetProperty("event", out var evt)
+            || !evt.TryGetProperty("event_type", out var evtType)
+            || evtType.GetString() != "state_changed"
+            || !evt.TryGetProperty("data", out var data))
+        {
+            return false;
+        }
+
+        if (!data.TryGetProperty("entity_id", out var id) || id.ValueKind != JsonValueKind.String)
+        {
+            return false;
+        }
+
+        if (!data.TryGetProperty("new_state", out var state) || state.ValueKind != JsonValueKind.Object)
+        {
+            return false; // entity removed (new_state null) — nothing to ingest.
+        }
+
+        entityId = id.GetString()!;
+        newState = state;
+        return true;
+    }
+}

src/Infrastructure/Ingestion/HomeAssistantWebSocketWorker.cs

@@ -0,0 +1,292 @@
+using System.Collections.Concurrent;
+using System.Net.WebSockets;
+using System.Text;
+using System.Text.Json;
+using MeterVault.Core.Domain;
+using MeterVault.Infrastructure.Persistence;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace MeterVault.Infrastructure.Ingestion;
+
+/// <summary>
+/// Real-time Home Assistant ingestion via the WebSocket API (SDD §6.2 push path). For each enabled
+/// HA endpoint whose config sets <c>UseWebSocket</c>, holds a persistent connection that authenticates,
+/// subscribes to <c>state_changed</c> events, and ingests changes for the endpoint's configured
+/// entities as they happen. Reconnects with capped backoff. Endpoints without <c>UseWebSocket</c> stay
+/// on the REST poll worker (<see cref="HomeAssistantWorker"/>) — each endpoint is served by exactly one.
+/// Tokens are resolved from environment variables named in the endpoint config, never stored plaintext.
+/// </summary>
+public sealed class HomeAssistantWebSocketWorker(
+    IServiceScopeFactory scopeFactory,
+    ILogger<HomeAssistantWebSocketWorker> logger) : BackgroundService
+{
+    private static readonly TimeSpan SuperviseInterval = TimeSpan.FromSeconds(15);
+    private static readonly TimeSpan InitialBackoff = TimeSpan.FromSeconds(2);
+    private static readonly TimeSpan MaxBackoff = TimeSpan.FromSeconds(60);
+    private static readonly TimeSpan EntityMapTtl = TimeSpan.FromSeconds(30);
+
+    private readonly IServiceScopeFactory _scopeFactory = scopeFactory;
+    private readonly ILogger<HomeAssistantWebSocketWorker> _logger = logger;
+    private readonly ConcurrentDictionary<int, Task> _connections = new();
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        using var timer = new PeriodicTimer(SuperviseInterval);
+        do
+        {
+            try
+            {
+                await SuperviseAsync(stoppingToken).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                break;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogWarning(ex, "Home Assistant WebSocket supervisor tick failed; will retry");
+            }
+        }
+        while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false));
+    }
+
+    /// <summary>Starts a connection loop for each WebSocket-enabled endpoint that isn't already running.</summary>
+    private async Task SuperviseAsync(CancellationToken stoppingToken)
+    {
+        List<int> wsEndpointIds;
+        await using (var scope = _scopeFactory.CreateAsyncScope())
+        {
+            var db = scope.ServiceProvider.GetRequiredService<MeterVaultDbContext>();
+            var endpoints = await db.IngestionEndpoints
+                .Where(e => e.IsEnabled && e.Type == EndpointType.HomeAssistant)
+                .ToListAsync(stoppingToken).ConfigureAwait(false);
+            wsEndpointIds = endpoints
+                .Where(e => HaEndpointConfig.Parse(e.Config).UseWebSocket)
+                .Select(e => e.Id)
+                .ToList();
+        }
+
+        // Reap finished connection loops (endpoint disabled/removed, or terminal failure).
+        foreach (var id in _connections.Where(kv => kv.Value.IsCompleted).Select(kv => kv.Key).ToList())
+        {
+            _connections.TryRemove(id, out _);
+        }
+
+        foreach (var endpointId in wsEndpointIds)
+        {
+            // Fire-and-forget: the loop stores its own Task in _connections and self-terminates when
+            // the endpoint is disabled/removed; the supervisor reaps completed entries above.
+            _ = _connections.GetOrAdd(endpointId, id => Task.Run(() => RunConnectionAsync(id, stoppingToken), stoppingToken));
+        }
+    }
+
+    /// <summary>Connect → listen → reconnect loop for one endpoint, until it is disabled or the app stops.</summary>
+    private async Task RunConnectionAsync(int endpointId, CancellationToken stoppingToken)
+    {
+        var backoff = InitialBackoff;
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            HaEndpointConfig config;
+            string? token;
+            await using (var scope = _scopeFactory.CreateAsyncScope())
+            {
+                var db = scope.ServiceProvider.GetRequiredService<MeterVaultDbContext>();
+                var endpoint = await db.IngestionEndpoints
+                    .FirstOrDefaultAsync(e => e.Id == endpointId, stoppingToken).ConfigureAwait(false);
+                if (endpoint is null || !endpoint.IsEnabled)
+                {
+                    return; // gone/disabled — stop; the supervisor will restart it if it comes back.
+                }
+
+                config = HaEndpointConfig.Parse(endpoint.Config);
+                if (!config.UseWebSocket)
+                {
+                    return; // switched to poll mode.
+                }
+
+                token = config.ResolveToken();
+            }
+
+            if (string.IsNullOrWhiteSpace(config.BaseUrl) || string.IsNullOrWhiteSpace(token))
+            {
+                _logger.LogWarning("HA WebSocket endpoint {EndpointId} missing base URL or token; retrying", endpointId);
+            }
+            else
+            {
+                try
+                {
+                    await ListenAsync(endpointId, config.BaseUrl!, token!, stoppingToken).ConfigureAwait(false);
+                    backoff = InitialBackoff; // clean close → reset backoff.
+                }
+                catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+                {
+                    return;
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogWarning(ex, "HA WebSocket connection to endpoint {EndpointId} dropped; reconnecting", endpointId);
+                    await UpdateEndpointStatusAsync(endpointId, "disconnected", CancellationToken.None).ConfigureAwait(false);
+                }
+            }
+
+            try
+            {
+                await Task.Delay(backoff, stoppingToken).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                return;
+            }
+
+            backoff = TimeSpan.FromSeconds(Math.Min(MaxBackoff.TotalSeconds, backoff.TotalSeconds * 2));
+        }
+    }
+
+    private async Task ListenAsync(int endpointId, string baseUrl, string token, CancellationToken stoppingToken)
+    {
+        using var ws = new ClientWebSocket();
+        await ws.ConnectAsync(HaWebSocketProtocol.WebSocketUri(baseUrl), stoppingToken).ConfigureAwait(false);
+
+        // Handshake: auth_required → auth → auth_ok.
+        using (var required = await ReceiveJsonAsync(ws, stoppingToken).ConfigureAwait(false))
+        {
+            if (!HaWebSocketProtocol.IsAuthRequired(required.RootElement))
+            {
+                // Some setups may not send auth_required first; proceed to auth regardless.
+                _logger.LogDebug("HA WebSocket did not send auth_required first (type={Type})", HaWebSocketProtocol.MessageType(required.RootElement));
+            }
+        }
+
+        await SendAsync(ws, HaWebSocketProtocol.AuthMessage(token), stoppingToken).ConfigureAwait(false);
+        using (var authResult = await ReceiveJsonAsync(ws, stoppingToken).ConfigureAwait(false))
+        {
+            if (!HaWebSocketProtocol.IsAuthOk(authResult.RootElement))
+            {
+                throw new InvalidOperationException(
+                    $"Home Assistant WebSocket auth failed (type={HaWebSocketProtocol.MessageType(authResult.RootElement)}).");
+            }
+        }
+
+        await SendAsync(ws, HaWebSocketProtocol.SubscribeStateChanged(1), stoppingToken).ConfigureAwait(false);
+        await UpdateEndpointStatusAsync(endpointId, "connected (ws)", stoppingToken).ConfigureAwait(false);
+        _logger.LogInformation("HA WebSocket connected for endpoint {EndpointId}", endpointId);
+
+        var entityMap = await LoadEntityMapAsync(endpointId, stoppingToken).ConfigureAwait(false);
+        var mapLoadedAt = DateTimeOffset.UtcNow;
+
+        while (!stoppingToken.IsCancellationRequested && ws.State == WebSocketState.Open)
+        {
+            using var doc = await ReceiveJsonAsync(ws, stoppingToken).ConfigureAwait(false);
+            if (!HaWebSocketProtocol.TryReadStateChanged(doc.RootElement, out var entityId, out var newState))
+            {
+                continue;
+            }
+
+            if (DateTimeOffset.UtcNow - mapLoadedAt > EntityMapTtl)
+            {
+                entityMap = await LoadEntityMapAsync(endpointId, stoppingToken).ConfigureAwait(false);
+                mapLoadedAt = DateTimeOffset.UtcNow;
+            }
+
+            if (!entityMap.TryGetValue(entityId, out var sources))
+            {
+                continue;
+            }
+
+            foreach (var (sourceId, attribute) in sources)
+            {
+                if (HaStateClient.ParseStateElement(newState, attribute) is { } state)
+                {
+                    await IngestAsync(sourceId, state, stoppingToken).ConfigureAwait(false);
+                }
+            }
+        }
+    }
+
+    /// <summary>Maps each configured entity id to the sources (and optional attribute) that read it.</summary>
+    private async Task<Dictionary<string, List<(int SourceId, string? Attribute)>>> LoadEntityMapAsync(
+        int endpointId, CancellationToken cancellationToken)
+    {
+        await using var scope = _scopeFactory.CreateAsyncScope();
+        var db = scope.ServiceProvider.GetRequiredService<MeterVaultDbContext>();
+        var sources = await db.MeterSources.AsNoTracking()
+            .Where(s => s.IsEnabled && s.SourceType == SourceType.HomeAssistant && s.EndpointId == endpointId)
+            .ToListAsync(cancellationToken).ConfigureAwait(false);
+
+        var map = new Dictionary<string, List<(int, string?)>>(StringComparer.OrdinalIgnoreCase);
+        foreach (var source in sources)
+        {
+            var config = SourceConfig.Parse(source.Config);
+            if (string.IsNullOrWhiteSpace(config.EntityId))
+            {
+                continue;
+            }
+
+            if (!map.TryGetValue(config.EntityId, out var list))
+            {
+                map[config.EntityId] = list = [];
+            }
+
+            list.Add((source.Id, config.Attribute));
+        }
+
+        return map;
+    }
+
+    private async Task IngestAsync(int sourceId, HaState state, CancellationToken cancellationToken)
+    {
+        await using var scope = _scopeFactory.CreateAsyncScope();
+        var ingestion = scope.ServiceProvider.GetRequiredService<IngestionService>();
+        await ingestion.IngestAsync(sourceId, state.Time, state.Value, cancellationToken).ConfigureAwait(false);
+    }
+
+    private async Task UpdateEndpointStatusAsync(int endpointId, string status, CancellationToken cancellationToken)
+    {
+        try
+        {
+            await using var scope = _scopeFactory.CreateAsyncScope();
+            var db = scope.ServiceProvider.GetRequiredService<MeterVaultDbContext>();
+            var endpoint = await db.IngestionEndpoints
+                .FirstOrDefaultAsync(e => e.Id == endpointId, cancellationToken).ConfigureAwait(false);
+            if (endpoint is not null)
+            {
+                endpoint.LastStatus = status;
+                endpoint.LastSeenAt = DateTimeOffset.UtcNow;
+                await db.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
+            }
+        }
+        catch (Exception ex)
+        {
+            _logger.LogDebug(ex, "Failed to update HA endpoint {EndpointId} status", endpointId);
+        }
+    }
+
+    private static Task SendAsync(ClientWebSocket ws, string json, CancellationToken cancellationToken) =>
+        ws.SendAsync(Encoding.UTF8.GetBytes(json), WebSocketMessageType.Text, endOfMessage: true, cancellationToken);
+
+    /// <summary>Reads one (possibly fragmented) text message and parses it as JSON.</summary>
+    private static async Task<JsonDocument> ReceiveJsonAsync(ClientWebSocket ws, CancellationToken cancellationToken)
+    {
+        using var buffer = new MemoryStream();
+        var chunk = new byte[8192];
+        WebSocketReceiveResult result;
+        do
+        {
+            result = await ws.ReceiveAsync(chunk, cancellationToken).ConfigureAwait(false);
+            if (result.MessageType == WebSocketMessageType.Close)
+            {
+                throw new InvalidOperationException(
+                    $"Home Assistant closed the WebSocket ({result.CloseStatus}: {result.CloseStatusDescription}).");
+            }
+
+            buffer.Write(chunk, 0, result.Count);
+        }
+        while (!result.EndOfMessage);
+
+        buffer.Position = 0;
+        return await JsonDocument.ParseAsync(buffer, cancellationToken: cancellationToken).ConfigureAwait(false);
+    }
+}

src/Infrastructure/Ingestion/HomeAssistantWorker.cs

@@ -54,9 +54,14 @@ public sealed class HomeAssistantWorker(
         var ingestion = scope.ServiceProvider.GetRequiredService<IngestionService>();
         var client = new HaStateClient(_httpClientFactory.CreateClient());
 
-        var endpoints = await db.IngestionEndpoints
+        // Endpoints using the WebSocket push path are served by HomeAssistantWebSocketWorker; skip
+        // them here so a source is never both polled and pushed.
+        var enabled = await db.IngestionEndpoints
             .Where(e => e.IsEnabled && e.Type == EndpointType.HomeAssistant)
-            .ToDictionaryAsync(e => e.Id, cancellationToken).ConfigureAwait(false);
+            .ToListAsync(cancellationToken).ConfigureAwait(false);
+        var endpoints = enabled
+            .Where(e => !HaEndpointConfig.Parse(e.Config).UseWebSocket)
+            .ToDictionary(e => e.Id);
         if (endpoints.Count == 0)
         {
             return;

tests/Core.Tests/InstantRateNormalizerTests.cs

@@ -0,0 +1,79 @@
+using MeterVault.Core.Domain;
+using MeterVault.Core.Normalization;
+using static MeterVault.Core.Tests.TestData;
+
+namespace MeterVault.Core.Tests;
+
+/// <summary>
+/// instant_rate: the reading value is an instantaneous rate (e.g. kW) integrated over time into
+/// consumption (kWh). Uses the trapezoidal rule between consecutive samples, attributed to the
+/// interval's end reading — the first reading only seeds the integral.
+/// </summary>
+public sealed class InstantRateNormalizerTests
+{
+    private readonly INormalizationEngine _engine = NormalizationEngine.CreateDefault();
+
+    private static DateTimeOffset At(int hour) => new(new DateTime(2024, 6, 1, 0, 0, 0, DateTimeKind.Utc).AddHours(hour));
+
+    [Fact]
+    public void Constant_rate_integrates_to_rate_times_hours()
+    {
+        // 2 kW held steady for 3 hours → 6 kWh, booked at the end of the interval.
+        var ctx = new NormalizationContext
+        {
+            Meter = new MeterConfig { MeterId = 1, Mode = MeterMode.InstantRate, Unit = "kWh" },
+            Readings = [Reading(1, At(0), 2), Reading(1, At(3), 2)],
+        };
+
+        var result = _engine.Normalize(ctx);
+
+        var row = Assert.Single(result);
+        Assert.Equal(6d, row.Amount, 6);
+        Assert.Equal(At(3), row.Time);
+        Assert.Equal(ConsumptionKind.Consumption, row.Kind);
+    }
+
+    [Fact]
+    public void Linear_ramp_integrates_trapezoidally_per_interval()
+    {
+        // 0 kW → 2 kW → 4 kW at 1-hour steps. Intervals: (0+2)/2·1 = 1, (2+4)/2·1 = 3.
+        var ctx = new NormalizationContext
+        {
+            Meter = new MeterConfig { MeterId = 1, Mode = MeterMode.InstantRate, Unit = "kWh" },
+            Readings = [Reading(1, At(0), 0), Reading(1, At(1), 2), Reading(1, At(2), 4)],
+        };
+
+        var result = _engine.Normalize(ctx);
+
+        // First reading only seeds the integral: N readings → N−1 rows.
+        Assert.Equal([1d, 3d], result.Select(c => c.Amount));
+        Assert.Equal([At(1), At(2)], result.Select(c => c.Time));
+    }
+
+    [Fact]
+    public void Negative_rate_is_preserved_as_export()
+    {
+        // A bidirectional power sensor reading −4 kW for an hour → −4 kWh (net export), not clamped.
+        var ctx = new NormalizationContext
+        {
+            Meter = new MeterConfig { MeterId = 1, Mode = MeterMode.InstantRate, Unit = "kWh" },
+            Readings = [Reading(1, At(0), -4), Reading(1, At(1), -4)],
+        };
+
+        var result = _engine.Normalize(ctx);
+
+        Assert.Equal(-4d, Assert.Single(result).Amount, 6);
+    }
+
+    [Fact]
+    public void Single_reading_produces_no_consumption()
+    {
+        var ctx = new NormalizationContext
+        {
+            Meter = new MeterConfig { MeterId = 1, Mode = MeterMode.InstantRate, Unit = "kWh" },
+            Readings = [Reading(1, At(0), 5)],
+        };
+
+        Assert.Empty(_engine.Normalize(ctx));
+    }
+}

tests/Integration.Tests/Ingestion/HaWebSocketProtocolTests.cs

@@ -0,0 +1,119 @@
+using System.Text.Json;
+using MeterVault.Infrastructure.Ingestion;
+
+namespace MeterVault.Integration.Tests.Ingestion;
+
+/// <summary>
+/// The pure Home Assistant WebSocket protocol helpers: URL derivation, the auth/subscribe frames,
+/// and reading the entity id + numeric value out of a <c>state_changed</c> event (the transport is
+/// exercised separately). No network — these are the parts that must be provably correct.
+/// </summary>
+public sealed class HaWebSocketProtocolTests
+{
+    [Theory]
+    [InlineData("http://ha.local:8123", "ws", "ha.local", 8123)]
+    [InlineData("http://ha.local:8123/", "ws", "ha.local", 8123)]
+    [InlineData("https://ha.example.com", "wss", "ha.example.com", 443)]
+    public void WebSocketUri_maps_scheme_and_appends_api_path(string baseUrl, string scheme, string host, int port)
+    {
+        var uri = HaWebSocketProtocol.WebSocketUri(baseUrl);
+
+        Assert.Equal(scheme, uri.Scheme);
+        Assert.Equal(host, uri.Host);
+        Assert.Equal(port, uri.Port);
+        Assert.Equal("/api/websocket", uri.AbsolutePath);
+    }
+
+    [Fact]
+    public void AuthMessage_carries_type_and_token()
+    {
+        using var doc = JsonDocument.Parse(HaWebSocketProtocol.AuthMessage("secret-token"));
+
+        Assert.Equal("auth", doc.RootElement.GetProperty("type").GetString());
+        Assert.Equal("secret-token", doc.RootElement.GetProperty("access_token").GetString());
+    }
+
+    [Fact]
+    public void SubscribeStateChanged_requests_state_changed_events()
+    {
+        using var doc = JsonDocument.Parse(HaWebSocketProtocol.SubscribeStateChanged(7));
+
+        Assert.Equal(7, doc.RootElement.GetProperty("id").GetInt32());
+        Assert.Equal("subscribe_events", doc.RootElement.GetProperty("type").GetString());
+        Assert.Equal("state_changed", doc.RootElement.GetProperty("event_type").GetString());
+    }
+
+    [Theory]
+    [InlineData("{\"type\":\"auth_required\",\"ha_version\":\"2024.6\"}", true, false, false)]
+    [InlineData("{\"type\":\"auth_ok\"}", false, true, false)]
+    [InlineData("{\"type\":\"auth_invalid\",\"message\":\"bad token\"}", false, false, true)]
+    public void Auth_message_types_are_recognized(string json, bool required, bool ok, bool invalid)
+    {
+        using var doc = JsonDocument.Parse(json);
+        var root = doc.RootElement;
+
+        Assert.Equal(required, HaWebSocketProtocol.IsAuthRequired(root));
+        Assert.Equal(ok, HaWebSocketProtocol.IsAuthOk(root));
+        Assert.Equal(invalid, HaWebSocketProtocol.IsAuthInvalid(root));
+    }
+
+    [Fact]
+    public void TryReadStateChanged_extracts_entity_and_numeric_state()
+    {
+        using var doc = JsonDocument.Parse(StateChangedEvent(state: "1234.5"));
+
+        Assert.True(HaWebSocketProtocol.TryReadStateChanged(doc.RootElement, out var entityId, out var newState));
+        Assert.Equal("sensor.house_power", entityId);
+
+        var parsed = HaStateClient.ParseStateElement(newState, attribute: null);
+        Assert.NotNull(parsed);
+        Assert.Equal(1234.5, parsed!.Value.Value, 3);
+        Assert.Equal(new DateTimeOffset(2024, 6, 15, 10, 0, 0, TimeSpan.Zero), parsed.Value.Time);
+    }
+
+    [Fact]
+    public void TryReadStateChanged_reads_a_named_attribute()
+    {
+        using var doc = JsonDocument.Parse(StateChangedEvent(state: "on"));
+
+        Assert.True(HaWebSocketProtocol.TryReadStateChanged(doc.RootElement, out _, out var newState));
+        // state "on" is non-numeric, but the 'current' attribute is a number.
+        Assert.Null(HaStateClient.ParseStateElement(newState, attribute: null));
+        var byAttribute = HaStateClient.ParseStateElement(newState, attribute: "current");
+        Assert.Equal(42, byAttribute!.Value.Value, 3);
+    }
+
+    [Fact]
+    public void TryReadStateChanged_ignores_removed_entities_and_other_events()
+    {
+        using var removed = JsonDocument.Parse(
+            "{\"type\":\"event\",\"event\":{\"event_type\":\"state_changed\",\"data\":{\"entity_id\":\"sensor.x\",\"new_state\":null}}}");
+        Assert.False(HaWebSocketProtocol.TryReadStateChanged(removed.RootElement, out _, out _));
+
+        using var other = JsonDocument.Parse(
+            "{\"type\":\"event\",\"event\":{\"event_type\":\"call_service\",\"data\":{}}}");
+        Assert.False(HaWebSocketProtocol.TryReadStateChanged(other.RootElement, out _, out _));
+
+        using var result = JsonDocument.Parse("{\"id\":1,\"type\":\"result\",\"success\":true}");
+        Assert.False(HaWebSocketProtocol.TryReadStateChanged(result.RootElement, out _, out _));
+    }
+
+    private static string StateChangedEvent(string state) => $$"""
+        {
+          "id": 1,
+          "type": "event",
+          "event": {
+            "event_type": "state_changed",
+            "data": {
+              "entity_id": "sensor.house_power",
+              "new_state": {
+                "entity_id": "sensor.house_power",
+                "state": "{{state}}",
+                "attributes": { "unit_of_measurement": "W", "current": 42 },
+                "last_updated": "2024-06-15T10:00:00+00:00"
+              }
+            }
+          }
+        }
+        """;
+}

tests/Integration.Tests/Ingestion/HomeAssistantWebSocketWorkerTests.cs

@@ -0,0 +1,182 @@
+using System.Net.WebSockets;
+using System.Text;
+using System.Text.Json;
+using MeterVault.Core.Domain;
+using MeterVault.Infrastructure.Ingestion;
+using MeterVault.Infrastructure.Persistence;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Hosting.Server;
+using Microsoft.AspNetCore.Hosting.Server.Features;
+using Microsoft.AspNetCore.Http;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+
+namespace MeterVault.Integration.Tests.Ingestion;
+
+/// <summary>
+/// End-to-end proof of the HA WebSocket push path against an in-process fake Home Assistant server:
+/// the worker performs the auth handshake, subscribes to <c>state_changed</c>, and a pushed change
+/// for a configured entity lands as a <c>reading</c>. This exercises the real ClientWebSocket
+/// transport and handshake sequencing that the pure-protocol unit tests can't.
+/// </summary>
+[Collection("Timescale")]
+public sealed class HomeAssistantWebSocketWorkerTests(TimescaleFixture fx)
+{
+    private const string TokenEnvVar = "MV_TEST_HA_WS_TOKEN";
+
+    [Fact]
+    public async Task Pushed_state_change_becomes_a_reading()
+    {
+        await using var db = fx.CreateContext();
+        Environment.SetEnvironmentVariable(TokenEnvVar, "test-token");
+        await using var fake = await FakeHaServer.StartAsync(entityId: "sensor.house_power", state: "4711");
+
+        try
+        {
+            var type = new EnergyType { Key = "ha_ws_test", DisplayName = "HA WS", BaseUnit = "kWh", DefaultMode = MeterMode.CumulativeCounter };
+            db.EnergyTypes.Add(type);
+            await db.SaveChangesAsync();
+
+            var meter = new Meter { Name = "HA WS Meter", EnergyTypeId = type.Id, Mode = MeterMode.CumulativeCounter, Unit = "kWh" };
+            db.Meters.Add(meter);
+            await db.SaveChangesAsync();
+
+            var endpoint = new IngestionEndpoint
+            {
+                Type = EndpointType.HomeAssistant,
+                Name = "Fake HA",
+                IsEnabled = true,
+                Config = new HaEndpointConfig { BaseUrl = fake.BaseUrl, TokenEnv = TokenEnvVar, UseWebSocket = true }.ToJson(),
+            };
+            db.IngestionEndpoints.Add(endpoint);
+            await db.SaveChangesAsync();
+
+            db.MeterSources.Add(new MeterSource
+            {
+                MeterId = meter.Id,
+                SourceType = SourceType.HomeAssistant,
+                EndpointId = endpoint.Id,
+                IsEnabled = true,
+                Config = JsonSerializer.Serialize(new { entityId = "sensor.house_power" }),
+            });
+            await db.SaveChangesAsync();
+
+            await using var provider = BuildProvider(fx.ConnectionString);
+            var worker = new HomeAssistantWebSocketWorker(
+                provider.GetRequiredService<IServiceScopeFactory>(), NullLogger<HomeAssistantWebSocketWorker>.Instance);
+
+            await worker.StartAsync(CancellationToken.None);
+            try
+            {
+                Reading? reading = null;
+                for (var i = 0; i < 60 && reading is null; i++)
+                {
+                    await Task.Delay(200);
+                    reading = await db.Readings.AsNoTracking().FirstOrDefaultAsync(r => r.MeterId == meter.Id);
+                }
+
+                Assert.NotNull(reading);
+                Assert.Equal(4711, reading!.Value, 3);
+                Assert.Equal(ReadingQuality.Measured, reading.Quality);
+            }
+            finally
+            {
+                await worker.StopAsync(CancellationToken.None);
+            }
+        }
+        finally
+        {
+            Environment.SetEnvironmentVariable(TokenEnvVar, null);
+            await db.Readings.ExecuteDeleteAsync();
+            await db.MeterSources.ExecuteDeleteAsync();
+            await db.IngestionEndpoints.ExecuteDeleteAsync();
+            await db.Meters.ExecuteDeleteAsync();
+            await db.EnergyTypes.Where(t => t.Key == "ha_ws_test").ExecuteDeleteAsync();
+        }
+    }
+
+    private static ServiceProvider BuildProvider(string connectionString)
+    {
+        var services = new ServiceCollection();
+        services.AddLogging();
+        services.AddDbContextFactory<MeterVaultDbContext>(o => o
+            .UseNpgsql(connectionString, n => n.MigrationsAssembly(typeof(MeterVaultDbContext).Assembly.FullName))
+            .UseSnakeCaseNamingConvention());
+        services.AddScoped<MeterVaultDbContext>(sp => sp.GetRequiredService<IDbContextFactory<MeterVaultDbContext>>().CreateDbContext());
+        services.AddScoped<IngestionService>();
+        return services.BuildServiceProvider();
+    }
+
+    /// <summary>A minimal Home Assistant WebSocket server: handshake, then push one state_changed event.</summary>
+    private sealed class FakeHaServer : IAsyncDisposable
+    {
+        private readonly WebApplication _app;
+
+        private FakeHaServer(WebApplication app, string baseUrl)
+        {
+            _app = app;
+            BaseUrl = baseUrl;
+        }
+
+        public string BaseUrl { get; }
+
+        public static async Task<FakeHaServer> StartAsync(string entityId, string state)
+        {
+            var builder = WebApplication.CreateBuilder();
+            builder.Logging.ClearProviders();
+            builder.WebHost.UseUrls("http://127.0.0.1:0");
+            var app = builder.Build();
+            app.UseWebSockets();
+
+            var eventJson =
+                "{\"id\":1,\"type\":\"event\",\"event\":{\"event_type\":\"state_changed\",\"data\":{\"entity_id\":\""
+                + entityId + "\",\"new_state\":{\"entity_id\":\"" + entityId + "\",\"state\":\"" + state
+                + "\",\"attributes\":{},\"last_updated\":\"2024-06-15T10:00:00+00:00\"}}}}";
+
+            app.Map("/api/websocket", async context =>
+            {
+                if (!context.WebSockets.IsWebSocketRequest)
+                {
+                    context.Response.StatusCode = 400;
+                    return;
+                }
+
+                using var ws = await context.WebSockets.AcceptWebSocketAsync();
+                await SendAsync(ws, """{"type":"auth_required","ha_version":"2024.6"}""");
+                await ReceiveAsync(ws); // client "auth"
+                await SendAsync(ws, """{"type":"auth_ok"}""");
+                await ReceiveAsync(ws); // client "subscribe_events"
+                await SendAsync(ws, """{"id":1,"type":"result","success":true}""");
+                await SendAsync(ws, eventJson);
+
+                try
+                {
+                    await Task.Delay(Timeout.Infinite, context.RequestAborted);
+                }
+                catch (OperationCanceledException)
+                {
+                    // client/test closed — expected.
+                }
+            });
+
+            await app.StartAsync();
+            var address = app.Services.GetRequiredService<IServer>().Features
+                .Get<IServerAddressesFeature>()!.Addresses.First();
+            return new FakeHaServer(app, address);
+        }
+
+        public async ValueTask DisposeAsync() => await _app.DisposeAsync();
+
+        private static Task SendAsync(WebSocket ws, string json) =>
+            ws.SendAsync(Encoding.UTF8.GetBytes(json), WebSocketMessageType.Text, endOfMessage: true, CancellationToken.None);
+
+        private static async Task ReceiveAsync(WebSocket ws)
+        {
+            var buffer = new byte[8192];
+            await ws.ReceiveAsync(buffer, CancellationToken.None);
+        }
+    }
+}