feat(sense): implement Control API v1 [T-011]

Sense/internal/store/control.go

@@ -0,0 +1,164 @@
+package store
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/base64"
+	"errors"
+	"strconv"
+	"time"
+
+	"yovision/sense/internal/device"
+)
+
+var (
+	ErrETagMismatch          = errors.New("device ETag mismatch")
+	ErrIdempotencyConflict   = errors.New("idempotency key body conflict")
+	ErrDuplicateSerialNumber = errors.New("duplicate device serial number")
+)
+
+type ControlProjectionVersions struct {
+	QuotaSourceVersion      *int64     `json:"quota_source_version"`
+	AreaPolicySourceVersion *int64     `json:"area_policy_source_version"`
+	SyncedAt                *time.Time `json:"synced_at"`
+}
+
+// ControlDevice is deliberately safe to serialize. It contains configured
+// booleans, never endpoint, credential or profile-token values.
+type ControlDevice struct {
+	ID                   string                    `json:"id"`
+	TenantID             string                    `json:"tenant_id"`
+	SiteID               string                    `json:"site_id"`
+	SerialNumber         string                    `json:"serial_number"`
+	Name                 string                    `json:"name"`
+	Modality             device.Modality           `json:"modality"`
+	Capabilities         []device.Capability       `json:"capabilities"`
+	AreaID               string                    `json:"area_id"`
+	DesiredState         device.DesiredState       `json:"desired_state"`
+	ActualState          device.ActualState        `json:"actual_state"`
+	AdapterStatus        string                    `json:"adapter_status"`
+	EndpointConfigured   bool                      `json:"endpoint_configured"`
+	CredentialConfigured bool                      `json:"credential_configured"`
+	Generation           int64                     `json:"generation"`
+	ObservedGeneration   int64                     `json:"observed_generation"`
+	Converged            bool                      `json:"converged"`
+	FailureCount         int                       `json:"failure_count"`
+	NextAttemptAt        *time.Time                `json:"next_attempt_at,omitempty"`
+	LastErrorCode        *string                   `json:"last_error_code,omitempty"`
+	ProjectionVersions   ControlProjectionVersions `json:"projection_versions"`
+	CreatedAt            time.Time                 `json:"created_at"`
+	UpdatedAt            time.Time                 `json:"updated_at"`
+	ResourceVersion      int64                     `json:"-"`
+}
+
+type ControlSiteQuota struct {
+	Status                 string
+	UsedVideoChannels      int
+	MaxVideoChannels       *int
+	AvailableVideoChannels *int
+	OverLimit              bool
+	SourceVersion          *int64
+	SyncedAt               *time.Time
+}
+
+type ControlListFilter struct {
+	Limit         int
+	AfterCreated  *time.Time
+	AfterDeviceID string
+	Modality      *device.Modality
+	Capability    *device.Capability
+	DesiredState  *device.DesiredState
+	ActualState   *device.ActualState
+}
+
+type ControlDevicePage struct {
+	Items   []ControlDevice
+	HasMore bool
+	Quota   ControlSiteQuota
+}
+
+type IdempotencyScope struct {
+	PrincipalID string
+	TenantID    string
+	SiteID      string
+	Operation   string
+	Key         string
+	RequestHash [sha256.Size]byte
+	TraceID     string
+}
+
+type ControlCreateRequest struct {
+	Scope  IdempotencyScope
+	Device device.Device
+}
+
+type ControlCreateResult struct {
+	Device     ControlDevice
+	AcceptedAt time.Time
+	TraceID    string
+	ETag       string
+	Location   string
+	Replay     bool
+}
+
+type ControlPatch struct {
+	Name          *string
+	AreaID        *string
+	EndpointRef   *string
+	CredentialRef *string
+	ProfileToken  *string
+}
+
+type ControlMutationResult struct {
+	Device     ControlDevice
+	AcceptedAt time.Time
+	TraceID    string
+	ETag       string
+}
+
+type ControlBatchItem struct {
+	DeviceID     string
+	ETag         string
+	DesiredState device.DesiredState
+}
+
+type ControlBatchRequest struct {
+	Scope  IdempotencyScope
+	Reason string
+	Items  []ControlBatchItem
+}
+
+type ControlBatchItemResult struct {
+	DeviceID   string  `json:"device_id"`
+	Status     string  `json:"status"`
+	ErrorCode  *string `json:"error_code"`
+	Message    *string `json:"message"`
+	Generation *int64  `json:"generation"`
+}
+
+type ControlBatchOperation struct {
+	ID          string                   `json:"id"`
+	TenantID    string                   `json:"-"`
+	SiteID      string                   `json:"-"`
+	Status      string                   `json:"status"`
+	SubmittedAt time.Time                `json:"submitted_at"`
+	CompletedAt *time.Time               `json:"completed_at"`
+	Results     []ControlBatchItemResult `json:"results"`
+	TraceID     string                   `json:"trace_id"`
+	Replay      bool                     `json:"-"`
+}
+
+type ControlRepository interface {
+	ListControlDevices(context.Context, string, string, ControlListFilter) (ControlDevicePage, error)
+	CreateControlDevice(context.Context, ControlCreateRequest) (ControlCreateResult, error)
+	GetControlDevice(context.Context, string, string, string) (ControlDevice, error)
+	PatchControlDevice(context.Context, string, string, string, string, ControlPatch) (ControlMutationResult, error)
+	SetControlDesiredState(context.Context, string, string, string, string, device.DesiredState) (ControlMutationResult, error)
+	BatchSetControlDesiredState(context.Context, ControlBatchRequest) (ControlBatchOperation, error)
+	GetControlOperation(context.Context, string, string) (ControlBatchOperation, error)
+}
+
+func DeviceETag(deviceID string, resourceVersion int64) string {
+	digest := sha256.Sum256([]byte(deviceID + "\x00" + strconv.FormatInt(resourceVersion, 10)))
+	return `"` + base64.RawURLEncoding.EncodeToString(digest[:18]) + `"`
+}

Sense/internal/store/control_postgres.go

