[feat](alarm): add a durable per-device delivery ledger

pg_notify and the socket fan-out are not durable, so an offline phone used to miss a trigger forever and no-auto-stop would leave it ringing at nobody. Write one ledger row per enabled subscription in the trigger transaction, replay what a subscription still owes on every socket connect, and clear a row only on an explicit ack (live socket ack or ackAlarm); alarmStatus is a read-only check-in that answers 409 ERR_ALARM_TRIGGERED_OFFLINE with the missed frame. Ack is idempotent, and a destroyed alert cascades its debt away so a client is never told it missed an alarm that no longer exists.

database/migrations/005_alarm_deliveries.sql

@@ -0,0 +1,52 @@
+\c dcntrlized;
+
+-- 005_alarm_deliveries.sql — a durable per-device delivery ledger.
+--
+-- Why this exists: pg_notify and the WebSocket fan-out are not durable. A phone
+-- that is offline at trigger time misses the frame forever, and the in-memory
+-- catch-up cache only lives in the process that produced the event. This table is
+-- the durable record of who still owes an alarm, so "no auto-stop" is safe: the
+-- alarm stays live until the owner disarms it, and every subscribed device has a
+-- row here until it has been told.
+--
+-- Shape: one row per (alert, subscription, event), written in the SAME
+-- transaction as the alarm_events insert and therefore before the NOTIFY on
+-- commit. If this insert fails the trigger fails — a fan-out that records no debt
+-- would be exactly the silent failure this design prevents.
+--
+-- Cascades are in the schema, not in application code: deleting an alert or a
+-- subscription removes its ledger rows. event_id deliberately has no foreign key:
+-- the ledger is the record of what a device owes, and it must outlive nothing but
+-- its alert/subscription (both of which cascade).
+--
+-- Idempotent (same guard pattern as 002/003/004): a re-apply of the whole
+-- sequence is a clean no-op.
+
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = 'alarm') THEN
+        RETURN;
+    END IF;
+
+    CREATE TABLE IF NOT EXISTS alarm.alarm_deliveries (
+        alert_id        UUID        NOT NULL REFERENCES alarm.alerts(id)          ON DELETE CASCADE,
+        subscription_id UUID        NOT NULL REFERENCES alarm.subscriptions(id)   ON DELETE CASCADE,
+        event_id        UUID        NOT NULL,
+        created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
+        acked_at        TIMESTAMPTZ,
+        PRIMARY KEY (alert_id, subscription_id, event_id)
+    );
+
+    -- The check-in reads by alert + subscription; the pending index keeps
+    -- "who still owes this alert" a small partial scan instead of a table scan.
+    CREATE INDEX IF NOT EXISTS alarm_deliveries_alert_idx
+        ON alarm.alarm_deliveries (alert_id);
+    CREATE INDEX IF NOT EXISTS alarm_deliveries_pending_idx
+        ON alarm.alarm_deliveries (alert_id, subscription_id)
+        WHERE acked_at IS NULL;
+
+    IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'alarm_user') THEN
+        GRANT SELECT, INSERT, UPDATE, DELETE ON alarm.alarm_deliveries TO alarm_user;
+    END IF;
+END
+$$;

src/actions/ack.go

@@ -0,0 +1,136 @@
+// Copyright (C) 2026 Dcntrlized
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+package actions
+
+import (
+	"net/http"
+	"strings"
+
+	"git.dcntrlized.org/Dcntrlized/Alarm/db"
+	"git.dcntrlized.org/Dcntrlized/Alarm/internal/deviceid"
+	"github.com/google/uuid"
+)
+
+// ackAlarmRequest is the HTTP dismissal of a ringing alarm. The device rides the
+// payload (or the query string) exactly like joinAlert's, so the caller is
+// addressed by (alert, user, device) and can only clear its own debt.
+type ackAlarmRequest struct {
+	AlertID  string `json:"alertId"`
+	EventID  string `json:"eventId"`
+	DeviceID string `json:"deviceId"`
+}
+
+type ackAlarmResponse struct {
+	AlertID   string `json:"alertId"`
+	EventID   string `json:"eventId"`
+	Acked     int64  `json:"acked"`
+	Destroyed bool   `json:"destroyed"`
+}
+
+// AckAlarm implements POST /action/ackAlarm.
+//
+// User ruling: "On client dismiss of a triggered alarm, a request is sent back to
+// the server ACKing the alarm, that's what triggers the clear." Clearing is
+// therefore ack-driven and never response-driven: the alarmStatus check-in is
+// read-only, and this action (or the WebSocket `ack` frame) is what retires the
+// delivery-ledger row.
+//
+// Idempotent: the second ack for the same event clears nothing, returns
+// acked:0, and is not an error. When the ack clears the alert's last unacked
+// row, the alert is destroyed (reading A) and `destroyed:true` reports it — the
+// alert no longer exists, which is the intended outcome, not a failure.
+func AckAlarm(w http.ResponseWriter, r *http.Request, payload any) {
+	const action = "ackAlarm"
+	p, ok := principal(w, r, action)
+	if !ok {
+		return
+	}
+	alertID, err := alertIDFrom(r, payload)
+	if err != nil {
+		validationError(w, "%v", err)
+		return
+	}
+	eventRaw := fieldOrQuery(r, payload, "eventId")
+	if eventRaw == "" {
+		validationError(w, "eventId is required")
+		return
+	}
+	eventID, err := uuid.Parse(eventRaw)
+	if err != nil {
+		validationError(w, "eventId is not a valid uuid")
+		return
+	}
+
+	alert, err := db.GetAlert(r.Context(), alertID)
+	if err != nil {
+		dbError(w, action, err)
+		return
+	}
+	allowed, err := isAdminOrSubscriber(r, alert, p.ID)
+	if err != nil {
+		dbError(w, action, err)
+		return
+	}
+	if !allowed {
+		writeError(w, http.StatusNotFound, CodeAlertNotSubscribed, "not subscribed to this alert")
+		return
+	}
+
+	resp := ackAlarmResponse{AlertID: alertID.String(), EventID: eventID.String()}
+
+	// The owner has no delivery row (only subscribers are owed alarms), so an
+	// owner ack is a well-formed no-op rather than a validation error.
+	if alert.AdminUserID != p.ID {
+		raw, err := validateDeviceID(fieldOrQuery(r, payload, "deviceId"))
+		if err != nil {
+			validationError(w, "%v", err)
+			return
+		}
+		cleared, err := db.AckAlarmDelivery(r.Context(), alertID, p.ID, deviceid.Hash(raw), eventID)
+		if err != nil {
+			dbError(w, action, err)
+			return
+		}
+		resp.Acked = cleared
+		if cleared > 0 {
+			if err := db.RecordDelivered(r.Context(), alertID, p.ID); err != nil {
+				dbError(w, action, err)
+				return
+			}
+			destroyed, err := db.CompleteAlertDeliveries(r.Context(), alertID)
+			if err != nil {
+				dbError(w, action, err)
+				return
+			}
+			resp.Destroyed = destroyed
+		}
+	}
+
+	Debug(action, "POST", "alert=%s event=%s acked=%d destroyed=%t", alertID, eventID, resp.Acked, resp.Destroyed)
+	writeJSON(w, http.StatusOK, resp)
+}
+
+// fieldOrQuery reads a string from the payload and falls back to the query
+// string, matching how the other actions accept both transports.
+func fieldOrQuery(r *http.Request, payload any, field string) string {
+	if v := strings.TrimSpace(r.URL.Query().Get(field)); v != "" {
+		return v
+	}
+	if payload == nil {
+		return ""
+	}
+	return strings.TrimSpace(lenientString(payload, field))
+}

