pigway-frame-flow/pigway_frame_flow/display/service.py

import hashlib
import json
import threading
import time
from datetime import UTC, datetime, timedelta
from uuid import uuid4

from pigway_frame_flow.core.photo_preferences import eligible
from pigway_frame_flow.display.cloud_library import CloudLibrary
from pigway_frame_flow.display.recovery import RecoveryMonitor
from pigway_frame_flow.display.retention import Retention
from pigway_frame_flow.display.upload_reconciliation import UploadReconciliation
from pigway_frame_flow.drivers.base import DriverRegistry
from pigway_frame_flow.errors import ErrorCode as E
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.models import (
    AutoCollageContent,
    DisplayRequest,
    FutureContent,
    ImageContent,
    PhotoSourceContent,
    ProviderContent,
    TemplateContent,
    URLContent,
)
from pigway_frame_flow.providers.photo_sources import PhotoSources
from pigway_frame_flow.renderers.frame_geometry import frame_canvas
from pigway_frame_flow.renderers.images import convert
from pigway_frame_flow.renderers.photo_caption import render_captioned_photo
from pigway_frame_flow.storage.database import encode, now_iso


class DisplayService:
    """One persistent queue; injected driver registry. No vendor imports."""

    def __init__(self, db, settings, registry: DriverRegistry, assets, renderer):
        self.photo_sources = PhotoSources(db, settings, assets)
        self.db, self.settings, self.registry, self.assets, self.renderer = (
            db,
            settings,
            registry,
            assets,
            renderer,
        )
        from typing import Callable

        self.operation_handlers: dict[str, Callable] = {}
        self.reconciliation_handlers: dict[str, Callable] = {}
        from pigway_frame_flow.providers.external import ExternalProviders

        self.providers = ExternalProviders(self)
        self.lock = threading.RLock()
        self.retention = Retention(self)
        self.cloud_library = CloudLibrary(self)
        self.recovery = RecoveryMonitor(self)
        self.upload_reconciliation = UploadReconciliation(self)
        from pigway_frame_flow.display.libraries import Libraries

        self.libraries = Libraries(self)
        self.expiration_checked: dict[str, float] = {}
        self.unknown_checked: dict[str, float] = {}
        # A killed write cannot be replayed safely; pre-write work can resume.
        db.execute(
            "UPDATE operations SET status='UNKNOWN', updated_at=? WHERE write_started=1 AND status IN ('UPLOADING','RENDERING')",
            (now_iso(),),
        )
        db.execute(
            "UPDATE operations SET status='QUEUED' WHERE write_started=0 AND status IN ('RENDERING','UPLOADING')"
        )

    def validate(self, request, writing=True):
        self.registry.get(request.device_id)
        if (
            writing
            and isinstance(request.content, ImageContent)
            and request.content.asset_id
        ):
            self.libraries.validate_asset_binding(
                request.device_id, request.content.asset_id
            )
        if request.options.thumbnail_plan_id:
            from pigway_frame_flow.display.thumbnail_plans import get_plan

            get_plan(self, request.options.thumbnail_plan_id)
        if isinstance(request.content, FutureContent):
            raise FlowError(
                E.UNSUPPORTED_CAPABILITY, "Content type is unsupported", 422
            )
        if isinstance(request.content, ProviderContent):
            self.providers.config(request.content.provider_id)
            self.renderer.info(request.content.template)
        if isinstance(request.content, TemplateContent):
            self.renderer.info(request.content.template)
            if request.content.template in ("collage", "poster"):
                from pigway_frame_flow.extensions.layout import LayoutPlan

                try:
                    plan = LayoutPlan.model_validate(request.content.data.get("layout"))
                except ValueError:
                    raise FlowError(
                        E.INVALID_REQUEST,
                        "Create a valid layout in Layout Studio first",
                    ) from None
                for region in plan.regions:
                    self.assets.path(region.asset_id)
        if (
            isinstance(request.content, AutoCollageContent)
            and request.content.source_id
        ):
            self.photo_sources.get(request.content.source_id)
        if isinstance(request.content, PhotoSourceContent):
            self.photo_sources.get(request.content.source_id)
        if writing and not self.settings.mock and not self.settings.write_enabled:
            raise FlowError(E.WRITE_DISABLED, "Real device writes are disabled", 403)

    def submit(self, action, payload, device_id, client_id, request_id, key=None):
        fingerprint = hashlib.sha256(
            encode({"action": action, "payload": payload, "device": device_id}).encode()
        ).hexdigest()
        with self.db.transaction() as tx:
            tx.execute("DELETE FROM idempotency WHERE expires_at<?", (now_iso(),))
            if key:
                existing = tx.execute(
                    "SELECT * FROM idempotency WHERE client_id=? AND key=?",
                    (client_id, key),
                ).fetchone()
                if existing:
                    if existing["fingerprint"] != fingerprint:
                        raise FlowError(
                            E.CONFLICT, "Idempotency key has a different request", 409
                        )
                    return existing["operation_id"]
            if (
                action
                not in (
                    "display",
                    "render",
                    "render-ai",
                    "local-upload",
                    "library-generate",
                )
                and tx.execute(
                    "SELECT id FROM operations WHERE device_id=? AND status='UNKNOWN' AND write_started=1 AND action!='render-ai' LIMIT 1",
                    (device_id,),
                ).fetchone()
            ):
                raise FlowError(
                    E.WRITE_RESULT_UNKNOWN,
                    "Write result unknown; reconcile device state",
                    409,
                )
            id = uuid4().hex
            now = now_iso()
            tx.execute(
                "INSERT INTO operations(id,status,action,payload,device_id,client_id,request_id,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?)",
                (
                    id,
                    "QUEUED",
                    action,
                    encode(self.settings.sanitize(payload)),
                    device_id,
                    client_id,
                    request_id,
                    now,
                    now,
                ),
            )
            if key:
                expires = (
                    datetime.now(UTC) + timedelta(hours=self.settings.idempotency_hours)
                ).isoformat()
                tx.execute(
                    "INSERT INTO idempotency VALUES(?,?,?,?,?)",
                    (client_id, key, fingerprint, id, expires),
                )
        return id

    def operation_image(self, device_id, result):
        result = dict(result or {})
        asset_id = result.get("rendered_asset_id") or result.get("asset_id")
        if not asset_id and result.get("image_id") is not None:
            # Cloud cleanup removes image_sources; the original upload receipt survives.
            row = self.db.one(
                "SELECT result FROM operations WHERE device_id=? "
                "AND json_extract(result,'$.image_id')=? "
                "AND json_extract(result,'$.rendered_asset_id') IS NOT NULL "
                "ORDER BY created_at DESC LIMIT 1",
                (device_id, result["image_id"]),
            )
            if row:
                original = json.loads(row["result"])
                asset_id = original["rendered_asset_id"]
                result.setdefault("image_name", original.get("image_name"))
        if asset_id:
            result.setdefault("asset_id", asset_id)
            asset = self.db.one("SELECT * FROM assets WHERE id=?", (asset_id,))
            if asset:
                result["image_name"] = result.get("image_name") or asset["name"]
                if self.assets.path(asset_id).is_file():
                    result["preview_url"] = f"/api/v1/assets/{asset_id}"
        return result or None

    def operation(self, id):
        op = self.db.one("SELECT * FROM operations WHERE id=?", (id,))
        if not op:
            raise FlowError(E.NOT_FOUND, "Operation not found", 404)
        return dict(
            operation_id=op["id"],
            status=op["status"],
            device_id=op["device_id"],
            created_at=op["created_at"],
            updated_at=op["updated_at"],
            result=self.operation_image(
                op["device_id"], json.loads(op["result"]) if op["result"] else None
            ),
            error=json.loads(op["error"]) if op["error"] else None,
            request_id=op["request_id"],
            client_id=op["client_id"],
            action=op["action"],
            recovery=self.recovery.info(id),
        )

    def update(self, id, status, result=None, error=None):
        if result:
            op = self.db.one("SELECT device_id FROM operations WHERE id=?", (id,))
            result = self.operation_image(op["device_id"], result)
            result.pop("preview_url", None)
            row = self.db.one("SELECT result FROM operations WHERE id=?", (id,))
            previous = json.loads(row["result"]) if row and row["result"] else {}
            for key in (
                "photo_attempts",
                "upload_intent",
                "sync_observations",
                "wrist_thumbnail",
            ):
                if key in previous:
                    result = {**result, key: previous[key]}
        self.db.execute(
            "UPDATE operations SET status=?,updated_at=?,result=COALESCE(?,result),error=? WHERE id=?",
            (
                status,
                now_iso(),
                encode(result) if result else None,
                encode(error) if error else None,
                id,
            ),
        )
        if status == "COMPLETED":
            row = self.db.one("SELECT result FROM operations WHERE id=?", (id,))
            asset_id = json.loads(row["result"] or "{}").get("rendered_asset_id")
            if asset_id and self.db.one(
                "SELECT id FROM assets WHERE id=?", (asset_id,)
            ):
                self.assets.promote(asset_id)
        if status in ("COMPLETED", "FAILED"):
            self.recovery.resolve(id)

    def blocked(self, device_id):
        return (
            self.db.one(
                "SELECT id FROM operations WHERE device_id=? AND status='UNKNOWN' AND write_started=1 AND action!='render-ai' LIMIT 1",
                (device_id,),
            )
            is not None
        )

    def state(self, id):
        self.db.execute(
            "INSERT OR IGNORE INTO device_state(device_id) VALUES(?)", (id,)
        )
        return self.db.one("SELECT * FROM device_state WHERE device_id=?", (id,))

    def clear_self_restore_reference(self, device_id, device):
        # A confirmed return to the managed image consumes its restore target.
        # Keep references while an override or an uncertain write still needs them.
        image_id = device.current_content.id
        if (
            device.status == "offline"
            or not device.current_content.synced
            or image_id is None
        ):
            return
        with self.lock:
            self.db.execute(
                """UPDATE device_state SET previous_image_id=NULL
                WHERE device_id=? AND previous_image_id=? AND managed_image_id=?
                AND base_image_id IS NULL
                AND NOT EXISTS (SELECT 1 FROM overrides WHERE device_id=?)
                AND NOT EXISTS (SELECT 1 FROM operations WHERE device_id=?
                    AND write_started=1 AND status IN ('UPLOADING','SYNCING','UNKNOWN'))""",
                (device_id, image_id, image_id, device_id, device_id),
            )

    def content_asset(
        self,
        content,
        device_id=None,
        operation_id=None,
        *,
        staging=False,
        source_image=None,
        source_only=False,
        source_size=None,
    ):
        if isinstance(content, AutoCollageContent):
            from pigway_frame_flow.extensions.auto_collage import prepare

            return self.content_asset(
                prepare(self, content, device_id),
                device_id,
                operation_id,
                staging=staging,
            )
        if isinstance(content, ProviderContent):
            values = {**content.data, **self.providers.get(content.provider_id)}
            return self.content_asset(
                TemplateContent(
                    type="template",
                    template=content.template,
                    data=values,
                    profile=content.profile,
                ),
                device_id,
                operation_id,
                staging=staging,
            )
        if isinstance(content, TemplateContent):
            values = self.renderer.resolve(content.template, content.data)
            data = self.renderer.render(content.template, values, content.profile)
            asset = self.assets.put(
                data, content.template + ".jpg", "cache" if staging else "generated"
            )
            source = {
                "type": "template",
                "template": content.template,
                "data": self.settings.sanitize(values),
                "profile": content.profile,
            }
            self.db.execute(
                "INSERT OR REPLACE INTO asset_sources VALUES(?,?)",
                (asset["asset_id"], encode(source)),
            )
            return asset
        if isinstance(content, PhotoSourceContent):
            source = self.photo_sources.get(content.source_id)
            if source_only:
                size = source_size or (1200, 1600)
            else:
                display = (
                    self.registry.get(device_id or self.settings.device_id)
                    .device()
                    .display
                )
                size = frame_canvas(display, source.orientation)
            import random

            items = self.photo_sources.catalog(content.source_id, target_size=size)
            if not content.item_id:
                items = eligible(self.db, content.source_id, items)
            candidates = (
                [item for item in items if item["id"] == content.item_id]
                if content.item_id
                else random.SystemRandom().sample(items, min(5, len(items)))
            )
            if not candidates:
                raise FlowError(
                    E.NOT_FOUND, "Photo source contains no matching images", 404
                )
            attempts = []
            for item in candidates:
                attempt = {
                    "source_id": content.source_id,
                    "item_id": item["id"],
                    "name": item["name"],
                    "status": "SELECTED",
                }
                attempts.append(attempt)
                if operation_id:
                    self.db.execute(
                        "UPDATE operations SET result=? WHERE id=?",
                        (encode({"photo_attempts": attempts}), operation_id),
                    )
                try:
                    data, name, caption = self.photo_sources.read(
                        content.source_id, item["id"], target_size=size
                    )
                    if (
                        source.minimum_sharpness
                        or source.minimum_brightness
                        or source.maximum_brightness < 255
                        or source.avoid_duplicates
                    ):
                        from pigway_frame_flow.extensions.quality import (
                            passes,
                            photo_quality,
                        )

                        quality = photo_quality(data)
                        if not passes(quality, source):
                            raise FlowError(
                                E.INVALID_REQUEST,
                                "Photo fails configured quality thresholds",
                            )
                        previous_quality = self.db.one(
                            "SELECT payload FROM app_config WHERE key=?",
                            ("recent-photo:" + content.source_id,),
                        )
                        fingerprints = (
                            json.loads(previous_quality["payload"])
                            if previous_quality
                            else []
                        )
                        if source.avoid_duplicates and any(
                            int(quality["fingerprint"], 16)
                            .__xor__(int(old, 16))
                            .bit_count()
                            <= 4
                            for old in fingerprints
                        ):
                            raise FlowError(
                                E.CONFLICT,
                                "Photo resembles a recently selected photo",
                                409,
                            )
                        self.db.execute(
                            "INSERT OR REPLACE INTO app_config VALUES(?,?)",
                            (
                                "recent-photo:" + content.source_id,
                                encode(
                                    (fingerprints + [quality["fingerprint"]])[-200:]
                                ),
                            ),
                        )
                    if source_only:
                        selected = self.libraries.ingest(
                            content.source_id, item["id"], data, name, caption
                        )
                        attempt["status"] = "PREPARED"
                        if operation_id:
                            self.db.execute(
                                "UPDATE operations SET result=? WHERE id=?",
                                (encode({"photo_attempts": attempts}), operation_id),
                            )
                        return selected
                    rendered = (
                        render_captioned_photo(
                            data,
                            caption,
                            self.renderer.font,
                            content.fit,
                            content.profile,
                            size=size,
                        )
                        if caption
                        else convert(data, content.fit, content.profile, size=size)
                    )
                except FlowError as exc:
                    attempt["status"] = "SKIPPED"
                    attempt["code"] = exc.code
                    if operation_id:
                        self.db.execute(
                            "UPDATE operations SET result=? WHERE id=?",
                            (encode({"photo_attempts": attempts}), operation_id),
                        )
                    # Only local preparation can choose another image. No device
                    # write has begun; explicit selections never change silently.
                    if content.item_id or exc.code not in (
                        E.INVALID_REQUEST,
                        E.NOT_FOUND,
                    ):
                        raise
                    continue
                if source_image is not None:
                    source_image["data"] = data
                attempt["status"] = "PREPARED"
                if operation_id:
                    self.db.execute(
                        "UPDATE operations SET result=? WHERE id=?",
                        (encode({"photo_attempts": attempts}), operation_id),
                    )
                asset = self.assets.put(
                    rendered, name, "cache" if staging else "images"
                )
                self.db.execute(
                    "INSERT OR REPLACE INTO asset_sources VALUES(?,?)",
                    (
                        asset["asset_id"],
                        encode(
                            {
                                "type": "photo_source",
                                "source_id": content.source_id,
                                "item_id": item["id"],
                                "source_path": item.get("source_path"),
                            }
                        ),
                    ),
                )
                return self.assets.info(asset["asset_id"])
            raise FlowError(
                E.INVALID_REQUEST, "No decodable photo found within 5 candidates"
            )
        if isinstance(content, URLContent):
            data = self.assets.fetch_url(content.url)
        elif isinstance(content, ImageContent) and content.asset_id:
            return self.assets.info(content.asset_id)
        elif isinstance(content, ImageContent) and content.base64:
            data = self.assets.decode(content.base64)
        else:
            raise FlowError(E.INVALID_REQUEST, "Content requires renderable image data")
        return self.assets.put(
            convert(data, content.fit, content.profile),
            content.name,
            "cache" if staging else "images",
        )

    def current_content(self, device_id):
        device = self.registry.get(device_id).device()
        self.clear_self_restore_reference(device_id, device)
        image_id = device.current_content.id
        row = self.db.one(
            "SELECT * FROM image_sources WHERE device_id=? AND image_id=?",
            (device_id, image_id),
        )
        source = json.loads(row["payload"]) if row else None
        # Acknowledged upload is not proof that the frame has refreshed yet.
        if row:
            op = self.operation(row["operation_id"])
            if op["status"] not in ("COMPLETED", "SYNCING"):
                source = None
        state = self.state(device_id)
        return {
            "device_id": device_id,
            "image_id": image_id,
            "synced": device.current_content.synced,
            "source_known": source is not None,
            "source": source,
            "asset_id": row["asset_id"] if row and source else None,
            "operation_id": row["operation_id"] if row and source else None,
            "preview_url": f"/api/v1/devices/{device_id}/images/{image_id}/file?thumbnail=true"
            if image_id
            else None,
            "previous_image_id": state["previous_image_id"],
            "overrides": self.db.all(
                "SELECT id,image_id,priority,expires_at FROM overrides WHERE device_id=? ORDER BY priority DESC,created_at DESC",
                (device_id,),
            ),
        }

    def top(self, device_id):
        return self.db.one(
            "SELECT * FROM overrides WHERE device_id=? AND expires_at>? ORDER BY priority DESC,created_at DESC LIMIT 1",
            (device_id, now_iso()),
        )

    def queued_operations(self):
        cursor_time, cursor_id = "", ""
        while True:
            rows = self.db.all(
                "SELECT * FROM operations WHERE status='QUEUED' AND "
                "(created_at>? OR (created_at=? AND id>?)) "
                "ORDER BY created_at,id LIMIT 50",
                (cursor_time, cursor_time, cursor_id),
            )
            if not rows:
                return
            yield from rows
            cursor_time, cursor_id = rows[-1]["created_at"], rows[-1]["id"]

    def upload_retry_expired(self, op):
        if op["client_id"] != "scheduler-retry" or not op["request_id"].startswith(
            "upload-retry-"
        ):
            return False
        root = op["request_id"].split("-")[-2]
        for row in self.db.all(
            "SELECT payload FROM app_config WHERE key LIKE 'upload-retry:%'"
        ):
            plan = json.loads(row["payload"])
            if plan["root"] == root:
                return now_iso() >= plan["deadline"]
        return True

    def process_one(self):
        with self.lock:
            for op in self.queued_operations():
                if self.upload_retry_expired(op):
                    self.update(
                        op["id"],
                        "FAILED",
                        error={
                            "code": E.CONFLICT,
                            "message": "Upload retry window closed",
                            "request_id": op["request_id"],
                            "details": {"reason": "RETRY_WINDOW_CLOSED"},
                        },
                    )
                    self.db.history(op, "RETRY_WINDOW_CLOSED")
                    continue
                if op["action"] not in (
                    "display",
                    "render",
                    "render-ai",
                    "local-upload",
                    "library-generate",
                ) and self.blocked(op["device_id"]):
                    continue
                payload = json.loads(op["payload"])
                if (
                    op["action"] in ("display", "delete", "restore")
                    and op["client_id"]
                    in (
                        "scheduler",
                        "scheduler-retry",
                        "library-terminal",
                        "cloud-library",
                        "retention",
                    )
                    and self.db.one(
                        "SELECT id FROM operations WHERE device_id=? AND action='display' "
                        "AND status='SYNCING' AND write_started=1 AND updated_at>? LIMIT 1",
                        (
                            op["device_id"],
                            (
                                datetime.now(UTC)
                                - timedelta(seconds=self.settings.sync_timeout)
                            ).isoformat(),
                        ),
                    )
                ):
                    # Both upload and deletion update the cloud frame counter.
                    # Automatic writes must not invalidate a pending display ack.
                    continue
                if op["action"] == "display":
                    priority = payload["options"]["priority"]
                    top = self.top(op["device_id"])
                    if top and priority < top["priority"]:
                        continue
                try:
                    self._execute(op, payload)
                except FlowError as exc:
                    uncertain = exc.code == E.WRITE_RESULT_UNKNOWN
                    error = {
                        "code": exc.code,
                        "message": exc.message,
                        "request_id": op["request_id"],
                        "details": exc.details,
                    }
                    self.update(
                        op["id"], "UNKNOWN" if uncertain else "FAILED", error=error
                    )
                    self.db.history(op, exc.code)
                    self.db.event(
                        "operation.failed", {"operation_id": op["id"], "code": exc.code}
                    )
                except Exception:
                    wrote = self.db.one(
                        "SELECT write_started FROM operations WHERE id=?", (op["id"],)
                    )["write_started"]
                    self.update(
                        op["id"],
                        "UNKNOWN" if wrote else "FAILED",
                        error={
                            "code": E.WRITE_RESULT_UNKNOWN
                            if wrote
                            else E.INTERNAL_ERROR,
                            "message": "Write outcome unknown; reconcile device state"
                            if wrote
                            else "Operation failed",
                            "request_id": op["request_id"],
                            "details": {},
                        },
                    )
                    self.db.history(op, "UNKNOWN" if wrote else "FAILED")
                return True
        return False

    def _execute(self, op, payload):
        id, device_id, action = op["id"], op["device_id"], op["action"]
        # A source run can generate watch versions without any frame driver.
        if action == "library-generate":
            return self.operation_handlers[action](op, payload)
        driver = self.registry.get(device_id)
        if action in self.operation_handlers:
            return self.operation_handlers[action](op, payload)
        if action in ("render", "render-ai", "local-upload"):
            self.update(id, "RENDERING")
            from pydantic import TypeAdapter

            from pigway_frame_flow.models import Content

            source_image: dict[str, bytes] = {}
            asset = self.content_asset(
                TypeAdapter(Content).validate_python(payload),
                device_id,
                id,
                source_image=source_image,
            )
            self.assets.record_origin(asset["asset_id"], op)
            if action == "local-upload":
                asset = self.assets.promote(asset["asset_id"])
            result = dict(asset_id=asset["asset_id"], url=asset["url"])
            if action == "local-upload":
                from pigway_frame_flow.display.wrist_thumbnail import publish

                thumbnail = publish(
                    self, device_id, asset["asset_id"], source_image.get("data")
                )
                if thumbnail:
                    result["wrist_thumbnail"] = thumbnail
            self.update(id, "COMPLETED", result)
            self.db.history(op, "COMPLETED")
            return
        device = driver.device(refresh=True)
        if device.status == "offline":
            raise FlowError(E.DEVICE_OFFLINE, "Target frame is offline", 409)
        # B101 layout is deliberately not transposed based on rotation.
        if (device.display.width, device.display.height) != (1200, 1600):
            raise FlowError(
                E.UNSUPPORTED_CAPABILITY, "Device geometry does not match renderer", 409
            )
        previous = device.current_content.id
        state = self.state(device_id)
        if action == "delete":
            self.clear_self_restore_reference(device_id, device)
            state = self.state(device_id)
            image_id = payload["image_id"]
            if payload.get("cloud_cleanup"):
                if (
                    not self.cloud_library.eligible(device_id, image_id)
                    or previous == image_id
                ):
                    self.db.execute(
                        "UPDATE managed_cloud_uploads SET cleanup_operation=NULL WHERE device_id=? AND image_id=?",
                        (device_id, image_id),
                    )
                    self.update(
                        id, "COMPLETED", {"image_id": image_id, "deferred": True}
                    )
                    return
                # The confirmed replacement supersedes this task-owned restore
                # reference; unrelated artwork and active overrides stay protected.
                self.db.execute(
                    "UPDATE device_state SET previous_image_id=NULL WHERE device_id=? AND previous_image_id=?",
                    (device_id, image_id),
                )
                state = self.state(device_id)
            if payload.get("retention") and (
                not self.retention.eligible(device_id, image_id) or image_id == previous
            ):
                self.db.execute(
                    "UPDATE image_retention SET cleanup_operation=NULL WHERE device_id=? AND image_id=?",
                    (device_id, image_id),
                )
                self.update(id, "COMPLETED", {"image_id": image_id, "deferred": True})
                self.db.history(op, "RETENTION_PROTECTED")
                return
            protected = {state["previous_image_id"], state["base_image_id"]} | {
                x["image_id"]
                for x in self.db.all(
                    "SELECT image_id FROM overrides WHERE device_id=?", (device_id,)
                )
            }
            if image_id in protected and not payload.get("purge"):
                raise FlowError(
                    E.CONFLICT,
                    "Image is needed for restoration",
                    409,
                    {"reason": "RESTORE_IMAGE_PROTECTED"},
                )
            if image_id == previous and not payload["confirm_current"]:
                raise FlowError(E.CONFLICT, "Current image needs confirmation", 409)
            self._start_write(id)
            driver.delete(image_id)
            self.db.execute(
                "DELETE FROM managed_cloud_uploads WHERE device_id=? AND image_id=?",
                (device_id, image_id),
            )
            if payload.get("purge"):
                self.db.execute(
                    "DELETE FROM overrides WHERE device_id=? AND image_id=?",
                    (device_id, image_id),
                )
                for field in ("previous_image_id", "base_image_id", "managed_image_id"):
                    self.db.execute(
                        f"UPDATE device_state SET {field}=NULL WHERE device_id=? AND {field}=?",
                        (device_id, image_id),
                    )
            for suffix in [".original", "-thumb.jpg"]:
                (
                    self.settings.data_dir / "cache" / f"{device_id}-{image_id}{suffix}"
                ).unlink(missing_ok=True)
            self.db.execute(
                "DELETE FROM image_sources WHERE device_id=? AND image_id=?",
                (device_id, image_id),
            )
            self.db.execute(
                "DELETE FROM image_retention WHERE device_id=? AND image_id=?",
                (device_id, image_id),
            )
            driver.invalidate()
            self.update(id, "COMPLETED", {"image_id": image_id})
            self.db.history(op, "COMPLETED")
            return
        if action == "restore":
            target = (
                state["base_image_id"]
                if self.top(device_id)
                else state["previous_image_id"]
            )
            if target is None:
                raise FlowError(E.CONFLICT, "No previous image to restore", 409)
            self._start_write(id)
            driver.display(target)
            self.db.execute("DELETE FROM overrides WHERE device_id=?", (device_id,))
            self.db.execute(
                "UPDATE device_state SET base_image_id=NULL,managed_image_id=? WHERE device_id=?",
                (target, device_id),
            )
            driver.invalidate()
            self.update(id, "SYNCING", {"image_id": target})
            self.db.history(op, "ACCEPTED")
            return
        # A new upload is independent of a previous failed write. Read back its
        # outcome before closing the old attempt; never replay that operation.
        for pending in self.db.all(
            "SELECT id FROM operations WHERE device_id=? AND status='UNKNOWN' AND write_started=1 AND action='display'",
            (device_id,),
        ):
            self.reconcile(pending["id"], acknowledge=True)
        request = DisplayRequest.model_validate(payload)
        if request.options.thumbnail_plan_id:
            from pigway_frame_flow.display.thumbnail_plans import get_plan

            get_plan(self, request.options.thumbnail_plan_id)
        content = request.content
        if isinstance(content, ImageContent) and content.asset_id:
            self.libraries.validate_asset_binding(device_id, content.asset_id)
        self.update(id, "RENDERING")
        asset = None
        source_image = {}
        if isinstance(content, ImageContent) and (
            content.image_id or content.image_name
        ):
            image_id = content.image_id
            if content.image_name:
                matches = [x for x in driver.images() if x.name == content.image_name]
                if len(matches) != 1:
                    raise FlowError(
                        E.CONFLICT, "Name must identify exactly one gallery image", 409
                    )
                image_id = matches[0].image_id
        else:
            asset = self.content_asset(
                content, device_id, id, staging=True, source_image=source_image
            )
            self.assets.record_origin(asset["asset_id"], op)
            image_id = None
        self.db.execute(
            "UPDATE device_state SET previous_image_id=? WHERE device_id=?",
            (previous, device_id),
        )
        if request.options.restore_previous and not self.top(device_id):
            self.db.execute(
                "UPDATE device_state SET base_image_id=? WHERE device_id=?",
                (previous, device_id),
            )
        if request.options.restore_previous and previous is None:
            raise FlowError(
                E.CONFLICT, "Temporary override requires an existing image", 409
            )
        if asset:
            self.upload_reconciliation.prepare(id, driver, asset, previous)
        self._start_write(id)
        if asset:
            self.update(id, "UPLOADING")
            image_id = driver.upload(self.assets.path(asset["asset_id"]), asset["name"])
            self.cloud_library.register(device_id, image_id, asset["asset_id"], id)
            if (
                request.options.cleanup_unstarred
                and isinstance(content, PhotoSourceContent)
                and content.item_id is None
            ):
                self.retention.register(device_id, image_id, id)
        elif isinstance(content, ImageContent) and content.image_name:
            driver.display_by_name(content.image_name)
        else:
            driver.display(image_id)
        result = {"image_id": image_id}
        if asset and request.options.generate_thumbnail:
            from pigway_frame_flow.display.wrist_thumbnail import publish

            thumbnail = publish(
                self,
                device_id,
                asset["asset_id"],
                source_image.get("data"),
                plan_id=request.options.thumbnail_plan_id,
            )
            if thumbnail:
                result["wrist_thumbnail"] = thumbnail
        if asset:
            result["rendered_asset_id"] = asset["asset_id"]
            source = self.db.one(
                "SELECT payload FROM asset_sources WHERE asset_id=?",
                (asset["asset_id"],),
            )
            descriptor = source["payload"] if source else encode({"type": "image"})
            self.db.execute(
                "INSERT OR REPLACE INTO image_sources VALUES(?,?,?,?,?)",
                (device_id, image_id, asset["asset_id"], id, descriptor),
            )
        self.update(id, "SYNCING", result)
        driver.invalidate()
        if request.options.restore_previous:
            expires = (
                datetime.now(UTC) + timedelta(seconds=request.options.duration)
            ).isoformat()
            self.db.execute(
                "INSERT INTO overrides VALUES(?,?,?,?,?,?)",
                (id, device_id, image_id, request.options.priority, expires, now_iso()),
            )
        else:
            self.db.execute("DELETE FROM overrides WHERE device_id=?", (device_id,))
            self.db.execute(
                "UPDATE device_state SET base_image_id=NULL WHERE device_id=?",
                (device_id,),
            )
        self.db.execute(
            "UPDATE device_state SET managed_image_id=? WHERE device_id=?",
            (image_id, device_id),
        )
        self.db.history(op, "ACCEPTED")
        self.db.event(
            "display.changed",
            {"device_id": device_id, "image_id": image_id, "operation_id": id},
        )

    def _start_write(self, id):
        if not self.settings.mock and not self.settings.write_enabled:
            raise FlowError(E.WRITE_DISABLED, "Real device writes are disabled", 403)
        self.db.execute(
            "UPDATE operations SET write_started=1,status='UPLOADING',updated_at=? WHERE id=?",
            (now_iso(), id),
        )

    def sync(self):
        with self.lock:
            self._sync()

    def _sync(self):
        # Late acknowledgements only resolve a known display target. Never replay
        # writes or infer success from gallery presence alone.
        for op in self.db.all(
            "SELECT * FROM operations WHERE status='UNKNOWN' AND action!='render-ai' AND write_started=1"
        ):
            expected = json.loads(op["result"] or "{}").get("image_id")
            intent = json.loads(op["result"] or "{}").get("upload_intent")
            checked = time.monotonic()
            if checked - self.unknown_checked.get(op["id"], float("-inf")) < 60:
                continue
            self.unknown_checked[op["id"]] = checked
            if op["action"] != "display" or (not expected and not intent):
                self.recovery.observe(
                    op,
                    "missing_upload_evidence"
                    if op["action"] == "display"
                    else "manual_reconciliation_required",
                )
                continue
            try:
                driver = self.registry.get(op["device_id"])
                device = driver.device(refresh=True)
                self.recovery.observe_device(op, device=device)
                if intent:
                    expected = self.upload_reconciliation.check(op, driver, device)
                    if (
                        self.db.one(
                            "SELECT status FROM operations WHERE id=?", (op["id"],)
                        )["status"]
                        == "FAILED"
                    ):
                        continue
                if (
                    expected
                    and device.current_content.id == expected
                    and device.current_content.synced
                ):
                    self.update(op["id"], "COMPLETED")
                    self.clear_self_restore_reference(op["device_id"], device)
                    self.cloud_library.confirm(op["id"])
                    self.db.history(op, "COMPLETED")
                    self.db.event("operation.completed", {"operation_id": op["id"]})
                    self.unknown_checked.pop(op["id"], None)
                elif expected:
                    self.recovery.observe(
                        op,
                        "waiting_device_sync"
                        if device.current_content.id == expected
                        else "target_mismatch",
                    )
            except FlowError as exc:
                self.recovery.observe_device(op, error=exc)
                self.recovery.observe(op, "read_unavailable")
            except Exception:
                # Bad vendor metadata for one operation must not stop the worker.
                self.recovery.observe(op, "reconciliation_error")
        for op in self.db.all("SELECT * FROM operations WHERE status='SYNCING'"):
            try:
                device = self.registry.get(op["device_id"]).device()
                observation = self.recovery.observe_device(op, device=device)
                target = json.loads(op["result"])["image_id"]
                if (
                    device.current_content.id == target
                    and device.current_content.synced
                ):
                    self.update(op["id"], "COMPLETED")
                    self.clear_self_restore_reference(op["device_id"], device)
                    self.cloud_library.confirm(op["id"])
                    self.db.history(op, "COMPLETED")
                    self.db.event("operation.completed", {"operation_id": op["id"]})
                elif (
                    datetime.now(UTC) - datetime.fromisoformat(op["updated_at"])
                ).total_seconds() > self.settings.sync_timeout:
                    self.update(
                        op["id"],
                        "UNKNOWN",
                        error={
                            "code": E.WRITE_RESULT_UNKNOWN,
                            "message": "Frame refresh was not confirmed",
                            "request_id": op["request_id"],
                            "details": {"last_device_observation": observation},
                        },
                    )
            except FlowError as exc:
                self.recovery.observe_device(op, error=exc)

        self.recovery.close_superseded_displays()

    def expire(self):
        if not self.settings.mock and not self.settings.write_enabled:
            return
        for device_id in self.registry.drivers:
            if self.blocked(device_id):
                continue
            expired = self.db.all(
                "SELECT * FROM overrides WHERE device_id=? AND expires_at<=?",
                (device_id, now_iso()),
            )
            if not expired:
                continue
            if not self.settings.mock:
                last = self.expiration_checked.get(device_id, 0)
                if time.monotonic() - last < 60:
                    continue
                self.expiration_checked[device_id] = time.monotonic()
            driver = self.registry.get(device_id)
            state = self.state(device_id)
            try:
                device = driver.device(refresh=True)
                if device.status == "offline":
                    continue
                top = self.top(device_id)
                target = top["image_id"] if top else state["base_image_id"]
                if target is None:
                    continue
                if device.current_content.id != state["managed_image_id"]:
                    # A user/app changed the artwork; do not overwrite it.
                    self.db.execute(
                        "DELETE FROM overrides WHERE device_id=?", (device_id,)
                    )
                    self.db.execute(
                        "UPDATE device_state SET base_image_id=NULL WHERE device_id=?",
                        (device_id,),
                    )
                    continue
                if target != device.current_content.id:
                    id = self.submit(
                        "auto-restore",
                        {"image_id": target},
                        device_id,
                        "scheduler",
                        "restore-" + uuid4().hex,
                    )
                    op = self.db.one("SELECT * FROM operations WHERE id=?", (id,))
                    self._start_write(id)
                    try:
                        driver.display(target)
                    except FlowError as exc:
                        self.update(
                            id,
                            "UNKNOWN"
                            if exc.code == E.WRITE_RESULT_UNKNOWN
                            else "FAILED",
                            error={
                                "code": exc.code,
                                "message": exc.message,
                                "request_id": op["request_id"],
                                "details": {},
                            },
                        )
                        continue
                    driver.invalidate()
                    self.update(id, "SYNCING", {"image_id": target})
                    self.db.history(op, "ACCEPTED")
                self.db.execute(
                    "DELETE FROM overrides WHERE device_id=? AND expires_at<=?",
                    (device_id, now_iso()),
                )
                self.db.execute(
                    "UPDATE device_state SET managed_image_id=?,base_image_id=? WHERE device_id=?",
                    (target, state["base_image_id"] if top else None, device_id),
                )
            except FlowError:
                continue

    def reconcile(self, id, acknowledge=False):
        with self.lock:
            op = self.operation(id)
            if op["action"] in self.reconciliation_handlers:
                return self.reconciliation_handlers[op["action"]](op, acknowledge)
            driver = self.registry.get(op["device_id"])
            device = driver.device(refresh=True)
            images = driver.images()
            expected = (op["result"] or {}).get("image_id")
            confirmed = (
                bool(expected)
                and expected == device.current_content.id
                and device.current_content.synced
            )
            resolved = op["status"] != "UNKNOWN"
            if op["status"] == "UNKNOWN" and (
                acknowledge
                or (
                    expected
                    and expected == device.current_content.id
                    and device.current_content.synced
                )
            ):
                # Acknowledge releases the block; it never resends a write.
                self.update(
                    id,
                    "COMPLETED" if confirmed else "FAILED",
                    op["result"],
                    error=None
                    if confirmed
                    else {
                        "code": E.DISPLAY_NOT_CONFIRMED
                        if op["action"] == "display"
                        else E.WRITE_RESULT_UNKNOWN,
                        "message": "Manually closed unconfirmed write; no write replayed",
                        "request_id": op["request_id"],
                        "details": {
                            "observed_image_id": device.current_content.id,
                            "previous_error": op["error"],
                        },
                    },
                )
                resolved = True
                if confirmed and op["action"] == "display":
                    self.cloud_library.confirm(id)
                    self.db.event("operation.completed", {"operation_id": id})
                self.db.history(
                    dict(
                        id=id,
                        client_id="admin",
                        request_id=op["request_id"],
                        device_id=op["device_id"],
                        action="reconcile",
                    ),
                    "CONFIRMED" if confirmed else "ACKNOWLEDGED",
                )
            return dict(
                operation_id=id,
                current_image_id=device.current_content.id,
                expected_image_id=expected,
                cloud_images=images,
                resolved=resolved,
            )