@@ -0,0 +1,889 @@
+package store
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/sha256"
+	"crypto/subtle"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"math/big"
+	"net/url"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgconn"
+
+	"yovision/sense/internal/device"
+)
+
+const controlReceiptTTL = 24 * time.Hour
+
+const controlDeviceColumns = `d.id, d.tenant_id, d.site_id, d.serial_number, d.name, d.modality,
+	d.area_id, d.desired_state, d.actual_state,
+	(d.endpoint_ref <> ''), (d.credential_ref <> ''),
+	d.generation, d.resource_version,
+	r.observed_generation, r.failure_count, r.next_attempt_at, r.last_error_code,
+	d.quota_source_version, d.area_policy_source_version,
+	GREATEST(
+		(SELECT q.synced_at FROM sense.site_quota_projection_state q
+		 WHERE q.tenant_id = d.tenant_id AND q.site_id = d.site_id
+		   AND q.source_version = d.quota_source_version),
+		(SELECT a.synced_at FROM sense.area_policy_projection_state a
+		 WHERE a.tenant_id = d.tenant_id AND a.site_id = d.site_id
+		   AND a.area_id = d.area_id AND a.source_version = d.area_policy_source_version)
+	),
+	d.created_at, d.updated_at,
+	COALESCE((SELECT jsonb_agg(c.capability ORDER BY c.capability)
+	 FROM sense.device_capabilities c WHERE c.device_id = d.id), '[]'::jsonb)::text`
+
+const controlDeviceSelect = `SELECT ` + controlDeviceColumns + `
+	FROM sense.devices d JOIN sense.reconcile_state r ON r.device_id = d.id`
+
+type controlScanner interface {
+	Scan(...any) error
+}
+
+func scanControlDevice(row controlScanner) (ControlDevice, error) {
+	var value ControlDevice
+	var nextAttempt, syncedAt sql.NullTime
+	var lastError sql.NullString
+	var quotaVersion, areaVersion sql.NullInt64
+	var capabilitiesJSON string
+	err := row.Scan(
+		&value.ID, &value.TenantID, &value.SiteID, &value.SerialNumber, &value.Name,
+		&value.Modality, &value.AreaID, &value.DesiredState, &value.ActualState,
+		&value.EndpointConfigured, &value.CredentialConfigured,
+		&value.Generation, &value.ResourceVersion, &value.ObservedGeneration,
+		&value.FailureCount, &nextAttempt, &lastError, &quotaVersion, &areaVersion,
+		&syncedAt, &value.CreatedAt, &value.UpdatedAt, &capabilitiesJSON,
+	)
+	if err != nil {
+		return ControlDevice{}, err
+	}
+	if err := json.Unmarshal([]byte(capabilitiesJSON), &value.Capabilities); err != nil {
+		return ControlDevice{}, errors.New("decode postgres control device capabilities")
+	}
+	if value.Capabilities == nil {
+		value.Capabilities = make([]device.Capability, 0)
+	}
+	if nextAttempt.Valid {
+		point := nextAttempt.Time.UTC()
+		value.NextAttemptAt = &point
+	}
+	if lastError.Valid {
+		code := lastError.String
+		value.LastErrorCode = &code
+	}
+	if quotaVersion.Valid {
+		version := quotaVersion.Int64
+		value.ProjectionVersions.QuotaSourceVersion = &version
+	}
+	if areaVersion.Valid {
+		version := areaVersion.Int64
+		value.ProjectionVersions.AreaPolicySourceVersion = &version
+	}
+	if syncedAt.Valid {
+		point := syncedAt.Time.UTC()
+		value.ProjectionVersions.SyncedAt = &point
+	}
+	value.Converged = value.ObservedGeneration >= value.Generation && value.FailureCount == 0
+	value.AdapterStatus = controlAdapterStatus(value)
+	return value, nil
+}
+
+func controlAdapterStatus(value ControlDevice) string {
+	if value.LastErrorCode != nil {
+		switch *value.LastErrorCode {
+		case "authentication_failed":
+			return "authentication_failed"
+		case "adapter_not_ready":
+			return "adapter_not_ready"
+		default:
+			return "unavailable"
+		}
+	}
+	if value.Converged {
+		return "ready"
+	}
+	if value.ActualState == device.ActualFailed || value.ActualState == device.ActualOffline {
+		return "unavailable"
+	}
+	return "pending"
+}
+
+func (s *Postgres) GetControlDevice(
+	ctx context.Context, tenantID, siteID, deviceID string,
+) (ControlDevice, error) {
+	value, err := scanControlDevice(s.db.QueryRowContext(ctx, controlDeviceSelect+`
+		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.id = $3`, tenantID, siteID, deviceID))
+	if errors.Is(err, sql.ErrNoRows) {
+		return ControlDevice{}, ErrNotFound
+	}
+	if err != nil {
+		return ControlDevice{}, errors.New("get postgres control device")
+	}
+	return value, nil
+}
+
+func (s *Postgres) ListControlDevices(
+	ctx context.Context, tenantID, siteID string, filter ControlListFilter,
+) (ControlDevicePage, error) {
+	quota, err := s.controlSiteQuota(ctx, tenantID, siteID)
+	if err != nil {
+		return ControlDevicePage{}, err
+	}
+	query := controlDeviceSelect + ` WHERE d.tenant_id = $1 AND d.site_id = $2`
+	arguments := []any{tenantID, siteID}
+	appendCondition := func(clause string, value any) {
+		arguments = append(arguments, value)
+		query += fmt.Sprintf(clause, len(arguments))
+	}
+	if filter.Modality != nil {
+		appendCondition(` AND d.modality = $%d`, *filter.Modality)
+	}
+	if filter.Capability != nil {
+		appendCondition(` AND EXISTS (SELECT 1 FROM sense.device_capabilities fc
+			WHERE fc.device_id = d.id AND fc.capability = $%d)`, *filter.Capability)
+	}
+	if filter.DesiredState != nil {
+		appendCondition(` AND d.desired_state = $%d`, *filter.DesiredState)
+	}
+	if filter.ActualState != nil {
+		appendCondition(` AND d.actual_state = $%d`, *filter.ActualState)
+	}
+	if filter.AfterCreated != nil {
+		arguments = append(arguments, filter.AfterCreated.UTC(), filter.AfterDeviceID)
+		query += fmt.Sprintf(` AND (d.created_at, d.id) > ($%d, $%d)`, len(arguments)-1, len(arguments))
+	}
+	arguments = append(arguments, filter.Limit+1)
+	query += fmt.Sprintf(` ORDER BY d.created_at ASC, d.id ASC LIMIT $%d`, len(arguments))
+	rows, err := s.db.QueryContext(ctx, query, arguments...)
+	if err != nil {
+		return ControlDevicePage{}, errors.New("list postgres control devices")
+	}
+	defer rows.Close()
+	values := make([]ControlDevice, 0, filter.Limit+1)
+	for rows.Next() {
+		value, scanErr := scanControlDevice(rows)
+		if scanErr != nil {
+			return ControlDevicePage{}, errors.New("scan postgres control device page")
+		}
+		values = append(values, value)
+	}
+	if err := rows.Err(); err != nil {
+		return ControlDevicePage{}, errors.New("iterate postgres control device page")
+	}
+	hasMore := len(values) > filter.Limit
+	if hasMore {
+		values = values[:filter.Limit]
+	}
+	return ControlDevicePage{Items: values, HasMore: hasMore, Quota: quota}, nil
+}
+
+func (s *Postgres) controlSiteQuota(ctx context.Context, tenantID, siteID string) (ControlSiteQuota, error) {
+	var maximum int
+	var sourceVersion int64
+	var syncedAt time.Time
+	err := s.db.QueryRowContext(ctx, `SELECT max_video_channels, source_version, source_updated_at
+		FROM bell.site_quota_v1 WHERE tenant_id = $1 AND site_id = $2`, tenantID, siteID).
+		Scan(&maximum, &sourceVersion, &syncedAt)
+	if errors.Is(err, sql.ErrNoRows) {
+		return ControlSiteQuota{}, ErrNotFound
+	}
+	if err != nil {
+		return ControlSiteQuota{}, errors.New("read postgres control site quota")
+	}
+	var used int
+	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM sense.devices d
+		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.desired_state = 'enabled'
+		AND EXISTS (SELECT 1 FROM sense.device_capabilities c
+			WHERE c.device_id = d.id AND c.capability = 'video_capture')`, tenantID, siteID).Scan(&used); err != nil {
+		return ControlSiteQuota{}, errors.New("count postgres control site video channels")
+	}
+	status := "current"
+	if maximum < 1 || maximum > device.MaximumVideoChannels || sourceVersion < 1 || syncedAt.IsZero() {
+		status = "invalid"
+		return ControlSiteQuota{Status: status, UsedVideoChannels: used, OverLimit: false}, nil
+	}
+	available := maximum - used
+	if available < 0 {
+		available = 0
+	}
+	point := syncedAt.UTC()
+	return ControlSiteQuota{
+		Status: status, UsedVideoChannels: used, MaxVideoChannels: &maximum,
+		AvailableVideoChannels: &available, OverLimit: used > maximum,
+		SourceVersion: &sourceVersion, SyncedAt: &point,
+	}, nil
+}
+
+type controlReceipt struct {
+	Status    int
+	Body      []byte
+	ETag      string
+	Location  string
+	TraceID   string
+	CreatedAt time.Time
+}
+
+func controlScopeHash(scope IdempotencyScope) [sha256.Size]byte {
+	encoded, _ := json.Marshal([]string{
+		scope.PrincipalID, scope.TenantID, scope.SiteID, scope.Operation, scope.Key,
+	})
+	return sha256.Sum256(encoded)
+}
+
+func readControlReceipt(
+	ctx context.Context, tx *sql.Tx, scope IdempotencyScope, now time.Time,
+) (controlReceipt, bool, error) {
+	var receipt controlReceipt
+	scopeHash := controlScopeHash(scope)
+	if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtext($1))`, hex.EncodeToString(scopeHash[:])); err != nil {
+		return receipt, false, errors.New("lock postgres Control API idempotency scope")
+	}
+	var storedRequestHash []byte
+	var etag, location sql.NullString
+	var expiresAt time.Time
+	err := tx.QueryRowContext(ctx, `SELECT request_hash, response_status, response_body::text,
+		response_etag, response_location, trace_id, created_at, expires_at
+		FROM sense.control_idempotency_receipts WHERE scope_hash = $1`, scopeHash[:]).
+		Scan(&storedRequestHash, &receipt.Status, &receipt.Body, &etag, &location,
+			&receipt.TraceID, &receipt.CreatedAt, &expiresAt)
+	if errors.Is(err, sql.ErrNoRows) {
+		return receipt, false, nil
+	}
+	if err != nil {
+		return receipt, false, errors.New("read postgres Control API idempotency receipt")
+	}
+	if !expiresAt.After(now) {
+		if _, err := tx.ExecContext(ctx, `DELETE FROM sense.control_idempotency_receipts
+			WHERE scope_hash = $1`, scopeHash[:]); err != nil {
+			return receipt, false, errors.New("expire postgres Control API idempotency receipt")
+		}
+		return controlReceipt{}, false, nil
+	}
+	if subtle.ConstantTimeCompare(storedRequestHash, scope.RequestHash[:]) != 1 {
+		return receipt, false, ErrIdempotencyConflict
+	}
+	receipt.ETag = etag.String
+	receipt.Location = location.String
+	return receipt, true, nil
+}
+
+func writeControlReceipt(
+	ctx context.Context, tx *sql.Tx, scope IdempotencyScope, receipt controlReceipt,
+) error {
+	scopeHash := controlScopeHash(scope)
+	_, err := tx.ExecContext(ctx, `INSERT INTO sense.control_idempotency_receipts(
+		scope_hash, request_hash, operation_name, principal_id, tenant_id, site_id,
+		response_status, response_body, response_etag, response_location, trace_id,
+		created_at, expires_at
+	) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb, NULLIF($9, ''), NULLIF($10, ''), $11, $12, $13)`,
+		scopeHash[:], scope.RequestHash[:], scope.Operation, scope.PrincipalID,
+		scope.TenantID, scope.SiteID, receipt.Status, string(receipt.Body),
+		receipt.ETag, receipt.Location, receipt.TraceID, receipt.CreatedAt,
+		receipt.CreatedAt.Add(controlReceiptTTL))
+	if err != nil {
+		return errors.New("write postgres Control API idempotency receipt")
+	}
+	// Bound opportunistic cleanup; never scans or deletes unexpired receipts.
+	_, _ = tx.ExecContext(ctx, `DELETE FROM sense.control_idempotency_receipts
+		WHERE scope_hash IN (SELECT scope_hash FROM sense.control_idempotency_receipts
+			WHERE expires_at <= $1 ORDER BY expires_at LIMIT 32)`, receipt.CreatedAt)
+	return nil
+}
+
+func (s *Postgres) CreateControlDevice(
+	ctx context.Context, request ControlCreateRequest,
+) (ControlCreateResult, error) {
+	now := time.Now().UTC()
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return ControlCreateResult{}, errors.New("begin postgres Control API device create")
+	}
+	defer tx.Rollback()
+	receipt, found, err := readControlReceipt(ctx, tx, request.Scope, now)
+	if err != nil {
+		return ControlCreateResult{}, err
+	}
+	if found {
+		var value ControlDevice
+		if err := json.Unmarshal(receipt.Body, &value); err != nil {
+			return ControlCreateResult{}, errors.New("decode postgres device creation receipt")
+		}
+		if err := tx.Commit(); err != nil {
+			return ControlCreateResult{}, errors.New("commit postgres device creation replay")
+		}
+		return ControlCreateResult{
+			Device: value, AcceptedAt: receipt.CreatedAt, TraceID: receipt.TraceID,
+			ETag: receipt.ETag, Location: receipt.Location, Replay: true,
+		}, nil
+	}
+	value := request.Device
+	if value.Generation == 0 {
+		value.Generation = 1
+	}
+	if value.ResourceVersion == 0 {
+		value.ResourceVersion = 1
+	}
+	if value.ActualState == "" {
+		value.ActualState = device.ActualPending
+	}
+	value.CreatedAt = now
+	value.UpdatedAt = now
+	if err := createControlDeviceTx(ctx, tx, value, now); err != nil {
+		return ControlCreateResult{}, err
+	}
+	created, err := scanControlDevice(tx.QueryRowContext(ctx, controlDeviceSelect+`
+		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.id = $3`, value.TenantID, value.SiteID, value.ID))
+	if err != nil {
+		return ControlCreateResult{}, errors.New("read created postgres control device")
+	}
+	responseBody, err := json.Marshal(created)
+	if err != nil {
+		return ControlCreateResult{}, errors.New("encode created postgres control device")
+	}
+	etag := DeviceETag(created.ID, created.ResourceVersion)
+	location := "/api/v1/sites/" + url.PathEscape(created.SiteID) + "/devices/" + url.PathEscape(created.ID)
+	receipt = controlReceipt{
+		Status: 201, Body: responseBody, ETag: etag, Location: location,
+		TraceID: request.Scope.TraceID, CreatedAt: now,
+	}
+	if err := writeControlReceipt(ctx, tx, request.Scope, receipt); err != nil {
+		return ControlCreateResult{}, err
+	}
+	if err := tx.Commit(); err != nil {
+		return ControlCreateResult{}, errors.New("commit postgres Control API device create")
+	}
+	return ControlCreateResult{
+		Device: created, AcceptedAt: now, TraceID: receipt.TraceID,
+		ETag: etag, Location: location,
+	}, nil
+}
+
+func createControlDeviceTx(ctx context.Context, tx *sql.Tx, value device.Device, now time.Time) error {
+	if err := value.Validate(); err != nil {
+		return fmt.Errorf("validate Control API device: %w", err)
+	}
+	areaVersion, err := checkPostgresAreaPolicy(
+		ctx, tx, value.TenantID, value.SiteID, value.AreaID,
+		value.HasCapability(device.CapabilityVideoCapture), now,
+	)
+	if err != nil {
+		return err
+	}
+	var quotaVersion int64
+	if value.ConsumesVideoChannel() {
+		quotaVersion, err = checkPostgresVideoQuota(ctx, tx, value.TenantID, value.SiteID, now)
+		if err != nil {
+			return err
+		}
+	}
+	_, err = tx.ExecContext(ctx, `INSERT INTO sense.devices(
+		id, tenant_id, site_id, area_id, serial_number, name, modality,
+		desired_state, actual_state, endpoint_ref, credential_ref, profile_token,
+		path_name, generation, resource_version, quota_source_version,
+		area_policy_source_version, created_at, updated_at
+	) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19)`,
+		value.ID, value.TenantID, value.SiteID, value.AreaID, value.SerialNumber, value.Name,
+		value.Modality, value.DesiredState, value.ActualState, value.EndpointRef,
+		value.CredentialRef, value.ProfileToken, value.PathName, value.Generation,
+		value.ResourceVersion, nullableVersion(quotaVersion), areaVersion,
+		value.CreatedAt, value.UpdatedAt)
+	if err != nil {
+		var postgresError *pgconn.PgError
+		if errors.As(err, &postgresError) && postgresError.Code == "23505" &&
+			strings.Contains(postgresError.ConstraintName, "serial_number") {
+			return ErrDuplicateSerialNumber
+		}
+		return errors.New("insert postgres Control API device")
+	}
+	for _, capability := range sortedCapabilities(value.Capabilities) {
+		if _, err := tx.ExecContext(ctx, `INSERT INTO sense.device_capabilities(device_id, capability)
+			VALUES ($1, $2)`, value.ID, capability); err != nil {
+			return errors.New("insert postgres Control API device capability")
+		}
+	}
+	if _, err := tx.ExecContext(ctx, `INSERT INTO sense.reconcile_state(device_id, updated_at)
+		VALUES ($1, $2)`, value.ID, now); err != nil {
+		return errors.New("insert postgres Control API reconcile state")
+	}
+	return insertPostgresAudit(ctx, tx, postgresAuditEvent{
+		EventType: "device.created", TenantID: value.TenantID, SiteID: value.SiteID,
+		DeviceID: value.ID, Generation: value.Generation,
+		QuotaSourceVersion: quotaVersion, AreaPolicySourceVersion: areaVersion,
+		OccurredAt: now,
+		Payload: map[string]any{
+			"kind": "device_created", "area_id": value.AreaID, "modality": value.Modality,
+			"capabilities": sortedCapabilities(value.Capabilities), "desired_state": value.DesiredState,
+		},
+	})
+}
+
+var _ ControlRepository = (*Postgres)(nil)
+
+func (s *Postgres) PatchControlDevice(
+	ctx context.Context, tenantID, siteID, deviceID, expectedETag string, patch ControlPatch,
+) (ControlMutationResult, error) {
+	now := time.Now().UTC()
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return ControlMutationResult{}, errors.New("begin postgres Control API device patch")
+	}
+	defer tx.Rollback()
+	var name, areaID, endpointRef, credentialRef, profileToken string
+	var generation, resourceVersion int64
+	var desired device.DesiredState
+	var quotaVersion, areaVersion sql.NullInt64
+	var hasVideo bool
+	err = tx.QueryRowContext(ctx, `SELECT d.name, d.area_id, d.endpoint_ref,
+		d.credential_ref, d.profile_token, d.generation, d.resource_version,
+		d.desired_state, d.quota_source_version, d.area_policy_source_version,
+		EXISTS (SELECT 1 FROM sense.device_capabilities c
+			WHERE c.device_id = d.id AND c.capability = 'video_capture')
+		FROM sense.devices d
+		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.id = $3 FOR UPDATE`,
+		tenantID, siteID, deviceID).Scan(
+		&name, &areaID, &endpointRef, &credentialRef, &profileToken,
+		&generation, &resourceVersion, &desired, &quotaVersion, &areaVersion, &hasVideo,
+	)
+	if errors.Is(err, sql.ErrNoRows) {
+		return ControlMutationResult{}, ErrNotFound
+	}
+	if err != nil {
+		return ControlMutationResult{}, errors.New("read postgres Control API device patch state")
+	}
+	if DeviceETag(deviceID, resourceVersion) != expectedETag {
+		return ControlMutationResult{}, ErrETagMismatch
+	}
+	changedFields := make([]string, 0, 5)
+	reconcileChanged := false
+	if patch.Name != nil && *patch.Name != name {
+		name = *patch.Name
+		changedFields = append(changedFields, "name")
+	}
+	if patch.AreaID != nil && *patch.AreaID != areaID {
+		version, policyErr := checkPostgresAreaPolicy(
+			ctx, tx, tenantID, siteID, *patch.AreaID, hasVideo, now,
+		)
+		if policyErr != nil {
+			return ControlMutationResult{}, policyErr
+		}
+		areaID = *patch.AreaID
+		areaVersion = sql.NullInt64{Int64: version, Valid: true}
+		changedFields = append(changedFields, "area_id")
+	}
+	if patch.EndpointRef != nil && *patch.EndpointRef != endpointRef {
+		endpointRef = *patch.EndpointRef
+		changedFields = append(changedFields, "endpoint_ref")
+		reconcileChanged = true
+	}
+	if patch.CredentialRef != nil && *patch.CredentialRef != credentialRef {
+		credentialRef = *patch.CredentialRef
+		changedFields = append(changedFields, "credential_ref")
+		reconcileChanged = true
+	}
+	if patch.ProfileToken != nil && *patch.ProfileToken != profileToken {
+		profileToken = *patch.ProfileToken
+		changedFields = append(changedFields, "profile_token")
+		reconcileChanged = true
+	}
+	if len(changedFields) > 0 {
+		resourceVersion++
+		if reconcileChanged {
+			generation++
+		}
+		_, err = tx.ExecContext(ctx, `UPDATE sense.devices SET
+			name = $1, area_id = $2, endpoint_ref = $3, credential_ref = $4,
+			profile_token = $5, generation = $6, resource_version = $7,
+			area_policy_source_version = $8,
+			actual_state = CASE WHEN $9 THEN 'pending' ELSE actual_state END,
+			updated_at = $10
+			WHERE tenant_id = $11 AND site_id = $12 AND id = $13`,
+			name, areaID, endpointRef, credentialRef, profileToken, generation,
+			resourceVersion, nullableVersion(areaVersion.Int64), reconcileChanged, now,
+			tenantID, siteID, deviceID)
+		if err != nil {
+			return ControlMutationResult{}, errors.New("update postgres Control API device configuration")
+		}
+		if reconcileChanged {
+			if _, err := tx.ExecContext(ctx, `UPDATE sense.reconcile_state SET
+				failure_count = 0, next_attempt_at = NULL, last_error_code = NULL,
+				updated_at = $1 WHERE device_id = $2`, now, deviceID); err != nil {
+				return ControlMutationResult{}, errors.New("reset postgres Control API reconcile state")
+			}
+		}
+	}
+	if err := insertPostgresAudit(ctx, tx, postgresAuditEvent{
+		EventType: "device.configuration.accepted", TenantID: tenantID, SiteID: siteID,
+		DeviceID: deviceID, Generation: generation,
+		QuotaSourceVersion: quotaVersion.Int64, AreaPolicySourceVersion: areaVersion.Int64,
+		OccurredAt: now,
+		Payload: map[string]any{
+			"kind": "configuration_accepted", "changed": len(changedFields) > 0,
+			"changed_fields": changedFields, "area_id": areaID,
+		},
+	}); err != nil {
+		return ControlMutationResult{}, err
+	}
+	updated, err := scanControlDevice(tx.QueryRowContext(ctx, controlDeviceSelect+`
+		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.id = $3`, tenantID, siteID, deviceID))
+	if err != nil {
+		return ControlMutationResult{}, errors.New("read patched postgres control device")
+	}
+	if err := tx.Commit(); err != nil {
+		return ControlMutationResult{}, errors.New("commit postgres Control API device patch")
+	}
+	return ControlMutationResult{
+		Device: updated, AcceptedAt: now, TraceID: auditFromContext(ctx).TraceID,
+		ETag: DeviceETag(updated.ID, updated.ResourceVersion),
+	}, nil
+}
+
+func (s *Postgres) SetControlDesiredState(
+	ctx context.Context, tenantID, siteID, deviceID, expectedETag string, desired device.DesiredState,
+) (ControlMutationResult, error) {
+	now := time.Now().UTC()
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return ControlMutationResult{}, errors.New("begin postgres Control API desired-state update")
+	}
+	defer tx.Rollback()
+	updated, err := setControlDesiredStateTx(
+		ctx, tx, tenantID, siteID, deviceID, expectedETag, desired, now,
+	)
+	if err != nil {
+		return ControlMutationResult{}, err
+	}
+	if err := tx.Commit(); err != nil {
+		return ControlMutationResult{}, errors.New("commit postgres Control API desired-state update")
+	}
+	return ControlMutationResult{
+		Device: updated, AcceptedAt: now, TraceID: auditFromContext(ctx).TraceID,
+		ETag: DeviceETag(updated.ID, updated.ResourceVersion),
+	}, nil
+}
+
+func setControlDesiredStateTx(
+	ctx context.Context, tx *sql.Tx, tenantID, siteID, deviceID, expectedETag string,
+	desired device.DesiredState, now time.Time,
+) (ControlDevice, error) {
+	var areaID, endpointRef, pathName string
+	var current device.DesiredState
+	var generation, resourceVersion int64
+	var quotaVersion, areaVersion sql.NullInt64
+	var hasVideo bool
+	err := tx.QueryRowContext(ctx, `SELECT d.area_id, d.desired_state, d.endpoint_ref,
+		d.path_name, d.generation, d.resource_version, d.quota_source_version,
+		d.area_policy_source_version,
+		EXISTS (SELECT 1 FROM sense.device_capabilities c
+			WHERE c.device_id = d.id AND c.capability = 'video_capture')
+		FROM sense.devices d
+		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.id = $3 FOR UPDATE`,
+		tenantID, siteID, deviceID).Scan(
+		&areaID, &current, &endpointRef, &pathName, &generation, &resourceVersion,
+		&quotaVersion, &areaVersion, &hasVideo,
+	)
+	if errors.Is(err, sql.ErrNoRows) {
+		return ControlDevice{}, ErrNotFound
+	}
+	if err != nil {
+		return ControlDevice{}, errors.New("read postgres Control API desired state")
+	}
+	if DeviceETag(deviceID, resourceVersion) != expectedETag {
+		return ControlDevice{}, ErrETagMismatch
+	}
+	if current != desired {
+		var admittedQuota, admittedArea int64
+		if desired == device.DesiredEnabled && hasVideo {
+			if strings.TrimSpace(endpointRef) == "" || strings.TrimSpace(pathName) == "" {
+				return ControlDevice{}, errors.New("video adapter configuration is incomplete")
+			}
+			admittedArea, err = checkPostgresAreaPolicy(ctx, tx, tenantID, siteID, areaID, true, now)
+			if err != nil {
+				return ControlDevice{}, err
+			}
+			admittedQuota, err = checkPostgresVideoQuota(ctx, tx, tenantID, siteID, now)
+			if err != nil {
+				return ControlDevice{}, err
+			}
+		}
+		generation++
+		resourceVersion++
+		err = tx.QueryRowContext(ctx, `UPDATE sense.devices SET
+			desired_state = $1, actual_state = 'pending', generation = $2,
+			resource_version = $3,
+			quota_source_version = COALESCE($4, quota_source_version),
+			area_policy_source_version = COALESCE($5, area_policy_source_version),
+			updated_at = $6
+			WHERE tenant_id = $7 AND site_id = $8 AND id = $9
+			RETURNING quota_source_version, area_policy_source_version`,
+			desired, generation, resourceVersion, nullableVersion(admittedQuota),
+			nullableVersion(admittedArea), now, tenantID, siteID, deviceID).
+			Scan(&quotaVersion, &areaVersion)
+		if err != nil {
+			return ControlDevice{}, errors.New("update postgres Control API desired state")
+		}
+		if _, err := tx.ExecContext(ctx, `UPDATE sense.reconcile_state SET
+			failure_count = 0, next_attempt_at = NULL, last_error_code = NULL,
+			updated_at = $1 WHERE device_id = $2`, now, deviceID); err != nil {
+			return ControlDevice{}, errors.New("reset postgres Control API desired-state reconciliation")
+		}
+	}
+	if err := insertPostgresAudit(ctx, tx, postgresAuditEvent{
+		EventType: "device.desired_state.accepted", TenantID: tenantID, SiteID: siteID,
+		DeviceID: deviceID, Generation: generation,
+		QuotaSourceVersion: quotaVersion.Int64, AreaPolicySourceVersion: areaVersion.Int64,
+		OccurredAt: now,
+		Payload: map[string]any{
+			"kind": "desired_state_accepted", "previous_desired_state": current,
+			"desired_state": desired, "changed": current != desired,
+		},
+	}); err != nil {
+		return ControlDevice{}, err
+	}
+	value, err := scanControlDevice(tx.QueryRowContext(ctx, controlDeviceSelect+`
+		WHERE d.tenant_id = $1 AND d.site_id = $2 AND d.id = $3`, tenantID, siteID, deviceID))
+	if err != nil {
+		return ControlDevice{}, errors.New("read updated postgres control desired state")
+	}
+	return value, nil
+}
+
+func (s *Postgres) BatchSetControlDesiredState(
+	ctx context.Context, request ControlBatchRequest,
+) (ControlBatchOperation, error) {
+	now := time.Now().UTC()
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return ControlBatchOperation{}, errors.New("begin postgres Control API batch")
+	}
+	defer tx.Rollback()
+	receipt, found, err := readControlReceipt(ctx, tx, request.Scope, now)
+	if err != nil {
+		return ControlBatchOperation{}, err
+	}
+	if found {
+		var operation ControlBatchOperation
+		if err := json.Unmarshal(receipt.Body, &operation); err != nil {
+			return ControlBatchOperation{}, errors.New("decode postgres Control API batch receipt")
+		}
+		operation.TenantID = request.Scope.TenantID
+		operation.SiteID = request.Scope.SiteID
+		operation.Replay = true
+		if err := tx.Commit(); err != nil {
+			return ControlBatchOperation{}, errors.New("commit postgres Control API batch replay")
+		}
+		return operation, nil
+	}
+	operationID, err := newControlOperationID(now)
+	if err != nil {
+		return ControlBatchOperation{}, err
+	}
+	counts := make(map[string]int, len(request.Items))
+	for _, item := range request.Items {
+		counts[item.DeviceID]++
+	}
+	if err := lockControlBatchDevices(
+		ctx, tx, request.Scope.TenantID, request.Scope.SiteID, counts,
+	); err != nil {
+		return ControlBatchOperation{}, err
+	}
+	results := make([]ControlBatchItemResult, 0, len(request.Items))
+	succeeded := 0
+	for index, item := range request.Items {
+		if counts[item.DeviceID] > 1 {
+			code, message := "invalid_request", "device_id is duplicated in this request"
+			results = append(results, ControlBatchItemResult{
+				DeviceID: item.DeviceID, Status: "rejected", ErrorCode: &code, Message: &message,
+			})
+			continue
+		}
+		savepoint := fmt.Sprintf("control_batch_%d", index)
+		if _, err := tx.ExecContext(ctx, "SAVEPOINT "+savepoint); err != nil {
+			return ControlBatchOperation{}, errors.New("create postgres Control API batch savepoint")
+		}
+		updated, itemErr := setControlDesiredStateTx(
+			ctx, tx, request.Scope.TenantID, request.Scope.SiteID,
+			item.DeviceID, item.ETag, item.DesiredState, now,
+		)
+		if itemErr != nil {
+			if _, rollbackErr := tx.ExecContext(ctx, "ROLLBACK TO SAVEPOINT "+savepoint); rollbackErr != nil {
+				return ControlBatchOperation{}, errors.New("rollback postgres Control API batch item")
+			}
+			code, status, message := controlBatchError(itemErr)
+			results = append(results, ControlBatchItemResult{
+				DeviceID: item.DeviceID, Status: status, ErrorCode: &code, Message: &message,
+			})
+		} else {
+			generation := updated.Generation
+			results = append(results, ControlBatchItemResult{
+				DeviceID: item.DeviceID, Status: "succeeded", Generation: &generation,
+			})
+			succeeded++
+		}
+		if _, err := tx.ExecContext(ctx, "RELEASE SAVEPOINT "+savepoint); err != nil {
+			return ControlBatchOperation{}, errors.New("release postgres Control API batch savepoint")
+		}
+	}
+	status := "partially_succeeded"
+	if succeeded == len(results) {
+		status = "succeeded"
+	} else if succeeded == 0 {
+		status = "failed"
+	}
+	completedAt := now
+	operation := ControlBatchOperation{
+		ID: operationID, TenantID: request.Scope.TenantID, SiteID: request.Scope.SiteID,
+		Status: status, SubmittedAt: now, CompletedAt: &completedAt,
+		Results: results, TraceID: request.Scope.TraceID,
+	}
+	if _, err := tx.ExecContext(ctx, `INSERT INTO sense.batch_operations(
+		id, tenant_id, site_id, principal_id, status, trace_id, submitted_at, completed_at
+	) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)`, operation.ID, operation.TenantID,
+		operation.SiteID, request.Scope.PrincipalID, operation.Status, operation.TraceID,
+		operation.SubmittedAt, operation.CompletedAt); err != nil {
+		return ControlBatchOperation{}, errors.New("insert postgres Control API batch operation")
+	}
+	for index, result := range results {
+		if _, err := tx.ExecContext(ctx, `INSERT INTO sense.batch_operation_items(
+			operation_id, ordinal, device_id, status, error_code, message, generation
+		) VALUES ($1,$2,$3,$4,$5,$6,$7)`, operation.ID, index, result.DeviceID,
+			result.Status, result.ErrorCode, result.Message, result.Generation); err != nil {
+			return ControlBatchOperation{}, errors.New("insert postgres Control API batch item")
+		}
+	}
+	body, err := json.Marshal(operation)
+	if err != nil {
+		return ControlBatchOperation{}, errors.New("encode postgres Control API batch operation")
+	}
+	receipt = controlReceipt{
+		Status: 202, Body: body,
+		Location: "/api/v1/operations/" + url.PathEscape(operation.ID),
+		TraceID:  operation.TraceID, CreatedAt: now,
+	}
+	if err := writeControlReceipt(ctx, tx, request.Scope, receipt); err != nil {
+		return ControlBatchOperation{}, err
+	}
+	if err := tx.Commit(); err != nil {
+		return ControlBatchOperation{}, errors.New("commit postgres Control API batch")
+	}
+	return operation, nil
+}
+
+func lockControlBatchDevices(
+	ctx context.Context, tx *sql.Tx, tenantID, siteID string, deviceIDs map[string]int,
+) error {
+	ordered := make([]string, 0, len(deviceIDs))
+	for deviceID := range deviceIDs {
+		ordered = append(ordered, deviceID)
+	}
+	sort.Strings(ordered)
+	for _, deviceID := range ordered {
+		var locked string
+		err := tx.QueryRowContext(ctx, `SELECT id FROM sense.devices
+			WHERE tenant_id = $1 AND site_id = $2 AND id = $3 FOR UPDATE`,
+			tenantID, siteID, deviceID).Scan(&locked)
+		if err != nil && !errors.Is(err, sql.ErrNoRows) {
+			return errors.New("lock postgres Control API batch devices")
+		}
+	}
+	return nil
+}
+
+func controlBatchError(err error) (code, status, message string) {
+	switch {
+	case errors.Is(err, ErrNotFound):
+		return "not_found", "rejected", "device was not found"
+	case errors.Is(err, ErrETagMismatch):
+		return "etag_mismatch", "rejected", "device ETag does not match"
+	case errors.Is(err, ErrAreaPolicyDenied):
+		return "area_policy_denied", "rejected", "Area policy denies this change"
+	case errors.Is(err, ErrAreaPolicyUnavailable), errors.Is(err, ErrAreaPolicyInvalid):
+		return "area_policy_unavailable", "failed", "Area policy is unavailable"
+	case errors.Is(err, ErrQuotaProjectionUnavailable):
+		return "quota_projection_unavailable", "failed", "Site quota is unavailable"
+	case errors.Is(err, ErrQuotaProjectionInvalid):
+		return "quota_projection_invalid", "failed", "Site quota is invalid"
+	}
+	var quotaError *device.QuotaExceededError
+	if errors.As(err, &quotaError) {
+		return "quota_exceeded", "rejected", "Site video channel quota is exceeded"
+	}
+	return "service_unavailable", "failed", "device change could not be accepted"
+}
+
+func (s *Postgres) GetControlOperation(
+	ctx context.Context, tenantID, operationID string,
+) (ControlBatchOperation, error) {
+	var value ControlBatchOperation
+	var completedAt sql.NullTime
+	err := s.db.QueryRowContext(ctx, `SELECT id, tenant_id, site_id, status,
+		submitted_at, completed_at, trace_id
+		FROM sense.batch_operations WHERE tenant_id = $1 AND id = $2`, tenantID, operationID).
+		Scan(&value.ID, &value.TenantID, &value.SiteID, &value.Status,
+			&value.SubmittedAt, &completedAt, &value.TraceID)
+	if errors.Is(err, sql.ErrNoRows) {
+		return ControlBatchOperation{}, ErrNotFound
+	}
+	if err != nil {
+		return ControlBatchOperation{}, errors.New("read postgres Control API batch operation")
+	}
+	if completedAt.Valid {
+		point := completedAt.Time.UTC()
+		value.CompletedAt = &point
+	}
+	rows, err := s.db.QueryContext(ctx, `SELECT device_id, status, error_code, message, generation
+		FROM sense.batch_operation_items WHERE operation_id = $1 ORDER BY ordinal`, operationID)
+	if err != nil {
+		return ControlBatchOperation{}, errors.New("list postgres Control API batch results")
+	}
+	defer rows.Close()
+	value.Results = make([]ControlBatchItemResult, 0)
+	for rows.Next() {
+		var item ControlBatchItemResult
+		var code, message sql.NullString
+		var generation sql.NullInt64
+		if err := rows.Scan(&item.DeviceID, &item.Status, &code, &message, &generation); err != nil {
+			return ControlBatchOperation{}, errors.New("scan postgres Control API batch result")
+		}
+		if code.Valid {
+			item.ErrorCode = &code.String
+		}
+		if message.Valid {
+			item.Message = &message.String
+		}
+		if generation.Valid {
+			item.Generation = &generation.Int64
+		}
+		value.Results = append(value.Results, item)
+	}
+	if err := rows.Err(); err != nil {
+		return ControlBatchOperation{}, errors.New("iterate postgres Control API batch results")
+	}
+	return value, nil
+}
+
+func newControlOperationID(now time.Time) (string, error) {
+	value := make([]byte, 16)
+	milliseconds := uint64(now.UTC().UnixMilli())
+	value[0], value[1], value[2] = byte(milliseconds>>40), byte(milliseconds>>32), byte(milliseconds>>24)
+	value[3], value[4], value[5] = byte(milliseconds>>16), byte(milliseconds>>8), byte(milliseconds)
+	if _, err := rand.Read(value[6:]); err != nil {
+		return "", errors.New("generate Control API operation ID")
+	}
+	number := new(big.Int).SetBytes(value)
+	base, remainder := big.NewInt(32), new(big.Int)
+	const alphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+	encoded := make([]byte, 26)
+	for index := len(encoded) - 1; index >= 0; index-- {
+		number.QuoRem(number, base, remainder)
+		encoded[index] = alphabet[remainder.Int64()]
+	}
+	return "op_" + string(encoded), nil
+}