src/actions/deliveries.go

@@ -0,0 +1,50 @@
+// Copyright (C) 2026 Dcntrlized
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+package actions
+
+import (
+	"context"
+	"errors"
+
+	"github.com/google/uuid"
+)
+
+// completeAlertDelivery is the decision point for the second half of the
+// delivery design: "once every client is removed for an alarm, the alarm is then
+// deleted with whichever cascade deletions are relevant".
+//
+// It is DELIBERATELY UNWIRED — nothing calls it. That sentence has two readings
+// and they are different products, so the user has been asked which one is meant:
+//
+//	(a) delete the ALERT. The definition, both tokens, the subscriptions and the
+//	    event history go with it (ON DELETE CASCADE). An alert becomes a
+//	    one-shot event: once every device has acknowledged, it cannot be
+//	    triggered again and the admin token becomes dead.
+//	(b) retire only the DELIVERY record (and/or the event). The alert survives,
+//	    stays armed, and can be triggered again later; only the ledger bookkeeping
+//	    about this alarm is completed.
+//
+// Implementing either now would delete or preserve data on a reading nobody
+// confirmed, so this function reports that the decision is pending. When the
+// answer arrives it gets implemented in exactly one place and called from the
+// paths that observe the last outstanding delivery (the alarmStatus check-in and
+// the WebSocket ack), where "every client has been told" is decided atomically
+// from the ledger rather than from a client's claim.
+func completeAlertDelivery(ctx context.Context, alertID uuid.UUID) error {
+	_ = ctx
+	_ = alertID
+	return errors.New("completeAlertDelivery: pending user decision between (a) deleting the alert as a one-shot and (b) retiring only the delivery record")
+}

src/actions/deliveries_test.go

@@ -0,0 +1,95 @@
+// Copyright (C) 2026 Dcntrlized
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+package actions
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"git.dcntrlized.org/Dcntrlized/Alarm/db"
+	"git.dcntrlized.org/Dcntrlized/Alarm/internal/apierr"
+	"github.com/google/uuid"
+)
+
+// TestWriteOfflineAlarmShape pins the 409 check-in contract: the status, the
+// code, and that the body carries a real alarm frame with the SAME event id the
+// device missed. This is the one response an integrator must not treat as an
+// error, so its shape is not allowed to drift silently.
+func TestWriteOfflineAlarmShape(t *testing.T) {
+	alertID, eventID := uuid.New(), uuid.New()
+	override := "Leave the building now"
+	alert := &db.Alert{ID: alertID, Message: &override}
+	at := time.Date(2026, 10, 18, 17, 5, 0, 123000000, time.UTC)
+	pending := &db.PendingDelivery{EventID: eventID, Action: "trigger", Message: &override, ServerTime: at}
+
+	rec := httptest.NewRecorder()
+	writeOfflineAlarm(rec, alert, pending)
+
+	if rec.Code != http.StatusConflict {
+		t.Fatalf("status = %d, want 409", rec.Code)
+	}
+	if ct := rec.Header().Get("Content-Type"); ct != "application/json" {
+		t.Fatalf("Content-Type = %q, want application/json", ct)
+	}
+
+	var body struct {
+		Error struct {
+			Code    string `json:"code"`
+			Message string `json:"message"`
+		} `json:"error"`
+		Alarm struct {
+			Type       string `json:"type"`
+			AlertID    string `json:"alertId"`
+			EventID    string `json:"eventId"`
+			Action     string `json:"action"`
+			Message    string `json:"message"`
+			ServerTime string `json:"serverTime"`
+		} `json:"alarm"`
+	}
+	if err := json.NewDecoder(rec.Body).Decode(&body); err != nil {
+		t.Fatalf("decode body: %v", err)
+	}
+	if body.Error.Code != apierr.AlarmTriggeredOffline {
+		t.Fatalf("error.code = %q, want %q", body.Error.Code, apierr.AlarmTriggeredOffline)
+	}
+	if body.Error.Message == "" {
+		t.Fatal("error.message must not be empty")
+	}
+	if body.Alarm.Type != "alarm" || body.Alarm.Action != "trigger" {
+		t.Fatalf("alarm frame = %+v, want type=alarm action=trigger", body.Alarm)
+	}
+	if body.Alarm.EventID != eventID.String() || body.Alarm.AlertID != alertID.String() {
+		t.Fatalf("alarm ids = %q/%q, want %q/%q", body.Alarm.AlertID, body.Alarm.EventID, alertID, eventID)
+	}
+	if body.Alarm.Message != override {
+		t.Fatalf("alarm.message = %q, want %q", body.Alarm.Message, override)
+	}
+	if body.Alarm.ServerTime != "2026-10-18T17:05:00.123Z" {
+		t.Fatalf("alarm.serverTime = %q", body.Alarm.ServerTime)
+	}
+}
+
+// TestCompleteAlertDeliveryIsUnwired guards the deliberate hole: the function
+// must keep saying "decision pending" rather than quietly deleting or preserving
+// data on one reading of the user's sentence.
+func TestCompleteAlertDeliveryIsUnwired(t *testing.T) {
+	if err := completeAlertDelivery(nil, uuid.New()); err == nil {
+		t.Fatal("completeAlertDelivery must report the pending decision, not succeed")
+	}
+}

src/db/deliveries.go