Sense/internal/store/postgres.go

@@ -53,8 +53,8 @@ func (s *Postgres) Close() error {
 func (s *Postgres) verifySchemaAndPrivileges(ctx context.Context) error {
 	var version sql.NullInt64
 	if err := s.db.QueryRowContext(ctx,
-		`SELECT MAX(version) FROM sense.schema_migrations`).Scan(&version); err != nil || !version.Valid || version.Int64 < 3 {
-		return errors.New("postgres sense schema migration v3 is required")
+		`SELECT MAX(version) FROM sense.schema_migrations`).Scan(&version); err != nil || !version.Valid || version.Int64 < 4 {
+		return errors.New("postgres sense schema migration v4 is required")
 	}
 	var canReadQuotaView, canWriteQuotaView, canReadSiteSource, canWriteSiteSource bool
 	var canReadAreaView, canWriteAreaView, canReadAreaSource, canWriteAreaSource bool
@@ -77,6 +77,24 @@ func (s *Postgres) verifySchemaAndPrivileges(ctx context.Context) error {
 		!canReadAreaView || canWriteAreaView || canReadAreaSource || canWriteAreaSource {
 		return errors.New("postgres role violates Bell projection privilege boundary")
 	}
+	var canUseReceipts, canUseOperations, canUseOperationItems bool
+	var publicReceipts, publicOperations, publicOperationItems bool
+	if err := s.db.QueryRowContext(ctx, `SELECT
+		has_table_privilege(current_user, 'sense.control_idempotency_receipts', 'SELECT,INSERT,UPDATE,DELETE'),
+		has_table_privilege(current_user, 'sense.batch_operations', 'SELECT,INSERT,UPDATE,DELETE'),
+		has_table_privilege(current_user, 'sense.batch_operation_items', 'SELECT,INSERT,UPDATE,DELETE'),
+		has_table_privilege('public', 'sense.control_idempotency_receipts', 'SELECT,INSERT,UPDATE,DELETE'),
+		has_table_privilege('public', 'sense.batch_operations', 'SELECT,INSERT,UPDATE,DELETE'),
+		has_table_privilege('public', 'sense.batch_operation_items', 'SELECT,INSERT,UPDATE,DELETE')`).Scan(
+		&canUseReceipts, &canUseOperations, &canUseOperationItems,
+		&publicReceipts, &publicOperations, &publicOperationItems,
+	); err != nil {
+		return errors.New("verify postgres Control API state privileges")
+	}
+	if !canUseReceipts || !canUseOperations || !canUseOperationItems ||
+		publicReceipts || publicOperations || publicOperationItems {
+		return errors.New("postgres role violates Control API state privilege boundary")
+	}
 	return nil
 }
 
@@ -87,6 +105,9 @@ func (s *Postgres) CreateDevice(ctx context.Context, value device.Device) error
 	if value.ActualState == "" {
 		value.ActualState = device.ActualPending
 	}
+	if value.ResourceVersion == 0 {
+		value.ResourceVersion = 1
+	}
 	if err := value.Validate(); err != nil {
 		return fmt.Errorf("validate device: %w", err)
 	}
@@ -118,14 +139,15 @@ func (s *Postgres) CreateDevice(ctx context.Context, value device.Device) error
 	}
 	_, err = tx.ExecContext(ctx, `INSERT INTO sense.devices(
 		id, tenant_id, site_id, area_id, serial_number, name, modality,
-		desired_state, actual_state, endpoint_ref, credential_ref,
-		path_name, generation, quota_source_version, area_policy_source_version,
+		desired_state, actual_state, endpoint_ref, credential_ref, profile_token,
+		path_name, generation, resource_version, quota_source_version, area_policy_source_version,
 		created_at, updated_at
-	) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)`,
+	) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)`,
 		value.ID, value.TenantID, value.SiteID, value.AreaID, value.SerialNumber, value.Name,
 		value.Modality, value.DesiredState, value.ActualState, value.EndpointRef,
-		value.CredentialRef, value.PathName, value.Generation, nullableVersion(quotaVersion),
-		areaVersion, value.CreatedAt, value.UpdatedAt)
+		value.CredentialRef, value.ProfileToken, value.PathName, value.Generation,
+		value.ResourceVersion, nullableVersion(quotaVersion), areaVersion,
+		value.CreatedAt, value.UpdatedAt)
 	if err != nil {
 		return errors.New("insert postgres device")
 	}
@@ -354,6 +376,7 @@ func (s *Postgres) SetDesiredState(ctx context.Context, id string, desired devic
 	var updatedQuotaVersion, updatedAreaVersion sql.NullInt64
 	err = tx.QueryRowContext(ctx, `UPDATE sense.devices SET
 		desired_state = $1, actual_state = 'pending', generation = generation + 1,