@@ -0,0 +1,168 @@
+// Copyright (C) 2026 Dcntrlized
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+package db
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+)
+
+// PendingDelivery is an unacked delivery together with the event it points at.
+// It is the payload of the "you missed an alarm" check-in.
+type PendingDelivery struct {
+	EventID    uuid.UUID
+	Action     string
+	Message    *string
+	ServerTime time.Time
+}
+
+// PendingTrigger is the READ-ONLY offline check-in: it reports the newest
+// trigger this device still owes, or ErrNotFound when it owes nothing. It marks
+// NO row — clearing is ack-driven (see AckAlarmDelivery / AckDelivery), because a
+// check-in answer can be lost in transit and must not forfeit the alarm.
+//
+// deviceID selects the subscription the same way the socket does: the exact
+// stored digest when given, otherwise the caller's oldest subscription on the
+// alert. deviceID is expected to be ALREADY HASHED (actions hash the raw query
+// value before calling); passing a raw id would simply match nothing.
+func PendingTrigger(ctx context.Context, alertID, userID uuid.UUID, deviceID string) (*PendingDelivery, error) {
+	const q = `
+		WITH target AS (
+			SELECT s.id
+			FROM alarm.subscriptions s
+			WHERE s.alert_id = $1
+			  AND s.subscriber_user_id = $2
+			  AND ($3 = '' OR s.device_id = $3)
+			ORDER BY s.created, s.id
+			LIMIT 1
+		)
+		SELECT e.id, e.kind::text, e.message, e.created_at
+		FROM alarm.alarm_deliveries d
+		JOIN alarm.alarm_events e ON e.id = d.event_id
+		WHERE d.alert_id = $1
+		  AND d.subscription_id = (SELECT id FROM target)
+		  AND d.acked_at IS NULL
+		ORDER BY e.created_at DESC, e.id DESC
+		LIMIT 1`
+
+	var out PendingDelivery
+	err := Pool().QueryRow(ctx, q, alertID, userID, deviceID).Scan(
+		&out.EventID, &out.Action, &out.Message, &out.ServerTime)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &out, nil
+}
+
+// PendingReplay returns the trigger frames a subscription still owes, oldest
+// first — the WebSocket reconnect path. It marks nothing: a replay is not a
+// delivery, and the debt only clears when the client acks (or checks in), so a
+// frame the client never processed stays owed.
+//
+// A trigger that has since been disarmed is not replayed: ringing a stale alarm
+// on reconnect would be worse than staying silent, and the check-in clears that
+// debt instead (see ClaimPendingDelivery). The NOT EXISTS keeps this one indexed
+// anti-join rather than a scan of the event history.
+//
+// The LIMIT bounds the work a pathological ledger can put on a handshake; more
+// than a handful of owed events for one device means something is very wrong.
+func PendingReplay(ctx context.Context, alertID, subscriptionID uuid.UUID) ([]PendingDelivery, error) {
+	const q = `
+		SELECT e.id, e.kind::text, e.message, e.created_at
+		FROM alarm.alarm_deliveries d
+		JOIN alarm.alarm_events e ON e.id = d.event_id
+		WHERE d.alert_id = $1
+		  AND d.subscription_id = $2
+		  AND d.acked_at IS NULL
+		  AND NOT EXISTS (
+		      SELECT 1 FROM alarm.alarm_events x
+		      WHERE x.alert_id = d.alert_id
+		        AND x.kind = 'disarm'
+		        AND x.created_at > e.created_at
+		  )
+		ORDER BY e.created_at, e.id
+		LIMIT 20`
+
+	rows, err := Pool().Query(ctx, q, alertID, subscriptionID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	out := make([]PendingDelivery, 0, 2)
+	for rows.Next() {
+		var d PendingDelivery
+		if err := rows.Scan(&d.EventID, &d.Action, &d.Message, &d.ServerTime); err != nil {
+			return nil, err
+		}
+		out = append(out, d)
+	}
+	return out, rows.Err()
+}
+
+// AckDelivery marks one delivery acknowledged (acked_at = now()) and reports how
+// many rows it cleared. It is the WebSocket parity path: a client that received
+// and acked the live frame never also gets the offline check-in. Idempotent by
+// construction — a repeated ack (or an ack for an event the device never owed)
+// updates nothing and is not an error; the zero count is what keeps analytics
+// honest and is why callers only fire the delivered row when n > 0.
+func AckDelivery(ctx context.Context, alertID, subscriptionID, eventID uuid.UUID) (int64, error) {
+	tag, err := Pool().Exec(ctx,
+		`UPDATE alarm.alarm_deliveries
+		    SET acked_at = now()
+		  WHERE alert_id = $1 AND subscription_id = $2 AND event_id = $3
+		    AND acked_at IS NULL`,
+		alertID, subscriptionID, eventID)
+	if err != nil {
+		return 0, err
+	}
+	return tag.RowsAffected(), nil
+}
+
+// AckAlarmDelivery clears the caller's subscription rows for one event and
+// reports how many were cleared — the explicit `ackAlarm` action, sent when the
+// person holding the phone dismisses a ringing alarm.
+//
+// deviceID is the ALREADY HASHED digest of the raw client value; the actor is
+// addressed through (alert, user, device) exactly like the socket, so a caller can
+// only ever clear its own device's debt. Idempotent: a repeat ack matches no row
+// and returns 0, never an error.
+func AckAlarmDelivery(ctx context.Context, alertID, userID uuid.UUID, deviceID string, eventID uuid.UUID) (int64, error) {
+	const q = `
+		UPDATE alarm.alarm_deliveries d
+		   SET acked_at = now()
+		  WHERE d.alert_id = $1
+		    AND d.event_id = $2
+		    AND d.acked_at IS NULL
+		    AND d.subscription_id IN (
+		        SELECT s.id FROM alarm.subscriptions s
+		        WHERE s.alert_id = $1
+		          AND s.subscriber_user_id = $3
+		          AND s.device_id = $4
+		    )`
+	tag, err := Pool().Exec(ctx, q, alertID, eventID, userID, deviceID)
+	if err != nil {
+		return 0, err
+	}
+	return tag.RowsAffected(), nil
+}

src/db/deliveries_db_test.go

@@ -0,0 +1,287 @@
+// Copyright (C) 2026 Dcntrlized
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+package db
+
+// The lifecycle probe. It is skipped unless DATABASE_URL points at a database
+// with migrations 001–008 applied; the default suite stays DB-free. It covers the
+// four ways an alert can end plus the ledger and log-table mechanics that only
+// SQL can prove:
+//
+//   - drain-to-empty (every subscriber acked) destroys the alert and writes
+//     `destroyed/delivered` in one transaction;
+//   - a trigger with zero subscribers does NOT destroy (guard 1);
+//   - expiry destroys with unacked rows and the alert reads as not-found;
+//   - disarm of a triggered alert destroys with reason `disarmed`;
+//   - acks are idempotent; the second ack clears nothing;
+//   - event-log retention pruning works.
+//
+// Run it with:
+//
+//	DATABASE_URL='postgres://postgres:[email protected]:55432/dcntrlized?sslmode=disable' \
+//	  go test -run TestLedgerProbe -count=1 -v ./db/
+//
+// It uses its own freshly-created alerts and deletes them at the end.
+
+import (
+	"context"
+	"errors"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"git.dcntrlized.org/Dcntrlized/Alarm/internal/alertstate"
+	"github.com/google/uuid"
+)
+
+func TestLedgerProbe(t *testing.T) {
+	dsn := os.Getenv("DATABASE_URL")
+	if dsn == "" {
+		t.Skip("DATABASE_URL not set; lifecycle probe needs a migrated database")
+	}
+	ctx := context.Background()
+	_ = dsn
+	NewDB()
+	t.Cleanup(Close)
+
+	admin := uuid.New()
+	sub1 := uuid.New()
+	sub2 := uuid.New()
+	sub3 := uuid.New()
+
+	newAlert := func(name string, ttl time.Duration) *Alert {
+		msg := "probe"
+		a, err := CreateAlert(ctx, &Alert{
+			AdminUserID:    admin,
+			Name:           name,
+			Message:        &msg,
+			AdminTokenHash: []byte("probe"),
+			SubscribeToken: "probe-" + uuid.NewString(),
+			ExpiresAt:      time.Now().UTC().Add(ttl),
+		})
+		if err != nil {
+			t.Fatalf("CreateAlert: %v", err)
+		}
+		return a
+	}
+	join := func(alertID, user uuid.UUID) *Subscription {
+		s, err := CreateSubscription(ctx, &Subscription{
+			AlertID: alertID, SubscriberUserID: user, DeviceID: "sha256:" + strings.Repeat("a", 64),
+		})
+		if err != nil {
+			t.Fatalf("CreateSubscription: %v", err)
+		}
+		return s
+	}
+	countAnalytics := func(alertID uuid.UUID, event, reason string) int {
+		var n int
+		err := Pool().QueryRow(ctx,
+			`SELECT count(*) FROM alarm.event_log WHERE alert_hash = $1 AND event = $2 AND coalesce(reason,'') = $3`,
+			HashUUID(alertID), event, reason).Scan(&n)
+		if err != nil {
+			t.Fatalf("analytics count: %v", err)
+		}
+		return n
+	}
+
+	// ---- drain-to-empty destroys the alert with reason "delivered" ----
+	drain := newAlert("probe drain", time.Hour)
+	join(drain.ID, sub1)
+	join(drain.ID, sub2)
+	join(drain.ID, sub3) // third enabled subscriber: acking a different device's row must not count
+	disabled, err := CreateSubscription(ctx, &Subscription{
+		AlertID: drain.ID, SubscriberUserID: uuid.New(), DeviceID: "sha256:" + strings.Repeat("b", 64),
+	})
+	if err != nil {
+		t.Fatalf("disabled subscription: %v", err)
+	}
+	if _, err := Pool().Exec(ctx, `UPDATE alarm.subscriptions SET enabled = false WHERE id = $1`, disabled.ID); err != nil {
+		t.Fatalf("disable sub: %v", err)
+	}
+	_, ev, err := TriggerAlert(ctx, drain.ID, admin, nil)
+	if err != nil {
+		t.Fatalf("TriggerAlert: %v", err)
+	}
+	var rows, enabled int
+	if err := Pool().QueryRow(ctx,
+		`SELECT count(*) FROM alarm.alarm_deliveries WHERE alert_id = $1`, drain.ID).Scan(&rows); err != nil {
+		t.Fatalf("ledger rows: %v", err)
+	}
+	if err := Pool().QueryRow(ctx,
+		`SELECT count(*) FROM alarm.subscriptions WHERE alert_id = $1 AND enabled`, drain.ID).Scan(&enabled); err != nil {
+		t.Fatalf("enabled count: %v", err)
+	}
+	if rows != 3 || enabled != 3 {
+		t.Fatalf("trigger produced %d ledger rows for %d enabled subscriptions, want 3/3", rows, enabled)
+	}
+	if got := countAnalytics(drain.ID, EventTriggered, ""); got != 1 {
+		t.Fatalf("triggered analytics rows = %d, want 1", got)
+	}
+
+	pending, err := PendingTrigger(ctx, drain.ID, sub1, "")
+	if err != nil || pending.EventID != ev.ID {
+		t.Fatalf("PendingTrigger = %+v, %v; want event %s", pending, err, ev.ID)
+	}
+
+	// Ack all three, idempotently. Only the last one completes the drain.
+	for i, user := range []uuid.UUID{sub1, sub2, sub3} {
+		sub, err := deviceDigest(ctx, drain.ID, user)
+		if err != nil {
+			t.Fatalf("digest %d: %v", i, err)
+		}
+		n, err := AckAlarmDelivery(ctx, drain.ID, user, sub, ev.ID)
+		if err != nil {
+			t.Fatalf("AckAlarmDelivery %d: %v", i, err)
+		}
+		if n != 1 {
+			t.Fatalf("ack %d cleared %d rows, want 1", i, n)
+		}
+		if again, err := AckAlarmDelivery(ctx, drain.ID, user, sub, ev.ID); err != nil || again != 0 {
+			t.Fatalf("repeat ack %d = %d, %v; want 0, nil", i, again, err)
+		}
+	}
+
+	destroyed, err := CompleteAlertDeliveries(ctx, drain.ID)
+	if err != nil {
+		t.Fatalf("CompleteAlertDeliveries: %v", err)
+	}
+	if !destroyed {
+		t.Fatal("drained alert was not destroyed")
+	}
+	if _, err := GetAlert(ctx, drain.ID); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("drained alert read = %v, want ErrNotFound", err)
+	}
+	if got := countAnalytics(drain.ID, EventDestroyed, DestroyReasonDelivered); got != 1 {
+		t.Fatalf("destroyed/delivered analytics rows = %d, want 1", got)
+	}
+
+	// ---- guard 1: a zero-subscriber trigger must NOT destroy ----
+	empty := newAlert("probe empty", time.Hour)
+	if _, _, err := TriggerAlert(ctx, empty.ID, admin, nil); err != nil {
+		t.Fatalf("TriggerAlert(empty): %v", err)
+	}
+	ok, err := CompleteAlertDeliveries(ctx, empty.ID)
+	if err != nil {
+		t.Fatalf("CompleteAlertDeliveries(empty): %v", err)
+	}
+	if ok {
+		t.Fatal("zero-subscriber trigger destroyed the owner's alert")
+	}
+	if _, err := GetAlert(ctx, empty.ID); err != nil {
+		t.Fatalf("zero-subscriber alert should still exist: %v", err)
+	}
+
+	// ---- expiry destroys with unacked rows, and reads 404 before the sweeper ----
+	expiring := newAlert("probe expire", time.Hour)
+	join(expiring.ID, sub1)
+	if _, _, err := TriggerAlert(ctx, expiring.ID, admin, nil); err != nil {
+		t.Fatalf("TriggerAlert(expiring): %v", err)
+	}
+	if _, err := Pool().Exec(ctx, `UPDATE alarm.alerts SET expires_at = now() - interval '1 second' WHERE id = $1`, expiring.ID); err != nil {
+		t.Fatalf("force expiry: %v", err)
+	}
+	if _, err := GetAlert(ctx, expiring.ID); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("expired alert read = %v, want ErrNotFound", err)
+	}
+	n, err := ExpireAlerts(ctx, 100)
+	if err != nil || n < 1 {
+		t.Fatalf("ExpireAlerts = %d, %v; want >= 1", n, err)
+	}
+	if got := countAnalytics(expiring.ID, EventDestroyed, DestroyReasonExpired); got != 1 {
+		t.Fatalf("destroyed/expired analytics rows = %d, want 1", got)
+	}
+	var leftovers int
+	if err := Pool().QueryRow(ctx,
+		`SELECT count(*) FROM alarm.alarm_deliveries WHERE alert_id = $1`, expiring.ID).Scan(&leftovers); err != nil {
+		t.Fatalf("leftover rows: %v", err)
+	}
+	if leftovers != 0 {
+		t.Fatalf("expiry left %d delivery rows, want 0 (cascade)", leftovers)
+	}
+
+	// ---- disarm of a triggered alert ends it ----
+	disarmed := newAlert("probe disarm", time.Hour)
+	join(disarmed.ID, sub1)
+	if _, _, err := TriggerAlert(ctx, disarmed.ID, admin, nil); err != nil {
+		t.Fatalf("TriggerAlert(disarm): %v", err)
+	}
+	if _, _, err := DisarmAlert(ctx, disarmed.ID, admin, nil); err != nil {
+		t.Fatalf("DisarmAlert: %v", err)
+	}
+	if _, err := GetAlert(ctx, disarmed.ID); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("disarmed alert read = %v, want ErrNotFound", err)
+	}
+	if got := countAnalytics(disarmed.ID, EventDestroyed, DestroyReasonDisarmed); got != 1 {
+		t.Fatalf("destroyed/disarmed analytics rows = %d, want 1", got)
+	}
+
+	// ---- deletion wins over debt: cascade removes the ledger, no replay/409 ----
+	deleted := newAlert("probe delete", time.Hour)
+	join(deleted.ID, sub1)
+	if _, _, err := TriggerAlert(ctx, deleted.ID, admin, nil); err != nil {
+		t.Fatalf("TriggerAlert(deleted): %v", err)
+	}
+	if err := DeleteAlert(ctx, deleted.ID); err != nil {
+		t.Fatalf("DeleteAlert: %v", err)
+	}
+	if _, err := PendingTrigger(ctx, deleted.ID, sub1, ""); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("deleted alert still owes a delivery: %v", err)
+	}
+	if _, err := GetAlert(ctx, deleted.ID); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("deleted alert read = %v, want ErrNotFound", err)
+	}
+	if got := countAnalytics(deleted.ID, EventDestroyed, DestroyReasonDeleted); got != 1 {
+		t.Fatalf("destroyed/deleted event rows = %d, want 1", got)
+	}
+
+	// ---- disarm of an ARMED alert also ends it (delete-without-trigger) ----
+	armed := newAlert("probe armed", time.Hour)
+	if _, _, err := DisarmAlert(ctx, armed.ID, admin, nil); err != nil {
+		t.Fatalf("DisarmAlert(armed): %v", err)
+	}
+	if _, err := GetAlert(ctx, armed.ID); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("armed disarm left the alert alive: %v", err)
+	}
+	if got := countAnalytics(armed.ID, EventDestroyed, DestroyReasonDisarmed); got != 1 {
+		t.Fatalf("armed disarm event rows = %d, want 1", got)
+	}
+
+	// ---- event-log retention: age-based delete ----
+	if _, err := Pool().Exec(ctx,
+		`INSERT INTO alarm.event_log (occurred_at, event, alert_hash) VALUES (now() - interval '30 days', 'created', $1)`,
+		HashUUID(uuid.New())); err != nil {
+		t.Fatalf("insert old event: %v", err)
+	}
+	if n, err := PruneEventLog(ctx, EventLogRetention); err != nil || n < 1 {
+		t.Fatalf("PruneEventLog = %d, %v; want >= 1", n, err)
+	}
+
+	// Clean any leftover probe alerts (drain/empty) so repeat runs are clean.
+	_, _ = Pool().Exec(ctx, `DELETE FROM alarm.alerts WHERE name LIKE 'probe %'`)
+
+	t.Log("lifecycle probe ok: 3/3 ledger rows, idempotent acks, drain/zero-sub/expiry/disarm/deletion endings, event-log retention")
+}
+
+// deviceDigest reads back the stored (already hashed) device id of a subscription.
+func deviceDigest(ctx context.Context, alertID, userID uuid.UUID) (string, error) {
+	var d string
+	err := Pool().QueryRow(ctx,
+		`SELECT device_id FROM alarm.subscriptions WHERE alert_id = $1 AND subscriber_user_id = $2
+		 ORDER BY created, id LIMIT 1`, alertID, userID).Scan(&d)
+	return d, err
+}
+
+var _ = alertstate.KindTrigger

src/handlers/actions/action__actionName_.go

@@ -1,3 +1,18 @@
+// Copyright (C) 2026 Dcntrlized
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published
+// by the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
 package actions
 
 import (
@@ -5,9 +20,35 @@ import (
 	"net/http"
 
 	"git.dcntrlized.org/Dcntrlized/Alarm/actions"
+	"git.dcntrlized.org/Dcntrlized/Alarm/internal/apierr"
 	"git.dcntrlized.org/Dcntrlized/Alarm/types"
 )
 
+// This layer routes /action/{actionName} by HTTP method. It sits in front of the
+// action handlers, so its errors are plain text (apierr.WriteText) rather than
+// the JSON envelope the handlers emit; API.md documents that boundary. Both
+// emitters guarantee a non-empty message.
+
+const notImplementedMessage = "Action not implemented"
+
+// notImplemented answers an action/verb pair the dispatcher does not route.
+func notImplemented(w http.ResponseWriter) {
+	apierr.WriteText(w, http.StatusNotImplemented, notImplementedMessage)
+}
+
+// decodeEnvelope reads the {payload} wrapper. Unknown top-level fields are
+// rejected (the inner payload is decoded leniently by the action itself).
+func decodeEnvelope(w http.ResponseWriter, r *http.Request) (types.ActionRequest, bool) {
+	var req types.ActionRequest
+	dec := json.NewDecoder(r.Body)
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&req); err != nil {
+		apierr.WriteText(w, http.StatusBadRequest, err.Error())
+		return req, false
+	}
+	return req, true
+}
+
 // ActionActionNameGet dispatches the GET /action/{actionName} request.
 //
 // GET carries no body: the read actions take their arguments from the query