+		resource_version = resource_version + 1,
 		quota_source_version = COALESCE($2, quota_source_version),
 		area_policy_source_version = COALESCE($3, area_policy_source_version),
 		updated_at = $4
@@ -413,10 +436,10 @@ func (s *Postgres) ListDueReconcile(ctx context.Context, now time.Time, limit in
 	rows, err := s.db.QueryContext(ctx, `SELECT `+postgresDeviceColumns+`, r.failure_count, r.next_attempt_at
 		FROM sense.devices d
 		JOIN sense.reconcile_state r ON r.device_id = d.id
-		WHERE d.desired_state = 'enabled'
-			AND EXISTS (SELECT 1 FROM sense.device_capabilities c
+		WHERE EXISTS (SELECT 1 FROM sense.device_capabilities c
 				WHERE c.device_id = d.id AND c.capability = 'video_capture')
-			AND (r.observed_generation < d.generation OR r.failure_count > 0)
+			AND (r.observed_generation < d.generation
+				OR (d.desired_state = 'enabled' AND r.failure_count > 0))
 			AND (r.next_attempt_at IS NULL OR r.next_attempt_at <= $1)
 		ORDER BY d.updated_at, d.id LIMIT $2`, now, limit)
 	if err != nil {
@@ -434,7 +457,8 @@ func (s *Postgres) ListDueReconcile(ctx context.Context, now time.Time, limit in
 			&areaID, &candidate.Device.SerialNumber, &candidate.Device.Name, &candidate.Device.Modality,
 			&candidate.Device.DesiredState, &candidate.Device.ActualState,
 			&candidate.Device.EndpointRef, &candidate.Device.CredentialRef,
-			&candidate.Device.PathName, &candidate.Device.Generation,
+			&candidate.Device.ProfileToken, &candidate.Device.PathName,
+			&candidate.Device.Generation, &candidate.Device.ResourceVersion,
 			&quotaVersion, &areaVersion,
 			&candidate.Device.CreatedAt, &candidate.Device.UpdatedAt,
 			&candidate.FailureCount, &nextAttempt,
@@ -511,7 +535,8 @@ func (s *Postgres) MarkReconciled(ctx context.Context, id string, generation int
 		return ErrNotFound
 	}
 	if _, err := tx.ExecContext(ctx, `UPDATE sense.devices
-		SET actual_state = 'pending', updated_at = $1 WHERE id = $2`, now, id); err != nil {
+		SET actual_state = CASE WHEN desired_state = 'disabled' THEN 'offline' ELSE 'pending' END,
+		updated_at = $1 WHERE id = $2`, now, id); err != nil {
 		return errors.New("mark postgres reconciled device pending")
 	}
 	if err := tx.Commit(); err != nil {
@@ -628,8 +653,8 @@ func (s *Postgres) ConvergenceSnapshot(ctx context.Context) (ConvergenceSnapshot
 }
 
 const postgresDeviceColumns = `d.id, d.tenant_id, d.site_id, d.area_id, d.serial_number, d.name, d.modality,
-		d.desired_state, d.actual_state, d.endpoint_ref, d.credential_ref,
-		d.path_name, d.generation, d.quota_source_version, d.area_policy_source_version,
+		d.desired_state, d.actual_state, d.endpoint_ref, d.credential_ref, d.profile_token,
+		d.path_name, d.generation, d.resource_version, d.quota_source_version, d.area_policy_source_version,
 		d.created_at, d.updated_at`
 
 const postgresDeviceSelect = `SELECT ` + postgresDeviceColumns + ` FROM sense.devices d`
@@ -641,8 +666,9 @@ func scanPostgresDevice(row scanner) (device.Device, error) {
 	err := row.Scan(
 		&value.ID, &value.TenantID, &value.SiteID, &areaID, &value.SerialNumber,
 		&value.Name, &value.Modality, &value.DesiredState, &value.ActualState,
-		&value.EndpointRef, &value.CredentialRef, &value.PathName,
-		&value.Generation, &quotaVersion, &areaVersion, &value.CreatedAt, &value.UpdatedAt,
+		&value.EndpointRef, &value.CredentialRef, &value.ProfileToken, &value.PathName,
+		&value.Generation, &value.ResourceVersion, &quotaVersion, &areaVersion,
+		&value.CreatedAt, &value.UpdatedAt,
 	)
 	value.AreaID = areaID.String
 	value.QuotaSourceVersion = quotaVersion.Int64

Sense/internal/store/postgres_test.go

@@ -2,6 +2,7 @@ package store
 
 import (
 	"context"
+	"crypto/sha256"
 	"database/sql"
 	"errors"
 	"fmt"
@@ -286,12 +287,12 @@ func TestPostgresOpenRejectsOverprivilegedRuntimeRole(t *testing.T) {
 	_, admin := openPostgresTestStore(t)
 	ctx := context.Background()
 	if _, err := admin.ExecContext(ctx,
-		`GRANT UPDATE ON bell.site_quota_v1 TO yovision_t010_sense`); err != nil {
+		`GRANT UPDATE ON bell.site_quota_v1 TO yovision_t011_sense`); err != nil {
 		t.Fatal(err)
 	}
 	defer func() {
 		_, _ = admin.ExecContext(context.Background(),
-			`REVOKE UPDATE ON bell.site_quota_v1 FROM yovision_t010_sense`)
+			`REVOKE UPDATE ON bell.site_quota_v1 FROM yovision_t011_sense`)
 	}()
 	value, err := OpenPostgres(ctx, os.Getenv(postgresTestDSNEnv))
 	if value != nil {
@@ -596,12 +597,12 @@ func TestPostgresOpenRejectsAreaSourcePrivilege(t *testing.T) {
 	_, admin := openPostgresTestStore(t)
 	ctx := context.Background()
 	if _, err := admin.ExecContext(ctx,
-		`GRANT SELECT ON bell.areas TO yovision_t010_sense`); err != nil {
+		`GRANT SELECT ON bell.areas TO yovision_t011_sense`); err != nil {
 		t.Fatal(err)
 	}
 	defer func() {
 		_, _ = admin.ExecContext(context.Background(),
-			`REVOKE SELECT ON bell.areas FROM yovision_t010_sense`)
+			`REVOKE SELECT ON bell.areas FROM yovision_t011_sense`)
 	}()
 	value, err := OpenPostgres(ctx, os.Getenv(postgresTestDSNEnv))
 	if value != nil {
@@ -613,6 +614,361 @@ func TestPostgresOpenRejectsAreaSourcePrivilege(t *testing.T) {
 	}
 }
 
+func TestPostgresOpenRejectsPublicControlStatePrivilege(t *testing.T) {
+	_, admin := openPostgresTestStore(t)
+	ctx := context.Background()
+	if _, err := admin.ExecContext(ctx, `GRANT SELECT ON sense.batch_operations TO PUBLIC`); err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		_, _ = admin.ExecContext(context.Background(), `REVOKE SELECT ON sense.batch_operations FROM PUBLIC`)
+	}()
+	value, err := OpenPostgres(ctx, os.Getenv(postgresTestDSNEnv))
+	if value != nil {
+		_ = value.Close()
+		t.Fatal("PUBLIC Control API table privilege was accepted")
+	}
+	if err == nil || !strings.Contains(err.Error(), "Control API state privilege boundary") {
+		t.Fatalf("expected Control API privilege-boundary error, got %v", err)
+	}
+}
+
+func TestPostgresControlCreateIdempotencyAndRedactedSnapshot(t *testing.T) {
+	postgres, admin := openPostgresTestStore(t)
+	insertBellSite(t, admin, "tenant", "site", 2)
+	value := videoDevice(1, "tenant", "site")
+	value.ID = "dev_control_create_1"
+	value.DesiredState = device.DesiredDisabled
+	hash := sha256.Sum256([]byte("canonical-create"))
+	ctx := WithAuditContext(context.Background(), AuditContext{
+		ActorType: AuditActorService, ActorID: "bell-control", TraceID: "trace-control-create",
+	})
+	request := ControlCreateRequest{
+		Scope: IdempotencyScope{
+			PrincipalID: "bell-control", TenantID: "tenant", SiteID: "site",
+			Operation: "createDevice", Key: "create-control-0001",
+			RequestHash: hash, TraceID: "trace-control-create",
+		},
+		Device: value,
+	}
+	first, err := postgres.CreateControlDevice(ctx, request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	request.Device.ID = "dev_control_create_retry"
+	replayed, err := postgres.CreateControlDevice(ctx, request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !replayed.Replay || replayed.Device.ID != first.Device.ID || replayed.TraceID != first.TraceID {
+		t.Fatalf("create replay drifted: first=%+v replay=%+v", first, replayed)
+	}
+	conflictHash := sha256.Sum256([]byte("different-create"))
+	request.Scope.RequestHash = conflictHash
+	if _, err := postgres.CreateControlDevice(ctx, request); !errors.Is(err, ErrIdempotencyConflict) {
+		t.Fatalf("same key with different body was not rejected: %v", err)
+	}
+	var devices, receipts int
+	var responseBody string
+	if err := admin.QueryRow(`SELECT count(*) FROM sense.devices`).Scan(&devices); err != nil {
+		t.Fatal(err)
+	}
+	if err := admin.QueryRow(`SELECT count(*), min(response_body::text)
+		FROM sense.control_idempotency_receipts`).Scan(&receipts, &responseBody); err != nil {
+		t.Fatal(err)
+	}
+	if devices != 1 || receipts != 1 {
+		t.Fatalf("idempotent create counts drifted: devices=%d receipts=%d", devices, receipts)
+	}
+	for _, forbidden := range []string{value.EndpointRef, value.CredentialRef, "profile_token", "path_name"} {
+		if strings.Contains(responseBody, forbidden) {
+			t.Fatalf("idempotency response snapshot leaked %q", forbidden)
+		}
+	}
+}
+
+func TestPostgresControlListIsStableFilteredAndTenantScoped(t *testing.T) {
+	postgres, admin := openPostgresTestStore(t)
+	insertBellSite(t, admin, "tenant", "site", 4)
+	insertBellSite(t, admin, "other", "site", 4)
+	for index := 1; index <= 3; index++ {
+		value := videoDevice(index, "tenant", "site")
+		value.ID = fmt.Sprintf("dev_list_%d", index)
+		value.DesiredState = device.DesiredDisabled
+		if index == 3 {
+			value.DesiredState = device.DesiredEnabled
+		}
+		if err := postgres.CreateDevice(context.Background(), value); err != nil {
+			t.Fatal(err)
+		}
+	}
+	other := videoDevice(9, "other", "site")
+	other.ID, other.DesiredState = "dev_list_other", device.DesiredDisabled
+	if err := postgres.CreateDevice(context.Background(), other); err != nil {
+		t.Fatal(err)
+	}
+	first, err := postgres.ListControlDevices(context.Background(), "tenant", "site", ControlListFilter{Limit: 1})
+	if err != nil || len(first.Items) != 1 || !first.HasMore || first.Quota.UsedVideoChannels != 1 {
+		t.Fatalf("unexpected first page: %+v %v", first, err)
+	}
+	second, err := postgres.ListControlDevices(context.Background(), "tenant", "site", ControlListFilter{
+		Limit: 1, AfterCreated: &first.Items[0].CreatedAt, AfterDeviceID: first.Items[0].ID,
+	})
+	if err != nil || len(second.Items) != 1 || second.Items[0].ID == first.Items[0].ID {
+		t.Fatalf("stable cursor position failed: %+v %v", second, err)
+	}
+	desired := device.DesiredEnabled
+	filtered, err := postgres.ListControlDevices(context.Background(), "tenant", "site", ControlListFilter{
+		Limit: 100, DesiredState: &desired,
+	})
+	if err != nil || len(filtered.Items) != 1 || filtered.Items[0].ID != "dev_list_3" {
+		t.Fatalf("desired-state filter or tenant scope failed: %+v %v", filtered, err)
+	}
+	if _, err := postgres.ListControlDevices(context.Background(), "tenant", "missing", ControlListFilter{Limit: 50}); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("missing Site did not return not found: %v", err)
+	}
+}
+
+func TestPostgresControlPatchUsesETagAndAuditV2(t *testing.T) {
+	postgres, admin := openPostgresTestStore(t)
+	insertBellSite(t, admin, "tenant", "site", 2)
+	value := videoDevice(1, "tenant", "site")
+	value.ID = "dev_control_patch_1"
+	value.DesiredState = device.DesiredDisabled
+	if err := postgres.CreateDevice(context.Background(), value); err != nil {
+		t.Fatal(err)
+	}
+	current, err := postgres.GetControlDevice(context.Background(), "tenant", "site", value.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	etag := DeviceETag(current.ID, current.ResourceVersion)
+	newName, newProfile := "Updated camera", "profile-main"
+	ctx := WithAuditContext(context.Background(), AuditContext{
+		ActorType: AuditActorUser, ActorID: "operator-1", TraceID: "trace-control-patch",
+	})
+	updated, err := postgres.PatchControlDevice(ctx, "tenant", "site", value.ID, etag, ControlPatch{
+		Name: &newName, ProfileToken: &newProfile,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if updated.Device.ResourceVersion != current.ResourceVersion+1 ||
+		updated.Device.Generation != current.Generation+1 || updated.ETag == etag {
+		t.Fatalf("patch did not advance versions: before=%+v after=%+v", current, updated)
+	}
+	if _, err := postgres.PatchControlDevice(ctx, "tenant", "site", value.ID, etag, ControlPatch{Name: &newName}); !errors.Is(err, ErrETagMismatch) {
+		t.Fatalf("stale ETag was accepted: %v", err)
+	}
+	var profileToken, eventType, payload string
+	if err := admin.QueryRow(`SELECT profile_token FROM sense.devices WHERE id = $1`, value.ID).Scan(&profileToken); err != nil {
+		t.Fatal(err)
+	}
+	if err := admin.QueryRow(`SELECT event_type, payload::text
+		FROM sense.device_operation_outbox WHERE event_type = 'device.configuration.accepted'`).
+		Scan(&eventType, &payload); err != nil {
+		t.Fatal(err)
+	}
+	if profileToken != newProfile || eventType != "device.configuration.accepted" ||
+		strings.Contains(payload, newProfile) || !strings.Contains(payload, "profile_token") {
+		t.Fatalf("configuration persistence/audit mismatch: profile=%q event=%q payload=%s", profileToken, eventType, payload)
+	}
+}
+
+func TestPostgresControlBatchIsPerItemDurableAndReplayable(t *testing.T) {
+	postgres, admin := openPostgresTestStore(t)
+	insertBellSite(t, admin, "tenant", "site", 4)
+	first := videoDevice(1, "tenant", "site")
+	first.ID, first.DesiredState = "dev_batch_1", device.DesiredDisabled
+	second := videoDevice(2, "tenant", "site")
+	second.ID, second.DesiredState = "dev_batch_2", device.DesiredDisabled
+	for _, value := range []device.Device{first, second} {
+		if err := postgres.CreateDevice(context.Background(), value); err != nil {
+			t.Fatal(err)
+		}
+	}
+	hash := sha256.Sum256([]byte("canonical-batch"))
+	ctx := WithAuditContext(context.Background(), AuditContext{
+		ActorType: AuditActorUser, ActorID: "operator-1", Reason: "approved", TraceID: "trace-batch",
+	})
+	request := ControlBatchRequest{
+		Scope: IdempotencyScope{
+			PrincipalID: "operator-1", TenantID: "tenant", SiteID: "site",
+			Operation: "batchSetDeviceDesiredState", Key: "batch-control-0001",
+			RequestHash: hash, TraceID: "trace-batch",
+		},
+		Reason: "approved",
+		Items: []ControlBatchItem{
+			{DeviceID: first.ID, ETag: DeviceETag(first.ID, 1), DesiredState: device.DesiredEnabled},
+			{DeviceID: first.ID, ETag: DeviceETag(first.ID, 1), DesiredState: device.DesiredEnabled},
+			{DeviceID: second.ID, ETag: DeviceETag(second.ID, 1), DesiredState: device.DesiredEnabled},
+		},
+	}
+	operation, err := postgres.BatchSetControlDesiredState(ctx, request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if operation.Status != "partially_succeeded" || len(operation.Results) != 3 ||
+		operation.Results[0].Status != "rejected" || operation.Results[1].Status != "rejected" ||
+		operation.Results[2].Status != "succeeded" {
+		t.Fatalf("unexpected batch result: %+v", operation)
+	}
+	replayed, err := postgres.BatchSetControlDesiredState(ctx, request)
+	if err != nil || !replayed.Replay || replayed.ID != operation.ID {
+		t.Fatalf("batch replay drifted: %+v %v", replayed, err)
+	}
+	read, err := postgres.GetControlOperation(context.Background(), "tenant", operation.ID)
+	if err != nil || read.SiteID != "site" || len(read.Results) != 3 {
+		t.Fatalf("stored operation could not be read: %+v %v", read, err)
+	}
+	if _, err := postgres.GetControlOperation(context.Background(), "other-tenant", operation.ID); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("cross-tenant operation was visible: %v", err)
+	}
+	var enabled, operations, receipts int
+	if err := admin.QueryRow(`SELECT count(*) FROM sense.devices WHERE desired_state = 'enabled'`).Scan(&enabled); err != nil {
+		t.Fatal(err)
+	}
+	if err := admin.QueryRow(`SELECT count(*) FROM sense.batch_operations`).Scan(&operations); err != nil {
+		t.Fatal(err)
+	}
+	if err := admin.QueryRow(`SELECT count(*) FROM sense.control_idempotency_receipts`).Scan(&receipts); err != nil {
+		t.Fatal(err)
+	}
+	if enabled != 1 || operations != 1 || receipts != 1 {
+		t.Fatalf("batch durability counts drifted: enabled=%d operations=%d receipts=%d", enabled, operations, receipts)
+	}
+}
+
+func TestPostgresConcurrentControlCreateExecutesOnce(t *testing.T) {
+	postgres, admin := openPostgresTestStore(t)
+	insertBellSite(t, admin, "tenant", "site", 2)
+	hash := sha256.Sum256([]byte("concurrent-create"))
+	start := make(chan struct{})
+	results := make(chan ControlCreateResult, 2)
+	errorsFound := make(chan error, 2)
+	var wait sync.WaitGroup
+	for index := 1; index <= 2; index++ {
+		wait.Add(1)
+		go func(index int) {
+			defer wait.Done()
+			value := videoDevice(index, "tenant", "site")
+			value.ID = fmt.Sprintf("dev_concurrent_%d", index)
+			value.SerialNumber = "same-semantic-serial"
+			value.DesiredState = device.DesiredDisabled
+			request := ControlCreateRequest{
+				Scope: IdempotencyScope{
+					PrincipalID: "service", TenantID: "tenant", SiteID: "site",
+					Operation: "createDevice", Key: "concurrent-create-0001",
+					RequestHash: hash, TraceID: fmt.Sprintf("trace-concurrent-%d", index),
+				}, Device: value,
+			}
+			<-start
+			result, err := postgres.CreateControlDevice(context.Background(), request)
+			if err != nil {
+				errorsFound <- err
+				return
+			}
+			results <- result
+		}(index)
+	}
+	close(start)
+	wait.Wait()
+	close(results)
+	close(errorsFound)
+	for err := range errorsFound {
+		t.Fatalf("concurrent idempotent create failed: %v", err)
+	}
+	ids := make(map[string]struct{})
+	for result := range results {
+		ids[result.Device.ID] = struct{}{}
+	}
+	if len(ids) != 1 {
+		t.Fatalf("concurrent create returned multiple resources: %+v", ids)
+	}
+	var devices, audits, receipts int
+	if err := admin.QueryRow(`SELECT count(*) FROM sense.devices`).Scan(&devices); err != nil {
+		t.Fatal(err)
+	}
+	if err := admin.QueryRow(`SELECT count(*) FROM sense.device_operation_outbox`).Scan(&audits); err != nil {
+		t.Fatal(err)
+	}
+	if err := admin.QueryRow(`SELECT count(*) FROM sense.control_idempotency_receipts`).Scan(&receipts); err != nil {
+		t.Fatal(err)
+	}
+	if devices != 1 || audits != 1 || receipts != 1 {
+		t.Fatalf("concurrent create executed more than once: devices=%d audits=%d receipts=%d", devices, audits, receipts)
+	}
+}
+
+func TestPostgresConcurrentControlBatchesUseStableDeviceLockOrder(t *testing.T) {
+	postgres, admin := openPostgresTestStore(t)
+	insertBellSite(t, admin, "tenant", "site", 4)
+	ids := []string{"dev_lock_a", "dev_lock_b"}
+	for index, id := range ids {
+		value := videoDevice(index+1, "tenant", "site")
+		value.ID, value.DesiredState = id, device.DesiredDisabled
+		if err := postgres.CreateDevice(context.Background(), value); err != nil {
+			t.Fatal(err)
+		}
+	}
+	start := make(chan struct{})
+	operations := make(chan ControlBatchOperation, 2)
+	errorsFound := make(chan error, 2)
+	var wait sync.WaitGroup
+	for index := 0; index < 2; index++ {
+		wait.Add(1)
+		go func(index int) {
+			defer wait.Done()
+			order := ids
+			if index == 1 {
+				order = []string{ids[1], ids[0]}
+			}
+			hash := sha256.Sum256([]byte(fmt.Sprintf("batch-order-%d", index)))
+			request := ControlBatchRequest{
+				Scope: IdempotencyScope{
+					PrincipalID: "operator", TenantID: "tenant", SiteID: "site",
+					Operation: "batchSetDeviceDesiredState",
+					Key:       fmt.Sprintf("batch-lock-order-%04d", index), RequestHash: hash,
+					TraceID: fmt.Sprintf("trace-lock-order-%d", index),
+				},
+				Reason: "concurrency test",
+				Items: []ControlBatchItem{
+					{DeviceID: order[0], ETag: DeviceETag(order[0], 1), DesiredState: device.DesiredEnabled},
+					{DeviceID: order[1], ETag: DeviceETag(order[1], 1), DesiredState: device.DesiredEnabled},
+				},
+			}
+			<-start
+			operation, err := postgres.BatchSetControlDesiredState(context.Background(), request)
+			if err != nil {
+				errorsFound <- err
+				return
+			}
+			operations <- operation
+		}(index)
+	}
+	close(start)
+	wait.Wait()
+	close(operations)
+	close(errorsFound)
+	for err := range errorsFound {
+		t.Fatalf("opposite-order batch failed or deadlocked: %v", err)
+	}
+	var succeeded, failed int
+	for operation := range operations {
+		switch operation.Status {
+		case "succeeded":
+			succeeded++
+		case "failed":
+			failed++
+		default:
+			t.Fatalf("unexpected concurrent batch status: %+v", operation)
+		}
+	}
+	if succeeded != 1 || failed != 1 {
+		t.Fatalf("expected one winner and one stale loser, got succeeded=%d failed=%d", succeeded, failed)
+	}
+}
+
 func openPostgresTestStore(t *testing.T) (*Postgres, *sql.DB) {
 	t.Helper()
 	dsn := os.Getenv(postgresTestDSNEnv)
@@ -629,6 +985,9 @@ func openPostgresTestStore(t *testing.T) (*Postgres, *sql.DB) {
 		t.Fatal("connect PostgreSQL test administrator")
 	}
 	if _, err := admin.ExecContext(context.Background(), `TRUNCATE
+		sense.control_idempotency_receipts,
+		sense.batch_operation_items,
+		sense.batch_operations,
 		sense.device_operation_outbox,
 		sense.device_capabilities,
 		sense.reconcile_state,

Sense/internal/store/sqlite.go

@@ -366,12 +366,12 @@ func (s *SQLite) ListDueReconcile(ctx context.Context, now time.Time, limit int)
 	rows, err := s.db.QueryContext(ctx, `SELECT `+deviceColumns+`, r.failure_count, r.next_attempt_at
 		FROM sense_devices d
 		JOIN sense_reconcile_state r ON r.device_id = d.id
-		WHERE d.desired_state = 'enabled'
-			AND EXISTS (
+		WHERE EXISTS (
 				SELECT 1 FROM sense_device_capabilities c
 				WHERE c.device_id = d.id AND c.capability = 'video_capture'
 			)
-			AND (r.observed_generation < d.generation OR r.failure_count > 0)
+			AND (r.observed_generation < d.generation
+				OR (d.desired_state = 'enabled' AND r.failure_count > 0))
 			AND (r.next_attempt_at IS NULL OR r.next_attempt_at <= ?)
 		ORDER BY d.updated_at, d.id
 		LIMIT ?`, formatTime(now), limit)
@@ -487,7 +487,9 @@ func (s *SQLite) MarkReconciled(ctx context.Context, id string, generation int64
 		return ErrNotFound
 	}
 	if _, err := tx.ExecContext(ctx, `
-		UPDATE sense_devices SET actual_state = 'pending', updated_at = ? WHERE id = ?`,
+		UPDATE sense_devices SET
+			actual_state = CASE WHEN desired_state = 'disabled' THEN 'offline' ELSE 'pending' END,
+			updated_at = ? WHERE id = ?`,
 		formatTime(now), id); err != nil {
 		return fmt.Errorf("mark reconciled device pending: %w", err)
 	}