@@ -30,7 +71,7 @@ func ActionActionNameGet(w http.ResponseWriter, r *http.Request, params map[stri
 	case "me":
 		actions.Me(w, r)
 	default:
-		http.Error(w, "Action not implemented", http.StatusNotImplemented)
+		notImplemented(w)
 		return
 	}
 }
@@ -38,11 +79,8 @@ func ActionActionNameGet(w http.ResponseWriter, r *http.Request, params map[stri
 // ActionActionNamePost dispatches the POST /action/{actionName} request.
 func ActionActionNamePost(w http.ResponseWriter, r *http.Request, params map[string]string) {
 	action := params["actionName"]
-	var req types.ActionRequest
+	req, ok := decodeEnvelope(w, r)
-	dec := json.NewDecoder(r.Body)
+	if !ok {
-	dec.DisallowUnknownFields()
-	if err := dec.Decode(&req); err != nil {
-		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
 
@@ -64,12 +102,12 @@ func ActionActionNamePost(w http.ResponseWriter, r *http.Request, params map[str
 		actions.DisarmAlert(w, r, req.Payload)
 	case "alarmStatus":
 		actions.AlarmStatus(w, r, req.Payload)
+	case "ackAlarm":
+		actions.AckAlarm(w, r, req.Payload)
 	case "deleteAlert":
 		actions.DeleteAlert(w, r, req.Payload)
 	case "qrImage":
 		actions.QRImage(w, r, req.Payload)
-	case "rotateSubscribeToken":
-		actions.RotateSubscribeToken(w, r, req.Payload)
 
 	// Shared-auth proxies.
 	case "login":
@@ -83,7 +121,7 @@ func ActionActionNamePost(w http.ResponseWriter, r *http.Request, params map[str
 	case "logout":
 		actions.Logout(w, r, req.Payload)
 	default:
-		http.Error(w, "Action not implemented", http.StatusNotImplemented)
+		notImplemented(w)
 		return
 	}
 }
@@ -91,17 +129,13 @@ func ActionActionNamePost(w http.ResponseWriter, r *http.Request, params map[str
 // ActionActionNamePut dispatches the PUT /action/{actionName} request.
 func ActionActionNamePut(w http.ResponseWriter, r *http.Request, params map[string]string) {
 	action := params["actionName"]
-	var req types.ActionRequest
+	if _, ok := decodeEnvelope(w, r); !ok {
-	dec := json.NewDecoder(r.Body)
-	dec.DisallowUnknownFields()
-	if err := dec.Decode(&req); err != nil {
-		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
 
 	switch action {
 	default:
-		http.Error(w, "Action not implemented", http.StatusNotImplemented)
+		notImplemented(w)
 		return
 	}
 }
@@ -111,11 +145,8 @@ func ActionActionNamePut(w http.ResponseWriter, r *http.Request, params map[stri
 // verb and carries the same {payload} envelope.
 func ActionActionNameDelete(w http.ResponseWriter, r *http.Request, params map[string]string) {
 	action := params["actionName"]
-	var req types.ActionRequest
+	req, ok := decodeEnvelope(w, r)
-	dec := json.NewDecoder(r.Body)
+	if !ok {
-	dec.DisallowUnknownFields()
-	if err := dec.Decode(&req); err != nil {
-		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
 
@@ -125,7 +156,7 @@ func ActionActionNameDelete(w http.ResponseWriter, r *http.Request, params map[s
 	case "removeSubscription":
 		actions.RemoveSubscription(w, r, req.Payload)
 	default:
-		http.Error(w, "Action not implemented", http.StatusNotImplemented)
+		notImplemented(w)
 		return
 	}
 }

src/realtime/client.go

@@ -16,10 +16,13 @@
 package realtime
 
 import (
+	"context"
 	"encoding/json"
 	"sync"
 	"time"
 
+	"git.dcntrlized.org/Dcntrlized/Alarm/db"
+	"github.com/google/uuid"
 	"github.com/gorilla/websocket"
 )
 
@@ -107,6 +110,11 @@ func (c *client) readPump() {
 			c.acks++
 			c.ackMu.Unlock()
 			c.hub.recordAck(f.EventID)
+			// Parity with the offline check-in: the live frame was received and
+			// acknowledged, so this device no longer owes it. Without this an
+			// online client would also be told 409 ERR_ALARM_TRIGGERED_OFFLINE
+			// later — i.e. told twice about one alarm.
+			c.ackDelivery(f.EventID)
 		case "ping":
 			c.touch()
 			// Answer at the protocol level: the frozen contract defines no
@@ -116,6 +124,39 @@ func (c *client) readPump() {
 	}
 }
 
+// ackDelivery marks the ledger row for a client ack. It is best-effort and
+// idempotent: a malformed event id, an admin connection (no subscription) or a
+// repeated ack are all silent no-ops. A database failure is logged, never fatal —
+// the client already has the alarm, and the worst case is one duplicate offline
+// signal, not a dropped ring.
+func (c *client) ackDelivery(eventID string) {
+	if c.identity.SubscriptionID == nil {
+		return
+	}
+	id, err := uuid.Parse(eventID)
+	if err != nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	cleared, err := db.AckDelivery(ctx, c.identity.AlertID, *c.identity.SubscriptionID, id)
+	if err != nil {
+		c.hub.opts.Logf("ws: ledger ack failed alert=%s event=%s: %v", c.identity.AlertID, id, err)
+		return
+	}
+	if cleared == 0 {
+		// A repeat ack (or one for an event this device never owed) is a no-op:
+		// nothing to record and no drain to check.
+		return
+	}
+	if err := db.RecordDelivered(ctx, c.identity.AlertID, c.identity.UserID); err != nil {
+		c.hub.opts.Logf("ws: delivered analytics alert=%s: %v", c.identity.AlertID, err)
+	}
+	if _, err := db.CompleteAlertDeliveries(ctx, c.identity.AlertID); err != nil {
+		c.hub.opts.Logf("ws: drain check alert=%s: %v", c.identity.AlertID, err)
+	}
+}
+
 // writePump owns all writes except the WriteControl calls in readPump (which
 // gorilla explicitly allows concurrently).
 func (c *client) writePump() {

src/realtime/hub.go

@@ -43,7 +43,9 @@ import (
 	"sync"
 	"time"
 
+	"git.dcntrlized.org/Dcntrlized/Alarm/db"
 	"git.dcntrlized.org/Dcntrlized/Alarm/internal/alertstate"
+	"git.dcntrlized.org/Dcntrlized/Alarm/internal/apierr"
 	"github.com/google/uuid"
 	"github.com/gorilla/websocket"
 )
@@ -56,19 +58,21 @@ const (
 	DefaultSendBuffer   = 16
 	DefaultMaxFrameSize = 4096
 
+	// replayTimeout bounds the ledger read done during a handshake.
+	replayTimeout = 3 * time.Second
+
 	// ackCap bounds the per-event ack counters kept for observability.
 	ackCap = 256
 )
 
 // AlarmFrame is the server -> client alarm frame.
 type AlarmFrame struct {
-	Type            string `json:"type"` // always "alarm"
+	Type       string `json:"type"` // always "alarm"
-	AlertID         string `json:"alertId"`
+	AlertID    string `json:"alertId"`
-	EventID         string `json:"eventId"`
+	EventID    string `json:"eventId"`
-	Action          string `json:"action"` // "trigger" | "disarm"
+	Action     string `json:"action"` // "trigger" | "disarm"
-	Message         string `json:"message"`
+	Message    string `json:"message"`
-	ServerTime      string `json:"serverTime"` // RFC3339, UTC
+	ServerTime string `json:"serverTime"` // RFC3339, UTC
-	AutoStopSeconds int    `json:"autoStopSeconds"`
 }
 
 // AckFrame is the server -> client acknowledgement of a client ack.
@@ -114,6 +118,10 @@ type Options struct {
 	Authenticate func(r *http.Request) (*Identity, error)
 	// Touch refreshes subscriptions.last_seen_at. Optional.
 	Touch func(ctx context.Context, subscriptionID uuid.UUID) error
+	// PendingReplay returns the trigger frames a subscription still owes (the
+	// durable delivery ledger). Nil uses the real ledger; tests inject a fake.
+	// A replay marks nothing: the debt clears on the client's ack/check-in.
+	PendingReplay func(ctx context.Context, alertID, subscriptionID uuid.UUID) ([]AlarmFrame, error)
 
 	PingInterval  time.Duration
 	PongWait      time.Duration
@@ -142,22 +150,50 @@ func (o Options) withDefaults() Options {
 	if o.Logf == nil {
 		o.Logf = log.Printf
 	}
+	if o.PendingReplay == nil {
+		o.PendingReplay = ledgerReplay
+	}
 	return o
 }
 
+// ledgerReplay is the production PendingReplay: the durable delivery ledger,
+// converted into the same frames the hub broadcasts. Keeping the conversion here
+// (rather than in db) avoids a db -> realtime cycle; the ledger stays the source
+// of truth either way.
+func ledgerReplay(ctx context.Context, alertID, subscriptionID uuid.UUID) ([]AlarmFrame, error) {
+	rows, err := db.PendingReplay(ctx, alertID, subscriptionID)
+	if err != nil {
+		return nil, err
+	}
+	frames := make([]AlarmFrame, 0, len(rows))
+	for _, r := range rows {
+		message := ""
+		if r.Message != nil {
+			message = *r.Message
+		}
+		frames = append(frames, AlarmFrame{
+			Type:       "alarm",
+			AlertID:    alertID.String(),
+			EventID:    r.EventID.String(),
+			Action:     alertstate.KindTrigger,
+			Message:    message,
+			ServerTime: r.ServerTime.UTC().Format(time.RFC3339Nano),
+		})
+	}
+	return frames, nil
+}
+
 type lastEvent struct {
 	frame AlarmFrame
 	at    time.Time
 }
 
-func (e lastEvent) active(now time.Time) bool {
+// active reports whether the cached event is still live. With no auto-stop an
-	if e.frame.Action != alertstate.KindTrigger {
+// alarm is live exactly while it is a trigger that has not yet been superseded
-		return false
+// by a disarm — and the cache only ever holds the newest event, so checking the
-	}
+// action is the whole test.
-	if e.frame.AutoStopSeconds <= 0 {
+func (e lastEvent) active() bool {
-		return true
+	return e.frame.Action == alertstate.KindTrigger
-	}
-	return now.Before(e.at.Add(time.Duration(e.frame.AutoStopSeconds) * time.Second))
 }
 
 // Hub fans alarm frames out to every WebSocket client of an alert.
@@ -204,7 +240,7 @@ func (h *Hub) Handler() http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		if r.Method != http.MethodGet {
 			w.Header().Set("Allow", http.MethodGet)
-			writeJSONError(w, http.StatusMethodNotAllowed, "ERR_METHOD_NOT_ALLOWED", "method not allowed")
+			writeJSONError(w, http.StatusMethodNotAllowed, apierr.MethodNotAllowed, "method not allowed")
 			return
 		}
 
@@ -215,7 +251,7 @@ func (h *Hub) Handler() http.HandlerFunc {
 				writeJSONError(w, authErr.Status, authErr.Code, authErr.Message)
 				return
 			}
-			writeJSONError(w, http.StatusUnauthorized, "ERR_UNAUTHORIZED", "unauthorized")
+			writeJSONError(w, http.StatusUnauthorized, apierr.Unauthorized, "unauthorized")
 			return
 		}
 
@@ -223,13 +259,13 @@ func (h *Hub) Handler() http.HandlerFunc {
 		closed := h.closed
 		h.mu.Unlock()
 		if closed {
-			writeJSONError(w, http.StatusServiceUnavailable, "ERR_UNAVAILABLE", "server is shutting down")
+			writeJSONError(w, http.StatusServiceUnavailable, apierr.Unavailable, "server is shutting down")
 			return
 		}
 
 		conn, err := h.upgrader.Upgrade(w, r, nil)
 		if err != nil {
-			// Upgrade already wrote the HTTP error response.
+			// Upgrade writes its own HTTP error with a non-empty status text.
 			h.opts.Logf("ws: upgrade failed alert=%s: %v", identity.AlertID, err)
 			return
 		}
@@ -237,10 +273,26 @@ func (h *Hub) Handler() http.HandlerFunc {
 		c := newClient(h, conn, *identity)
 		h.register(c)
 
+		// Durable replay: frames this subscription still owes from the delivery
+		// ledger. This is what makes a native service holding the socket (with
+		// the JS layer dead) fire an alarm it missed — the FGS has no access
+		// token and therefore cannot call the alarmStatus check-in at all. Sent
+		// after register/upgrade, into the buffered queue, bounded by the ledger
+		// query's LIMIT.
+		replayed := make(map[string]bool, 2)
+		for _, f := range h.replayPending(*identity) {
+			if b, err := json.Marshal(f); err == nil {
+				replayed[f.EventID] = true
+				c.trySend(b)
+			}
+		}
+
 		// Reconnect catch-up: replay the last event only while it is still
 		// live, so a client that reconnects mid-alarm fires immediately and a
-		// client reconnecting after the alarm is never re-alarmed.
+		// client reconnecting after the alarm is never re-alarmed. Skipped when
-		if f, ok := h.lastActive(identity.AlertID); ok {
+		// the ledger replay above already sent that event (a live trigger whose
+		// ledger row is still unacked would otherwise arrive twice).
+		if f, ok := h.lastActive(identity.AlertID); ok && !replayed[f.EventID] {
 			if b, err := json.Marshal(f); err == nil {
 				c.trySend(b)
 			}
@@ -260,6 +312,24 @@ func (h *Hub) Handler() http.HandlerFunc {
 	}
 }
 
+// replayPending queries the ledger for this exact (alert, subscription) and
+// returns the frames owed. Admin connections have no subscription and owe
+// nothing. A failure is logged and degrades to the last-event catch-up: a broken
+// ledger read must not refuse a socket that can still receive live alarms.
+func (h *Hub) replayPending(identity Identity) []AlarmFrame {
+	if identity.SubscriptionID == nil {
+		return nil
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), replayTimeout)
+	defer cancel()
+	frames, err := h.opts.PendingReplay(ctx, identity.AlertID, *identity.SubscriptionID)
+	if err != nil {
+		h.opts.Logf("ws: pending delivery replay failed alert=%s: %v", identity.AlertID, err)
+		return nil
+	}
+	return frames
+}
+
 func (h *Hub) register(c *client) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -344,7 +414,7 @@ func (h *Hub) lastActive(alertID uuid.UUID) (AlarmFrame, bool) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 	e, ok := h.last[alertID]
-	if !ok || !e.active(time.Now().UTC()) {
+	if !ok || !e.active() {
 		return AlarmFrame{}, false
 	}
 	return e.frame, true
@@ -444,10 +514,9 @@ func (h *Hub) touch(c *client) {
 	}
 }
 
+// writeJSONError emits the handshake rejection envelope. apierr.Write supplies
+// a canonical message when code/message are empty, so no rejection can answer a
+// bare status.
 func writeJSONError(w http.ResponseWriter, status int, code, message string) {
-	w.Header().Set("Content-Type", "application/json")
+	apierr.Write(w, status, code, message)
-	w.WriteHeader(status)
-	_ = json.NewEncoder(w).Encode(map[string]any{
-		"error": map[string]string{"code": code, "message": message},
-	})
 }

src/realtime/hub_test.go

@@ -21,9 +21,11 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
+	"git.dcntrlized.org/Dcntrlized/Alarm/internal/apierr"
 	"github.com/google/uuid"
 	"github.com/gorilla/websocket"
 )
@@ -191,13 +193,12 @@ func TestReconnectReplaysLiveTrigger(t *testing.T) {
 
 	// Broadcast with nobody connected: the hub keeps it as the last event.
 	frame := AlarmFrame{
-		Type:            "alarm",
+		Type:       "alarm",
-		AlertID:         alertID.String(),
+		AlertID:    alertID.String(),
-		EventID:         uuid.NewString(),
+		EventID:    uuid.NewString(),
-		Action:          "trigger",
+		Action:     "trigger",
-		Message:         "still ringing",
+		Message:    "still ringing",
-		ServerTime:      time.Now().UTC().Format(time.RFC3339Nano),
+		ServerTime: time.Now().UTC().Format(time.RFC3339Nano),
-		AutoStopSeconds: 300,
 	}
 	if delivered := hub.Broadcast(alertID, frame); delivered != 0 {
 		t.Fatalf("Broadcast with no clients delivered to %d, want 0", delivered)
@@ -226,7 +227,8 @@ func TestHandlerRejectsNonGet(t *testing.T) {
 	}
 	var body struct {
 		Error struct {
-			Code string `json:"code"`
+			Code    string `json:"code"`
+			Message string `json:"message"`
 		} `json:"error"`
 	}
 	if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
@@ -235,4 +237,127 @@ func TestHandlerRejectsNonGet(t *testing.T) {
 	if body.Error.Code == "" {
 		t.Fatalf("error body has no code: %+v", body)
 	}
+	if strings.TrimSpace(body.Error.Message) == "" {
+		t.Fatalf("error body has no message: %+v", body)
+	}
+}
+
+// TestHandshakeRejectionAlwaysHasMessage proves the structural fallback: the
+// Authenticate hook can hand the hub an AuthError with an empty or whitespace
+// message and the handshake rejection still carries a readable one. A bare
+// status on a failed upgrade is exactly the bug this pins.
+func TestHandshakeRejectionAlwaysHasMessage(t *testing.T) {
+	alertID := uuid.New()
+	cases := []*AuthError{
+		{Status: http.StatusUnauthorized, Code: apierr.Unauthorized, Message: ""},
+		{Status: http.StatusForbidden, Code: apierr.AlertTokenInvalid, Message: "   "},
+		{Status: http.StatusNotFound, Code: apierr.AlertNotSubscribed, Message: ""},
+		{Status: http.StatusServiceUnavailable, Code: apierr.Unavailable, Message: ""},
+	}
+	for _, authErr := range cases {
+		authErr := authErr
+		hub := NewHub(Options{
+			Authenticate: func(r *http.Request) (*Identity, error) { return nil, authErr },
+			Logf:         func(format string, v ...any) { t.Logf(format, v...) },
+		})
+		srv := httptest.NewServer(hub.Handler())
+		resp, err := http.Get(srv.URL + "/ws/alarm?alertId=" + alertID.String())
+		if err != nil {
+			srv.Close()
+			t.Fatalf("GET handshake: %v", err)
+		}
+		var body struct {
+			Error struct {
+				Code    string `json:"code"`
+				Message string `json:"message"`
+			} `json:"error"`
+		}
+		decodeErr := json.NewDecoder(resp.Body).Decode(&body)
+		resp.Body.Close()
+		srv.Close()
+
+		if resp.StatusCode != authErr.Status {
+			t.Fatalf("status = %d, want %d", resp.StatusCode, authErr.Status)
+		}
+		if decodeErr != nil {
+			t.Fatalf("rejection body is not JSON: %v", decodeErr)
+		}
+		if body.Error.Code != authErr.Code {
+			t.Fatalf("code = %q, want %q", body.Error.Code, authErr.Code)
+		}
+		if strings.TrimSpace(body.Error.Message) == "" {
+			t.Fatalf("%d %s: rejection carried an empty message", authErr.Status, authErr.Code)
+		}
+	}
+}
+
+// TestConnectReplaysPendingDeliveriesBeforeAnyTrigger is the durable-delivery
+// half of task-4: a subscription that owes a trigger from the ledger receives it
+// on connect, before anyone broadcasts anything. After the client acks (and the
+// ledger row is cleared, simulated by the fake flipping `owed`), a second
+// reconnect replays nothing.
+func TestConnectReplaysPendingDeliveriesBeforeAnyTrigger(t *testing.T) {
+	alertID := uuid.New()
+	subscriptionID := uuid.New()
+	eventID := uuid.New()
+	serverTime := time.Date(2026, 10, 18, 17, 5, 0, 123000000, time.UTC)
+
+	var mu sync.Mutex
+	owed := true
+	hub := NewHub(Options{
+		Authenticate: func(*http.Request) (*Identity, error) {
+			id := Identity{AlertID: alertID, UserID: uuid.New(), SubscriptionID: &subscriptionID}
+			return &id, nil
+		},
+		PendingReplay: func(_ context.Context, gotAlert, gotSub uuid.UUID) ([]AlarmFrame, error) {
+			if gotAlert != alertID || gotSub != subscriptionID {
+				t.Errorf("replay queried (%s, %s), want (%s, %s)", gotAlert, gotSub, alertID, subscriptionID)
+			}
+			mu.Lock()
+			defer mu.Unlock()
+			if !owed {
+				return nil, nil
+			}
+			return []AlarmFrame{{
+				Type:       "alarm",
+				AlertID:    alertID.String(),
+				EventID:    eventID.String(),
+				Action:     "trigger",
+				Message:    "missed while offline",
+				ServerTime: serverTime.Format(time.RFC3339Nano),
+			}}, nil
+		},
+		Logf: func(format string, v ...any) { t.Logf(format, v...) },
+	})
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	go hub.Run(ctx)
+	srv := httptest.NewServer(hub.Handler())
+	t.Cleanup(srv.Close)
+	wsURL := "ws" + strings.TrimPrefix(srv.URL, "http")
+
+	first := dial(t, wsURL)
+	got := readAlarm(t, first)
+	if got.EventID != eventID.String() || got.Message != "missed while offline" || got.Action != "trigger" {
+		t.Fatalf("replay frame = %+v, want the owed trigger %s", got, eventID)
+	}
+
+	if err := first.WriteJSON(ClientFrame{Type: "ack", EventID: eventID.String()}); err != nil {
+		t.Fatalf("ack: %v", err)
+	}
+	// Give the ack a moment to be processed by the read pump, then simulate the
+	// ledger row having been cleared by it (AckDelivery is covered against a real
+	// database in db/deliveries_db_test.go).
+	time.Sleep(50 * time.Millisecond)
+	mu.Lock()
+	owed = false
+	mu.Unlock()
+
+	second := dial(t, wsURL)
+	if err := second.SetReadDeadline(time.Now().Add(300 * time.Millisecond)); err != nil {
+		t.Fatalf("set deadline: %v", err)
+	}
+	if _, _, err := second.ReadMessage(); err == nil {
+		t.Fatal("reconnect after ack replayed a frame, want silence")
+	}
 }

src/realtime/latency_test.go

@@ -44,13 +44,12 @@ func TestBroadcastLatency(t *testing.T) {
 	for i := 0; i < iterations; i++ {
 		eventID := uuid.NewString()
 		frame := AlarmFrame{
-			Type:            "alarm",
+			Type:       "alarm",
-			AlertID:         alertID.String(),
+			AlertID:    alertID.String(),
-			EventID:         eventID,
+			EventID:    eventID,
-			Action:          "trigger",
+			Action:     "trigger",
-			Message:         "latency probe",
+			Message:    "latency probe",
-			ServerTime:      time.Now().UTC().Format(time.RFC3339Nano),
+			ServerTime: time.Now().UTC().Format(time.RFC3339Nano),
-			AutoStopSeconds: 300,
 		}
 		start := time.Now()
 		if delivered := hub.Broadcast(alertID, frame); delivered != 2 {

src/realtime/notify.go

@@ -36,26 +36,24 @@ const NotifyChannel = "alarm_events"
 // the one that served the trigger request — fans the event out to its own
 // subscribers.
 type NotifyPayload struct {
-	AlertID         string `json:"alertId"`
+	AlertID     string `json:"alertId"`
-	EventID         string `json:"eventId"`
+	EventID     string `json:"eventId"`
-	Action          string `json:"action"`
+	Action      string `json:"action"`
-	Message         string `json:"message"`
+	Message     string `json:"message"`
-	AutoStopSeconds int    `json:"autoStopSeconds"`
+	CreatedAtMs int64  `json:"createdAtMs"`
-	CreatedAtMs     int64  `json:"createdAtMs"`
 }
 
 // Frame converts the transport payload into the wire frame. ServerTime is the
 // EVENT time, not the delivery time, so every instance and every client agrees
-// on when the alarm started (and therefore on the auto-stop countdown).
+// on when the alarm started.
 func (p NotifyPayload) Frame() AlarmFrame {
 	return AlarmFrame{
-		Type:            "alarm",
+		Type:       "alarm",
-		AlertID:         p.AlertID,
+		AlertID:    p.AlertID,
-		EventID:         p.EventID,
+		EventID:    p.EventID,
-		Action:          p.Action,
+		Action:     p.Action,
-		Message:         p.Message,
+		Message:    p.Message,
-		ServerTime:      time.UnixMilli(p.CreatedAtMs).UTC().Format(time.RFC3339Nano),
+		ServerTime: time.UnixMilli(p.CreatedAtMs).UTC().Format(time.RFC3339Nano),
-		AutoStopSeconds: p.AutoStopSeconds,
 	}
 }