feat: recover failed scheduled uploads with bounded persistent retries [skip ci]

deploy/Dockerfile.runtime-update

@@ -29,4 +29,4 @@ CMD ["python", "-m", "uvicorn", "pigway_frame_flow.main:app", "--host", "0.0.0.0
 ARG BUILD_COMMIT
 ARG BUILD_TIME
 ENV BUILD_COMMIT=$BUILD_COMMIT BUILD_TIME=$BUILD_TIME
-LABEL org.opencontainers.image.title="Pigway Frame Flow" org.opencontainers.image.version="0.1.11" org.opencontainers.image.revision=$BUILD_COMMIT
+LABEL org.opencontainers.image.title="Pigway Frame Flow" org.opencontainers.image.version="0.1.15" org.opencontainers.image.revision=$BUILD_COMMIT

docs/releases/v0.1.12.md

@@ -0,0 +1,5 @@
+# v0.1.12
+
+修复临时展示到期或手动恢复后，“上一张图片”仍指向当前图片，导致已确认删除仍被恢复保护拦截的问题。只有设备已确认同步到受管理的当前图片、且没有基础恢复目标、临时展示或待确认写入时，才清除这条自身引用；不同图片的有效恢复引用继续受保护。既有残留引用在读取设备状态或删除校验时自动修正，不需要强制删除图片或重置数据库。
+
+有效恢复引用仍拒绝删除，并通过结构化原因显示中英文提示。保留当前图片删除确认，不主动删除照片或切换画框。

docs/releases/v0.1.13.md

@@ -0,0 +1,3 @@
+# v0.1.13
+
+照片底部同时显示所有已识别且已命名的人物，人物之间使用 ` & `；日期、人物、主题、地点及文件名之间继续使用 ` · `。例如 `2026-10-18 · Alice & Bob · Family · one.jpg`。人物仍来自 Photos 实际相册归属，未命名人物不显示；历史生成图片不自动重绘。

docs/releases/v0.1.14.md

@@ -0,0 +1,5 @@
+# v0.1.14
+
+底部文字仅显示日期、已命名人物（使用 & 连接）、文件夹主题及地点，不显示原始文件名。原始文件名仍用于图库和诊断记录。
+
+上传传输区分明确的 DNS/连接建立失败与已建立连接中断：前者尚未发送 HTTP 写入，最多重试三次，失败后结束该次操作，不进入阻塞后续写入的 UNKNOWN；读超时或连接中断等不确定结果继续保留核对保护，不盲目重放上传。既有未知操作通过只读设备/图库核验后使用原有 acknowledge 流程关闭，保留原始错误和审计记录。

docs/releases/v0.1.15.md

@@ -0,0 +1,9 @@
+# v0.1.15
+
+自动照片、拼版、模板及数据模板任务在上传失败后，持久保存同一张暂存图片，并在下一次计划之前安排最多 5 次额外重试。间隔按 1、2、4、8、16 权重递增，随剩余窗口缩放，计划时间集中在窗口前半段，为上传与设备同步留出时间。重试不移动原定计划，也不复用旧操作 ID；每次独立记录操作及事件。成功后取消剩余重试，重启后沿用已有计数与截止时间；停用任务不再追加重试，删除任务取消其重试计划。
+
+若额外重试均失败，下一次计划仍先上传同一张图；若这一次仍失败，只删除该任务未入库、未星标的生成缓存，再按照片源生成并上传新图。NAS 原照片、用户上传文件和成功图库图片不由此流程删除。新图仍失败时，进入它自己的下一轮恢复窗口。
+
+上传结果未知不再阻塞后续独立上传。每次重试前先只读核对设备及云图库，精确确认已成功的上传不再重传；需继续尝试时关闭旧尝试并保留错误和核对审计。删除、恢复及自动清理仍保留各自的保护条件。重试窗口到期后不再启动额外上传，待同步尝试经核对结束后交给原计划执行。
+
+任务生成图片先保存在缓存，确认任务成功后才进入本地图库；失败不会产生看似成功的本地图库记录。额外重试仍归属原任务的云端保留规则，任务列表显示最近尝试结果。

pigway_frame_flow/api/app.py

@@ -126,7 +126,7 @@ def create_app(settings=None, start_worker=True):
     app = FastAPI(
         title="Pigway Frame Flow API",
         docs_url=None,
-        version="0.1.11",
+        version="0.1.15",
         lifespan=lifespan,
         responses=error_responses,
     )
@@ -305,6 +305,7 @@ def create_app(settings=None, start_worker=True):
     )
     def device(device_id: str, refresh: bool = False):
         d = registry.get(device_id).device(refresh)
+        service.clear_self_restore_reference(device_id, d)
         state = db.one("SELECT * FROM device_state WHERE device_id=?", (device_id,))
         possible = bool(
             state
@@ -943,6 +944,7 @@ def create_app(settings=None, start_worker=True):
     )
     def delete_job(job_id: str):
         db.execute("DELETE FROM jobs WHERE id=?", (job_id,))
+        db.execute("DELETE FROM app_config WHERE key=?", ("upload-retry:" + job_id,))
         return Response(status_code=204)
 
     @app.get(
@@ -1031,7 +1033,7 @@ def create_app(settings=None, start_worker=True):
                     supports_upload=True,
                 ),
             ],
-            version="0.1.11",
+            version="0.1.15",
             mock=settings.mock,
             write_enabled=settings.write_enabled,
             api_key_configured=bool(settings.api_key)

pigway_frame_flow/display/cloud_library.py

@@ -13,18 +13,29 @@ class CloudLibrary:
 
     def register(self, device_id, image_id, asset_id, operation_id):
         operation = self.db.one(
-            "SELECT payload FROM operations WHERE id=?", (operation_id,)
+            "SELECT payload,client_id FROM operations WHERE id=?", (operation_id,)
         )
         payload = json.loads(operation["payload"])
         content = payload.get("content", {})
         job = self.db.one(
             "SELECT job_id FROM job_runs WHERE operation_id=?", (operation_id,)
         )
+        if not job and operation["client_id"] == "scheduler-retry":
+            job = self.db.one(
+                "SELECT job_id FROM asset_metadata WHERE asset_id=? AND job_id IS NOT NULL",
+                (asset_id,),
+            )
+        job_config = (
+            self.db.one("SELECT payload FROM jobs WHERE id=?", (job["job_id"],))
+            if job
+            else None
+        )
+        source = json.loads(job_config["payload"])["source"] if job_config else content
         job_id = (
             job["job_id"]
             if job
-            and content.get("type") in ("photo_source", "auto_collage")
-            and content.get("item_id") is None
+            and source.get("type") in ("photo_source", "auto_collage")
+            and source.get("item_id") is None
             else None
         )
         self.db.execute(

pigway_frame_flow/display/service.py

@@ -115,7 +115,7 @@ class DisplayService:
                         )
                     return existing["operation_id"]
             if (
-                action not in ("render", "render-ai", "local-upload")
+                action not in ("display", "render", "render-ai", "local-upload")
                 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,),
@@ -187,6 +187,13 @@ class DisplayService:
                 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)
 
@@ -205,12 +212,38 @@ class DisplayService:
         )
         return self.db.one("SELECT * FROM device_state WHERE device_id=?", (id,))
 
-    def content_asset(self, content, device_id=None, operation_id=None):
+    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
+    ):
         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
+                prepare(self, content, device_id),
+                device_id,
+                operation_id,
+                staging=staging,
             )
         if isinstance(content, ProviderContent):
             values = {**content.data, **self.providers.get(content.provider_id)}
@@ -223,11 +256,14 @@ class DisplayService:
                 ),
                 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", "generated")
+            asset = self.assets.put(
+                data, content.template + ".jpg", "cache" if staging else "generated"
+            )
             source = {
                 "type": "template",
                 "template": content.template,
@@ -275,8 +311,6 @@ class DisplayService:
                     data, name, caption = self.photo_sources.read(
                         content.source_id, item["id"], target_size=size
                     )
-                    if source.folder_caption:
-                        caption = {**(caption or {}), "filename": name}
                     if (
                         source.minimum_sharpness
                         or source.minimum_brightness
@@ -358,7 +392,7 @@ class DisplayService:
                         "UPDATE operations SET result=? WHERE id=?",
                         (encode({"photo_attempts": attempts}), operation_id),
                     )
-                return self.assets.put(rendered, name)
+                return self.assets.put(rendered, name, "cache" if staging else "images")
             raise FlowError(
                 E.INVALID_REQUEST, "No decodable photo found within 5 candidates"
             )
@@ -371,11 +405,14 @@ class DisplayService:
         else:
             raise FlowError(E.INVALID_REQUEST, "Content requires renderable image data")
         return self.assets.put(
-            convert(data, content.fit, content.profile), content.name
+            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=?",
@@ -426,10 +463,38 @@ class DisplayService:
             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",
@@ -513,6 +578,8 @@ class DisplayService:
         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 (
@@ -551,7 +618,12 @@ class DisplayService:
                 )
             }
             if image_id in protected and not payload.get("purge"):
-                raise FlowError(E.CONFLICT, "Image is needed for restoration", 409)
+                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)
@@ -605,6 +677,13 @@ class DisplayService:
             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)
         content = request.content
         self.update(id, "RENDERING")
@@ -621,7 +700,7 @@ class DisplayService:
                     )
                 image_id = matches[0].image_id
         else:
-            asset = self.content_asset(content, device_id, id)
+            asset = self.content_asset(content, device_id, id, staging=True)
             self.assets.record_origin(asset["asset_id"], op)
             image_id = None
         self.db.execute(
@@ -656,7 +735,6 @@ class DisplayService:
             driver.display(image_id)
         result = {"image_id": image_id}
         if asset:
-            self.assets.promote(asset["asset_id"])
             result["rendered_asset_id"] = asset["asset_id"]
             source = self.db.one(
                 "SELECT payload FROM asset_sources WHERE asset_id=?",
@@ -743,6 +821,7 @@ class DisplayService:
                     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"]})
@@ -768,6 +847,7 @@ class DisplayService:
                     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"]})

pigway_frame_flow/display/upload_reconciliation.py

@@ -77,7 +77,6 @@ class UploadReconciliation:
             result["image_id"] = expected
             result["upload_confirmed"] = True
             service.update(op["id"], "UNKNOWN", result, json.loads(op["error"] or "{}"))
-            service.assets.promote(result["rendered_asset_id"])
             source = service.db.one(
                 "SELECT payload FROM asset_sources WHERE asset_id=?",
                 (result["rendered_asset_id"],),

pigway_frame_flow/drivers/kokonna/client.py

@@ -10,6 +10,7 @@ from collections import deque
 from urllib.parse import quote
 
 import requests
+from urllib3.exceptions import MaxRetryError, NewConnectionError
 
 from pigway_frame_flow.errors import ErrorCode as E
 from pigway_frame_flow.errors import FlowError
@@ -124,7 +125,15 @@ class SafeTransport:
                 ),
                 "request_error",
             )
-            not_sent = isinstance(exc, requests.ConnectTimeout)
+            # A socket/DNS establishment failure precedes the HTTP write. A
+            # dropped established connection still has an uncertain outcome.
+            not_sent = isinstance(exc, requests.ConnectTimeout) or any(
+                isinstance(cause, MaxRetryError)
+                and isinstance(cause.reason, NewConnectionError)
+                for cause in exc.args
+            )
+            if not_sent and category == "connection_error":
+                category = "connect_error"
             uncertain = write and not not_sent
             raise FlowError(
                 E.WRITE_RESULT_UNKNOWN if uncertain else E.CLOUD_UNAVAILABLE,

pigway_frame_flow/extensions/storage.py

@@ -43,6 +43,7 @@ class LocalRetention:
             "SELECT payload FROM jobs",
             "SELECT payload FROM template_config",
             "SELECT payload FROM operations WHERE status IN ('QUEUED','RENDERING','UPLOADING','SYNCING','UNKNOWN')",
+            "SELECT payload FROM app_config WHERE key LIKE 'upload-retry:%'",
         ]:
             payloads += [row["payload"] for row in db.all(query)]
         for payload in payloads:

pigway_frame_flow/models.py

@@ -366,7 +366,7 @@ class SettingsView(BaseModel):
 
 class Health(BaseModel):
     status: str
-    version: str = "0.1.11"
+    version: str = "0.1.15"
 
 
 class Readiness(BaseModel):
@@ -396,7 +396,7 @@ class RestoreRequest(StrictModel):
 
 class Version(BaseModel):
     product: str = "Pigway Frame Flow"
-    version: str = "0.1.11"
+    version: str = "0.1.15"
     api_version: str = "v1"
     commit: str = "development"
     build_time: str = "unknown"

pigway_frame_flow/providers/photo_sources.py

@@ -621,9 +621,8 @@ class PhotoSources:
             )
             if source.folder_caption:
                 caption = dict(item.get("caption") or {})
-                caption["filename"] = item["name"]
                 try:
-                    caption["people"] = " · ".join(
+                    caption["people"] = " & ".join(
                         self.synology.item_people(item["id"], space=source.people_space)
                     )
                 except FlowError:
@@ -631,7 +630,10 @@ class PhotoSources:
                     logging.getLogger(__name__).warning(
                         "Photos person labels unavailable; keeping photo caption"
                     )
-                item = {**item, "caption": caption}
+                item = {
+                    **item,
+                    "caption": {k: v for k, v in caption.items() if v} or None,
+                }
         elif source.kind in ("google-drive", "alipan", "baidu"):
             from pigway_frame_flow.providers.cloud_drives import CloudDrives
 

pigway_frame_flow/renderers/images.py

@@ -1,5 +1,6 @@
 import base64
 import ipaddress
+import json
 import socket
 from io import BytesIO
 from urllib.parse import urlsplit
@@ -115,9 +116,17 @@ class Assets:
     def promote(self, asset_id):
         row = self.db.one("SELECT * FROM assets WHERE id=?", (asset_id,))
         if row and row["kind"] == "cache":
-            target = self.settings.data_dir / "images" / (asset_id + ".jpg")
+            source = self.db.one(
+                "SELECT payload FROM asset_sources WHERE asset_id=?", (asset_id,)
+            )
+            kind = (
+                "generated"
+                if source and json.loads(source["payload"]).get("type") == "template"
+                else "images"
+            )
+            target = self.settings.data_dir / kind / (asset_id + ".jpg")
             self.path(asset_id).replace(target)
-            self.db.execute("UPDATE assets SET kind='images' WHERE id=?", (asset_id,))
+            self.db.execute("UPDATE assets SET kind=? WHERE id=?", (kind, asset_id))
         return self.info(asset_id)
 
     def info(self, asset_id):

pigway_frame_flow/renderers/photo_caption.py

@@ -41,7 +41,7 @@ def render_captioned_photo(
     draw = ImageDraw.Draw(canvas)
     values = [
         " ".join(str(caption.get(key, "")).split())
-        for key in ("date", "people", "title", "place", "filename")
+        for key in ("date", "people", "title", "place")
     ]
     text = " · ".join(value for value in values if value)
     font_size = layout.font_size

pigway_frame_flow/scheduler/engine.py

@@ -18,6 +18,7 @@ from pigway_frame_flow.models import (
     PlaylistInput,
     RestoreContent,
 )
+from pigway_frame_flow.scheduler.upload_retries import UploadRetries
 from pigway_frame_flow.storage.database import encode, now_iso
 
 
@@ -60,6 +61,7 @@ class Scheduler:
         self.thread = None
         self.webhook_thread = None
         self.tick_lock = threading.Lock()
+        self.upload_retries = UploadRetries(service)
 
     def jobs(self):
         rows = [
@@ -147,6 +149,7 @@ class Scheduler:
     def tick(self, current=None):
         current = current or datetime.now(UTC)
         with self.tick_lock:
+            self.upload_retries.tick(current)
             for row in self.db.all(
                 "SELECT * FROM jobs WHERE enabled=1 AND next_run<=?",
                 (current.isoformat(),),
@@ -183,6 +186,10 @@ class Scheduler:
                                 ),
                             ).model_dump(mode="json")
                         )
+                        if action == "display":
+                            payload = self.upload_retries.scheduled_payload(
+                                row["id"], payload
+                            )
                         operation_id = self.service.submit(
                             action,
                             payload,
@@ -191,6 +198,10 @@ class Scheduler:
                             "schedule-" + uuid4().hex,
                             "job-" + row["id"] + "-" + slot,
                         )
+                        if action == "display":
+                            self.upload_retries.scheduled_submitted(
+                                row["id"], operation_id
+                            )
                     following = next_run(job.schedule, current)
                     enabled = bool(following)
                     updated = job.model_copy(update={"enabled": enabled}).model_dump(

pigway_frame_flow/scheduler/upload_retries.py

@@ -0,0 +1,246 @@
+"""Durable, job-local upload retries; scheduled execution remains authoritative."""
+
+import json
+from datetime import datetime, timedelta
+
+from pigway_frame_flow.errors import FlowError
+from pigway_frame_flow.storage.database import encode
+
+
+class UploadRetries:
+    def __init__(self, service):
+        self.service, self.db = service, service.db
+
+    def load(self, job_id):
+        row = self.db.one(
+            "SELECT payload FROM app_config WHERE key=?", ("upload-retry:" + job_id,)
+        )
+        return json.loads(row["payload"]) if row else None
+
+    def save(self, job_id, plan):
+        if plan is None:
+            self.db.execute(
+                "DELETE FROM app_config WHERE key=?", ("upload-retry:" + job_id,)
+            )
+        else:
+            self.db.execute(
+                "INSERT OR REPLACE INTO app_config VALUES(?,?)",
+                ("upload-retry:" + job_id, encode(plan)),
+            )
+
+    @staticmethod
+    def reuse(payload, asset_id):
+        return {**payload, "content": {"type": "image", "asset_id": asset_id}}
+
+    def scheduled_payload(self, job_id, payload):
+        plan = self.load(job_id)
+        if plan and plan["phase"] in ("extra", "scheduled", "scheduled_running"):
+            return self.reuse(payload, plan["asset_id"])
+        return payload
+
+    def scheduled_submitted(self, job_id, operation_id):
+        plan = self.load(job_id)
+        if plan and plan["phase"] in ("extra", "scheduled", "scheduled_running"):
+            plan.update(phase="scheduled_running", active=operation_id)
+            self.save(job_id, plan)
+
+    def close_unknown(self, operation):
+        if operation["status"] == "UNKNOWN":
+            raw = self.db.one(
+                "SELECT * FROM operations WHERE id=?", (operation["operation_id"],)
+            )
+            driver = self.service.registry.get(operation["device_id"])
+            self.service.upload_reconciliation.check(
+                raw, driver, driver.device(refresh=True)
+            )
+            self.service.reconcile(operation["operation_id"], acknowledge=True)
+            return self.service.operation(operation["operation_id"])
+        return operation
+
+    def discard(self, asset_id):
+        row = self.db.one("SELECT kind FROM assets WHERE id=?", (asset_id,))
+        meta = self.db.one(
+            "SELECT starred FROM asset_metadata WHERE asset_id=?", (asset_id,)
+        )
+        # Only disposable generated staging files are owned by this recovery.
+        if row and row["kind"] == "cache" and not (meta and meta["starred"]):
+            self.service.assets.path(asset_id).unlink(missing_ok=True)
+            with self.db.transaction() as tx:
+                for table, field in (
+                    ("assets", "id"),
+                    ("asset_sources", "asset_id"),
+                    ("asset_metadata", "asset_id"),
+                ):
+                    tx.execute(f"DELETE FROM {table} WHERE {field}=?", (asset_id,))
+
+    def tick(self, current):
+        for job in self.db.all("SELECT * FROM jobs WHERE enabled=1"):
+            try:
+                self.check(job, current)
+            except FlowError:
+                # A read failure leaves the durable plan in place for a later check.
+                continue
+
+    def check(self, job, current):
+        config = json.loads(job["payload"])
+        if config["source"]["type"] not in (
+            "photo_source",
+            "auto_collage",
+            "template",
+            "provider",
+        ):
+            return
+        plan = self.load(job["id"])
+        if plan and plan["source"] != config["source"]:
+            self.save(job["id"], None)
+            plan = None
+        if not plan:
+            if not job["last_result"] or not job["next_run"]:
+                return
+            operation = self.service.operation(job["last_result"])
+            asset_id = (operation["result"] or {}).get("rendered_asset_id")
+            if operation["status"] not in ("FAILED", "UNKNOWN") or not asset_id:
+                return
+            deadline = datetime.fromisoformat(job["next_run"])
+            remaining = (deadline - current).total_seconds()
+            if remaining <= 0:
+                times = []
+            else:
+                # Five increasing delays occupy the first half of the available
+                # window, leaving time for upload/sync before the next schedule.
+                unit = remaining / 62
+                times = [
+                    (current + timedelta(seconds=unit * n)).isoformat()
+                    for n in (1, 3, 7, 15, 31)
+                ]
+            plan = dict(
+                root=operation["operation_id"],
+                active=operation["operation_id"],
+                asset_id=asset_id,
+                source=config["source"],
+                times=times,
+                count=0,
+                deadline=deadline.isoformat(),
+                phase="extra",
+            )
+            self.save(job["id"], plan)
+        operation = self.service.operation(plan["active"])
+        deadline = min(
+            datetime.fromisoformat(plan["deadline"]),
+            datetime.fromisoformat(job["next_run"]),
+        )
+        if (
+            plan["phase"] == "extra"
+            and current >= deadline
+            and operation["status"] != "COMPLETED"
+        ):
+            if operation["status"] == "SYNCING":
+                self.service.update(
+                    operation["operation_id"],
+                    "UNKNOWN",
+                    operation["result"],
+                    operation["error"],
+                )
+                operation = self.service.operation(operation["operation_id"])
+            operation = self.close_unknown(operation)
+            if operation["status"] == "QUEUED":
+                self.service.update(
+                    operation["operation_id"],
+                    "FAILED",
+                    error={
+                        "code": "CONFLICT",
+                        "message": "Upload retry window closed",
+                        "request_id": operation["request_id"],
+                        "details": {"reason": "RETRY_WINDOW_CLOSED"},
+                    },
+                )
+            if operation["status"] != "COMPLETED":
+                plan["phase"] = "scheduled"
+                self.save(job["id"], plan)
+                return
+        if operation["status"] == "COMPLETED":
+            self.save(job["id"], None)
+            # Extra retries are not job_runs and do not move the next plan time.
+            self.db.execute(
+                "UPDATE jobs SET last_result=? WHERE id=?",
+                (operation["operation_id"], job["id"]),
+            )
+            return
+        if operation["status"] not in ("FAILED", "UNKNOWN"):
+            return
+        if plan["phase"] == "replacement":
+            # A failed replacement gets its own bounded recovery window.
+            self.db.execute(
+                "UPDATE jobs SET last_result=? WHERE id=?",
+                (operation["operation_id"], job["id"]),
+            )
+            self.save(job["id"], None)
+            return
+        if plan["phase"] == "scheduled_running":
+            operation = self.close_unknown(operation)
+            if operation["status"] == "COMPLETED":
+                return
+            self.discard(plan["asset_id"])
+            payload = self.payload(config)
+            key = "replacement-" + plan["root"]
+            op_id = self.service.submit(
+                "display", payload, config["device_id"], "scheduler-retry", key, key
+            )
+            plan.update(phase="replacement", active=op_id)
+            self.save(job["id"], plan)
+            self.db.execute(
+                "UPDATE jobs SET last_result=? WHERE id=?", (op_id, job["id"])
+            )
+            self.db.event(
+                "upload.regenerated",
+                {
+                    "job_id": job["id"],
+                    "operation_id": op_id,
+                    "failed_asset_id": plan["asset_id"],
+                },
+            )
+            return
+        if current >= deadline or plan["count"] >= len(plan["times"]):
+            plan["phase"] = "scheduled"
+            self.save(job["id"], plan)
+            return
+        if current < datetime.fromisoformat(plan["times"][plan["count"]]):
+            return
+        operation = self.close_unknown(operation)
+        if operation["status"] == "COMPLETED":
+            return
+        number = plan["count"] + 1
+        key = f"upload-retry-{plan['root']}-{number}"
+        op_id = self.service.submit(
+            "display",
+            self.reuse(self.payload(config), plan["asset_id"]),
+            config["device_id"],
+            "scheduler-retry",
+            key,
+            key,
+        )
+        plan.update(active=op_id, count=number)
+        self.save(job["id"], plan)
+        self.db.execute("UPDATE jobs SET last_result=? WHERE id=?", (op_id, job["id"]))
+        self.db.event(
+            "upload.retry",
+            {
+                "job_id": job["id"],
+                "operation_id": op_id,
+                "attempt": number,
+                "deadline": deadline.isoformat(),
+            },
+        )
+
+    @staticmethod
+    def payload(config):
+        return {
+            "device_id": config["device_id"],
+            "content": config["source"],
+            "options": {
+                "duration": config["display_duration"],
+                "restore_previous": config["restore_previous"],
+                "priority": config["priority"],
+                "cleanup_unstarred": config.get("cleanup_unstarred", False),
+            },
+        }

pigway_frame_flow/web/app.js

@@ -10,6 +10,7 @@ async function loadConnection(){
 }
 const icons={dashboard:'◫',gallery:'▧','photo-sources':'▣',upload:'↥',templates:'▦',studio:'▦',automation:'◷',playlists:'▤',history:'↺',connections:'↔',settings:'⚙'};
 const t=k=>window.MESSAGES[state.lang][k]||k;
+function operationErrorMessage(error){const reason=error?.details?.reason;if(reason&&t(reason)!==reason)return t(reason);return error?.code&&t(error.code)!==error.code?t(error.code):(error?.message||t('error'));}
 const esc=v=>String(v??'—').replace(/[&<>"']/g,c=>({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;',"'":'&#39;'}[c]));
 const fmt=v=>v?new Date(v).toLocaleString(state.lang==='zh'?'zh-CN':'en-GB',{timeZone:'Asia/Shanghai'}):'—';
 const padTime=(hour,minute)=>`${String(hour).padStart(2,'0')}:${String(minute).padStart(2,'0')}`;
@@ -25,7 +26,7 @@ async function api(path,options={}){
  if(options.body&&!(options.body instanceof FormData)){headers['Content-Type']='application/json';options.body=JSON.stringify(options.body);}
  const response=await fetch(path,{...options,headers});
  if(!response.ok){let error;try{error=(await response.json()).error;}catch{}if(response.status===401&&error?.code!=='OFFICIAL_AUTH_REQUIRED')forgetLogin();
- const e=new Error((error?.code||response.status)+': '+(error?.code&&t(error.code)!==error.code?t(error.code):(error?.message||t('error'))));e.code=error?.code;throw e;}
+ const e=new Error((error?.code||response.status)+': '+operationErrorMessage(error));e.code=error?.code;throw e;}
  if(response.status===204)return null;
  return options.binary?await response.blob():await response.json();
 }
@@ -82,7 +83,7 @@ function login(){document.body.classList.add('login-mode');$('#page-title').text
 // LAN HTTP is not a secure context; getRandomValues remains available there.
 function clientUUID(){if(typeof crypto.randomUUID==='function')return crypto.randomUUID();const bytes=crypto.getRandomValues(new Uint8Array(16));bytes[6]=(bytes[6]&15)|64;bytes[8]=(bytes[8]&63)|128;const hex=Array.from(bytes,b=>b.toString(16).padStart(2,'0')).join('');return `${hex.slice(0,8)}-${hex.slice(8,12)}-${hex.slice(12,16)}-${hex.slice(16,20)}-${hex.slice(20)}`;}
 async function accepted(path,payload){const key='web-'+clientUUID();const result=await api(path,{method:'POST',body:payload,headers:{'Idempotency-Key':key}});toast(t('accepted'));return result.operation_id;}
-async function waitOperation(id){for(let n=0;n<120;n++){const operation=await api('/api/v1/operations/'+id);if(operation.status==='COMPLETED')return operation;if(['FAILED','UNKNOWN'].includes(operation.status))throw new Error(operation.status==='UNKNOWN'?t('unknown'):(operation.error?.message||t('error')));await new Promise(resolve=>setTimeout(resolve,2000));}throw new Error(t('syncing'));}
+async function waitOperation(id){for(let n=0;n<120;n++){const operation=await api('/api/v1/operations/'+id);if(operation.status==='COMPLETED')return operation;if(['FAILED','UNKNOWN'].includes(operation.status))throw new Error(operation.status==='UNKNOWN'?t('unknown'):operationErrorMessage(operation.error));await new Promise(resolve=>setTimeout(resolve,2000));}throw new Error(t('syncing'));}
 function showModal(html){
  const dialog=$('#modal'),content=$('#modal-content');content.innerHTML=html;
  const image=content.querySelector('img');dialog.classList.toggle('image-viewer',!!image);document.body.classList.toggle('image-viewer-open',!!image);dialog.onclose=()=>document.body.classList.remove('image-viewer-open');

pigway_frame_flow/web/i18n.js

@@ -20,8 +20,8 @@ Object.assign(MESSAGES.en,{officialContentTitle:'KoKonna official widgets and su
 Object.assign(MESSAGES.zh,{photoSources:'照片数据源',sourceHint:'通过 Synology Photos API 管理照片来源、人物和筛选规则。浏览与预览不会改变画框。',addSource:'添加数据源',browse:'浏览照片',randomPreview:'随机选图预览',sourceJob:'定时随机照片',noSources:'暂无照片，请添加数据源或检查目录。',sourceKind:'数据源类型',sourceLocation:'目录 / WebDAV URL',sourceLocationHint:'Synology Photos：所选空间内的目录路径，. 表示整个空间。按需获取 XL 预览图，不下载原图。其他云盘使用对应目录地址或 ID。',sourceCredential:'WebDAV 凭据环境变量名',sourceCredentialHint:'仅填写变量名，禁止填写密码。管理员通过环境变量或 Secret 配置 JSON username/password。',sourceRecursive:'包含子目录',sourceImages:'可用照片',sourceBrowseHint:'按需读取目录，点击文件名才生成预览；不会下载全部原图。',previousPage:'上一页',nextPage:'下一页',sourcePreviewHint:'这是已选中照片的预览，显示时使用同一张照片。',randomPhoto:'随机照片',sourceJobHint:'每次执行重新随机选图。默认暂停；启用后会按时改变画框显示。'});
 Object.assign(MESSAGES.en,{photoSources:'Photo sources',sourceHint:'Manage photo sources, people and filters through the Synology Photos API. Browsing and previewing do not change the frame.',addSource:'Add source',browse:'Browse photos',randomPreview:'Preview random photo',sourceJob:'Schedule random photos',noSources:'No photos yet. Add a source or check the folder.',sourceKind:'Source type',sourceLocation:'Folder / WebDAV URL',sourceLocationHint:'Synology Photos: folder path within the selected space; . means the whole space. Downloads XL previews on demand, not originals. Other drives use their folder URL or ID.',sourceCredential:'WebDAV credential environment name',sourceCredentialHint:'Enter a variable name, never a password. Admin configures JSON username/password via environment or Secret.',sourceRecursive:'Include subfolders',sourceImages:'Available photos',sourceBrowseHint:'Catalog loaded on demand. Click a filename to preview; originals are not downloaded in bulk.',previousPage:'Previous page',nextPage:'Next page',sourcePreviewHint:'Display uses this exact previewed photo.',randomPhoto:'Random photo',sourceJobHint:'Choose a fresh random photo each run. Paused by default; enabling changes the frame on schedule.'});
 
-Object.assign(MESSAGES.zh,{organizedPhotos:'仅日期开头文件夹中的照片',folderCaption:'照片下方显示文件夹文字',folderCaptionHint:'底部单行显示日期、已命名人物、主题、@ 后的地点和文件名。子目录沿用最近的日期文件夹；没有日期文件夹时保留文件名，未命名人物不显示。'});
-Object.assign(MESSAGES.en,{organizedPhotos:'Only photos in date-prefixed folders',folderCaption:'Show folder caption below the photo',folderCaptionHint:'One footer line shows date, named people, title, place after @ and filename. Subfolders inherit their nearest dated folder; otherwise the filename is kept. Unnamed people are omitted.'});
+Object.assign(MESSAGES.zh,{organizedPhotos:'仅日期开头文件夹中的照片',folderCaption:'照片下方显示文件夹文字',folderCaptionHint:'底部单行显示日期、已命名人物、主题和 @ 后的地点。子目录沿用最近的日期文件夹；未命名人物不显示，不显示原始文件名。'});
+Object.assign(MESSAGES.en,{organizedPhotos:'Only photos in date-prefixed folders',folderCaption:'Show folder caption below the photo',folderCaptionHint:'One footer line shows date, named people, title and place after @. Subfolders inherit their nearest dated folder; unnamed people and original filenames are omitted.'});
 
 Object.assign(MESSAGES.zh,{});Object.assign(MESSAGES.en,{});
 
@@ -146,3 +146,6 @@ Object.assign(MESSAGES.zh,{photosReconnectHelp:'尚未连接或会话已失效
 Object.assign(MESSAGES.en,{photosReconnectHelp:'Not connected or session expired. Photos API sources and people-filtered tasks need a new sign-in before reading photos.',photosSharedDisconnectWarning:'Disconnect the shared session used by all Photos API sources and people filters? Their tasks cannot read photos until reconnected. Folder settings and separate drive connections are retained.',photosConfigure:'Configure SYNOLOGY_PHOTOS_URL and SYNOLOGY_PHOTOS_CERT_SHA256. Photos and people are read through the API.'});
 
 Object.assign(window.MESSAGES.zh,{signIn:'登录',livingRoom:'客厅'});Object.assign(window.MESSAGES.en,{signIn:'Sign in',livingRoom:'Living room'});
+
+Object.assign(MESSAGES.zh,{RESTORE_IMAGE_PROTECTED:'这张图片仍用于恢复显示，暂时不能删除。请先结束临时展示，或显示另一张图片以更新恢复目标。'});
+Object.assign(MESSAGES.en,{RESTORE_IMAGE_PROTECTED:'This image is still needed to restore the display. End the temporary display or show another image to update the restore target before deleting it.'});

pigway_frame_flow/web/index.html

@@ -1,3 +1,3 @@
 <!doctype html>
 <html lang="zh-CN"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1,viewport-fit=cover"><meta name="color-scheme" content="light dark"><title>Pigway Frame Flow</title><link rel="stylesheet" href="/ui/style.css"></head>
-<body><aside class="sidebar"><a class="brand" href="/"><span class="brand-icon">F<span>↗</span></span><span>Frame Flow<small>Pigway · Smart Frame Automation</small></span></a><nav id="nav"></nav><div class="sidebar-bottom"><small>Version 0.1.11</small></div></aside><div class="workspace"><header><div><span class="eyebrow">PIGWAY FRAME FLOW</span><h1 id="page-title"></h1></div><div class="toolbar"><select id="device" aria-label="Device"></select><button id="lang" class="icon-button" aria-label="Change language">EN</button><button id="theme" class="icon-button" aria-label="Change theme">◐</button></div></header><main id="main"></main><footer id="footer"></footer></div><div id="toast" role="status" aria-live="polite"></div><dialog id="modal"><div id="modal-content"></div></dialog><script src="/ui/i18n.js?v=photo-caption-v0111"></script><script src="/ui/features.js?v=photo-caption-v0111"></script><script src="/ui/extensions.js?v=photo-caption-v0111"></script><script src="/ui/app.js?v=photo-caption-v0111"></script></body></html>
+<body><aside class="sidebar"><a class="brand" href="/"><span class="brand-icon">F<span>↗</span></span><span>Frame Flow<small>Pigway · Smart Frame Automation</small></span></a><nav id="nav"></nav><div class="sidebar-bottom"><small>Version 0.1.15</small></div></aside><div class="workspace"><header><div><span class="eyebrow">PIGWAY FRAME FLOW</span><h1 id="page-title"></h1></div><div class="toolbar"><select id="device" aria-label="Device"></select><button id="lang" class="icon-button" aria-label="Change language">EN</button><button id="theme" class="icon-button" aria-label="Change theme">◐</button></div></header><main id="main"></main><footer id="footer"></footer></div><div id="toast" role="status" aria-live="polite"></div><dialog id="modal"><div id="modal-content"></div></dialog><script src="/ui/i18n.js?v=upload-retries-v0115"></script><script src="/ui/features.js?v=upload-retries-v0115"></script><script src="/ui/extensions.js?v=upload-retries-v0115"></script><script src="/ui/app.js?v=upload-retries-v0115"></script></body></html>

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pigway-frame-flow"
-version = "0.1.11"
+version = "0.1.15"
 requires-python = ">=3.12"
 dependencies = ["paho-mqtt==2.1.0", "python-dateutil==2.9.0.post0", "fastapi==0.142.2", "uvicorn==0.54.0", "pillow==12.3.0", "requests==2.34.2", "croniter==6.2.4", "python-multipart==0.0.32", "pyyaml==6.0.3"]
 [project.optional-dependencies]

scripts/server-release.py

@@ -151,7 +151,7 @@ def release(lock):
             assert dst.execute("pragma quick_check").fetchone()[0] == "ok"
     base = "pigway-frame-flow:release-base-" + commit[:12]
     run("docker", "image", "tag", before["Image"], base)
-    image = "pigway-frame-flow:0.1.11-" + commit[:12]
+    image = "pigway-frame-flow:0.1.15-" + commit[:12]
     try:
         run(
             "docker",

tests/test_api.py

@@ -338,7 +338,8 @@ def test_unknown_same_idempotency_key_returns_original(client, service):
     fresh = client.post(
         "/api/v1/display", json=payload, headers={"Idempotency-Key": "new-key"}
     )
-    assert fresh.status_code == 409
+    assert fresh.status_code == 202
+    assert fresh.json()["operation_id"] != original.json()["operation_id"]
 
 
 def test_schema_contract(client):

tests/test_client.py

@@ -165,6 +165,23 @@ def test_transport_failure_records_safe_category_without_replay(exception, categ
     assert s.request.call_count == (3 if category == "connect_timeout" else 1)
 
 
+def test_connection_establishment_failure_is_unsent_and_bounded(transport_clock):
+    from urllib3.exceptions import MaxRetryError, NewConnectionError
+
+    s = session()
+    s.request.side_effect = requests.ConnectionError(
+        MaxRetryError(None, "/private", NewConnectionError(None, "private"))
+    )
+    c = KokonnaClient("secret", True, s)
+    with pytest.raises(FlowError) as error:
+        c.display(1)
+    assert error.value.code == "CLOUD_UNAVAILABLE"
+    assert error.value.details["write_outcome"] == "not_sent"
+    assert error.value.details["transport_error"] == "connect_error"
+    assert s.request.call_count == 3
+    assert "private" not in str(error.value.details)
+
+
 @pytest.mark.parametrize("write", [False, True])
 def test_connect_timeout_retries_then_succeeds(write, transport_clock):
     s = session({"success": True})

tests/test_display.py

@@ -71,22 +71,24 @@ def test_nested_overrides(service):
     assert service.registry.get("living-room").get_current_image() == 1
 
 
-def test_no_retry_unknown(service):
+def test_unknown_does_not_block_new_independent_upload(service):
     driver = service.registry.get("living-room")
+    attempts = []
 
-    def timeout(*a):
+    def timeout(*args):
+        attempts.append(args)
         raise FlowError(ErrorCode.WRITE_RESULT_UNKNOWN, "Write outcome unknown", 502)
 
     driver.upload = timeout
-    id = submit(service)
+    first = submit(service)
     service.process_one()
-    assert service.operation(id)["status"] == "UNKNOWN"
-    with pytest.raises(FlowError):
-        submit(service)
-    assert service.process_one() is False
-    assert service.reconcile(id)["resolved"] is False
-    assert service.reconcile(id, True)["resolved"] is True
-    assert service.operation(id)["status"] == "FAILED"
+    assert service.operation(first)["status"] == "UNKNOWN"
+    second = submit(service, priority=10)
+    assert second != first
+    assert service.process_one()
+    assert service.operation(first)["status"] == "FAILED"
+    assert service.operation(second)["status"] == "UNKNOWN"
+    assert len(attempts) == 2
 
 
 def test_restart_preserves_override(service):
@@ -145,6 +147,105 @@ def test_delete_restore_reference_protected(service):
     service.process_one()
     assert service.operation(delete)["status"] == "FAILED"
     assert service.operation(delete)["error"]["code"] == "CONFLICT"
+    assert (
+        service.operation(delete)["error"]["details"]["reason"]
+        == "RESTORE_IMAGE_PROTECTED"
+    )
+
+
+@pytest.mark.parametrize("automatic", [True, False])
+def test_completed_restore_consumes_self_reference(client, service, automatic):
+    display = submit(service, 60, True)
+    service.process_one()
+    service.sync()
+    if automatic:
+        expired(service, display)
+        service.expire()
+    else:
+        service.submit("restore", {}, "living-room", "admin", "restore-test")
+        service.process_one()
+    service.sync()
+    assert service.state("living-room")["previous_image_id"] is None
+    assert not client.get("/api/v1/devices/living-room").json()["can_restore"]
+    delete = service.submit(
+        "delete",
+        {"image_id": 1, "confirm_current": True},
+        "living-room",
+        "admin",
+        "delete-restored",
+    )
+    service.process_one()
+    assert service.operation(delete)["status"] == "COMPLETED"
+
+
+@pytest.mark.parametrize("via_delete", [False, True])
+def test_existing_self_reference_is_repaired(client, service, via_delete):
+    service.state("living-room")
+    service.db.execute(
+        "UPDATE device_state SET previous_image_id=1,managed_image_id=1 WHERE device_id=?",
+        ("living-room",),
+    )
+    if via_delete:
+        operation = service.submit(
+            "delete",
+            {"image_id": 1, "confirm_current": True},
+            "living-room",
+            "admin",
+            "delete-stale-reference",
+        )
+        service.process_one()
+        assert service.operation(operation)["status"] == "COMPLETED"
+    else:
+        assert not client.get("/api/v1/devices/living-room?refresh=true").json()[
+            "can_restore"
+        ]
+    assert service.state("living-room")["previous_image_id"] is None
+
+
+@pytest.mark.parametrize(
+    "protection",
+    ["override", "base", "unsynced", "unknown", "syncing", "other_previous"],
+)
+def test_self_reference_cleanup_preserves_real_restore_requirements(
+    service, protection
+):
+    service.state("living-room")
+    service.db.execute(
+        "UPDATE device_state SET previous_image_id=1,managed_image_id=1 WHERE device_id=?",
+        ("living-room",),
+    )
+    device = service.registry.get("living-room").device()
+    if protection == "override":
+        service.db.execute(
+            "INSERT INTO overrides VALUES(?,?,?,?,?,?)",
+            ("active", "living-room", 1, 50, "2099-01-01", "2026-01-01"),
+        )
+    elif protection == "base":
+        service.db.execute(
+            "UPDATE device_state SET base_image_id=1 WHERE device_id=?",
+            ("living-room",),
+        )
+    elif protection == "unsynced":
+        device = device.model_copy(
+            update={
+                "current_content": device.current_content.model_copy(
+                    update={"synced": False}
+                )
+            }
+        )
+    elif protection in ("unknown", "syncing"):
+        operation = submit(service)
+        service._start_write(operation)
+        service.update(operation, protection.upper())
+    else:
+        service.db.execute(
+            "UPDATE device_state SET previous_image_id=2 WHERE device_id=?",
+            ("living-room",),
+        )
+    service.clear_self_restore_reference("living-room", device)
+    assert service.state("living-room")["previous_image_id"] == (
+        2 if protection == "other_previous" else 1
+    )
 
 
 def test_offline(service):

tests/test_photo_watermark.py

@@ -7,7 +7,9 @@ from pigway_frame_flow.renderers.photo_caption import render_captioned_photo
 from pigway_frame_flow.renderers.photo_layout import PhotoLayout, WatermarkLayout
 
 
-def test_watermark_includes_filename_and_recognized_people(service, monkeypatch):
+def test_watermark_omits_raw_filename_and_includes_recognized_people(
+    service, monkeypatch
+):
     drawn = []
     original = ImageDraw.ImageDraw.text
 
@@ -24,12 +26,12 @@ def test_watermark_includes_filename_and_recognized_people(service, monkeypatch)
             "date": "2026-10-17",
             "title": "Family",
             "filename": "one.jpg",
-            "people": "Alice · Bob",
+            "people": "Alice & Bob",
         },
         service.renderer.font,
         size=(1600, 1200),
     )
-    assert drawn == ["2026-10-17 · Alice · Bob · Family · one.jpg"]
+    assert drawn == ["2026-10-17 · Alice & Bob · Family"]
 
 
 @pytest.mark.parametrize("size", [(1600, 1200), (1200, 1600)])

tests/test_synology_photos_api.py

@@ -106,7 +106,7 @@ def test_caption_people_use_actual_membership_and_cache(settings, monkeypatch):
 
 
 @pytest.mark.parametrize("labels_available", [True, False])
-def test_api_caption_keeps_filename_and_folder_with_optional_people(
+def test_api_caption_keeps_folder_and_optional_people_without_filename(
     service, settings, monkeypatch, labels_available
 ):
     session(settings)
@@ -133,8 +133,8 @@ def test_api_caption_keeps_filename_and_folder_with_optional_people(
     _, _, caption = provider.read("captions", "1")
     assert caption["date"] == "2026-10-17"
     assert caption["title"] == "Family"
-    assert caption["filename"] == "one.jpg"
-    assert caption.get("people", "") == ("Alice · Bob" if labels_available else "")
+    assert "filename" not in caption
+    assert caption.get("people", "") == ("Alice & Bob" if labels_available else "")
 
 
 def test_source_without_mount_pagination_scope_and_session(

tests/test_upload_retries.py

@@ -0,0 +1,154 @@
+from datetime import UTC, datetime, timedelta
+
+from pigway_frame_flow.errors import ErrorCode, FlowError
+from pigway_frame_flow.models import JobInput
+from pigway_frame_flow.scheduler.engine import Scheduler
+
+
+def failed_job(app, service, monkeypatch):
+    scheduler = app.state.scheduler
+    job = scheduler.save_job(
+        JobInput(
+            name="Retry",
+            schedule={"type": "interval", "interval_minutes": 120},
+            source={"type": "template", "template": "quote"},
+        )
+    )
+    now = datetime.now(UTC)
+    service.db.execute(
+        "UPDATE jobs SET next_run=? WHERE id=?",
+        ((now - timedelta(seconds=1)).isoformat(), job["id"]),
+    )
+    driver = service.registry.get("living-room")
+    upload = driver.upload
+
+    def fail(*a):
+        raise FlowError(ErrorCode.UPLOAD_FAILED, "Rejected", 502)
+
+    monkeypatch.setattr(driver, "upload", fail)
+    scheduler.tick(now)
+    service.process_one()
+    scheduler.tick(now + timedelta(seconds=1))
+    return scheduler, job, now, driver, upload
+
+
+def test_five_backoffs_then_one_scheduled_failure_regenerates(
+    app, service, monkeypatch, client
+):
+    scheduler, job, now, driver, upload = failed_job(app, service, monkeypatch)
+    policy = scheduler.upload_retries
+    plan = policy.load(job["id"])
+    asset = plan["asset_id"]
+    path = service.assets.path(asset)
+    original = path.read_bytes()
+    assert client.get("/api/v1/assets").json() == []
+    assert (
+        service.db.one("SELECT kind FROM assets WHERE id=?", (asset,))["kind"]
+        == "cache"
+    )
+    times = [datetime.fromisoformat(x) for x in plan["times"]]
+    gaps = [(times[0] - now).total_seconds()] + [
+        (b - a).total_seconds() for a, b in zip(times, times[1:], strict=False)
+    ]
+    assert gaps == sorted(gaps) and len(times) == 5
+    assert times[-1] < datetime.fromisoformat(plan["deadline"])
+    for number, when in enumerate(times, 1):
+        scheduler.tick(when + timedelta(milliseconds=1))
+        attempt = policy.load(job["id"])
+        assert attempt["count"] == number
+        op = service.operation(attempt["active"])
+        assert (
+            service.db.one(
+                "SELECT payload FROM operations WHERE id=?", (op["operation_id"],)
+            )["payload"].find(asset)
+            >= 0
+        )
+        service.process_one()
+        assert path.read_bytes() == original
+    assert len(service.db.all("SELECT * FROM job_runs")) == 1
+    # The plan survives a process restart, without resetting the five-attempt cap.
+    scheduler = Scheduler(service)
+    deadline = datetime.fromisoformat(plan["deadline"])
+    scheduler.tick(deadline + timedelta(milliseconds=1))
+    planned = scheduler.upload_retries.load(job["id"])
+    assert planned["phase"] == "scheduled_running"
+    service.process_one()
+    scheduler.tick(deadline + timedelta(seconds=1))
+    replacement = scheduler.upload_retries.load(job["id"])
+    assert replacement["phase"] == "replacement"
+    assert not path.exists()
+    assert service.db.one("SELECT id FROM assets WHERE id=?", (asset,)) is None
+    monkeypatch.setattr(driver, "upload", upload)
+    service.process_one()
+    service.sync()
+    op = service.operation(replacement["active"])
+    assert op["status"] == "COMPLETED"
+    assert op["result"]["rendered_asset_id"] != asset
+    assert len(service.db.all("SELECT * FROM job_runs")) == 2
+
+
+def test_retry_success_cancels_rest_and_keeps_schedule(app, service, monkeypatch):
+    scheduler, job, now, driver, upload = failed_job(app, service, monkeypatch)
+    policy = scheduler.upload_retries
+    plan = policy.load(job["id"])
+    deadline = service.db.one("SELECT next_run FROM jobs WHERE id=?", (job["id"],))[
+        "next_run"
+    ]
+    monkeypatch.setattr(driver, "upload", upload)
+    when = datetime.fromisoformat(plan["times"][0]) + timedelta(milliseconds=1)
+    scheduler.tick(when)
+    operation = policy.load(job["id"])["active"]
+    service.process_one()
+    service.sync()
+    scheduler.tick(when + timedelta(seconds=1))
+    assert policy.load(job["id"]) is None
+    assert service.operation(operation)["status"] == "COMPLETED"
+    assert (
+        service.db.one("SELECT next_run FROM jobs WHERE id=?", (job["id"],))["next_run"]
+        == deadline
+    )
+    assert (
+        service.db.one("SELECT kind FROM assets WHERE id=?", (plan["asset_id"],))[
+            "kind"
+        ]
+        == "generated"
+    )
+    assert driver.client.uploads == 1
+
+
+def test_expired_queued_retry_never_uploads(app, service, monkeypatch):
+    scheduler, job, now, driver, upload = failed_job(app, service, monkeypatch)
+    policy = scheduler.upload_retries
+    plan = policy.load(job["id"])
+    scheduler.tick(datetime.fromisoformat(plan["times"][0]) + timedelta(milliseconds=1))
+    plan = policy.load(job["id"])
+    plan["deadline"] = (datetime.now(UTC) - timedelta(seconds=1)).isoformat()
+    policy.save(job["id"], plan)
+
+    def unexpected(*a):
+        raise AssertionError("Expired retry must never upload")
+
+    monkeypatch.setattr(driver, "upload", unexpected)
+    service.process_one()
+    operation = service.operation(plan["active"])
+    assert operation["status"] == "FAILED"
+    assert operation["error"]["details"]["reason"] == "RETRY_WINDOW_CLOSED"
+    assert service.assets.path(plan["asset_id"]).exists()
+
+
+def test_lost_response_is_reconciled_before_retry_without_duplicate(
+    app, service, monkeypatch
+):
+    scheduler, job, now, driver, upload = failed_job(app, service, monkeypatch)
+    policy = scheduler.upload_retries
+    plan = policy.load(job["id"])
+    asset = service.assets.info(plan["asset_id"])
+    monkeypatch.setattr(driver, "upload", upload)
+    # The original request reached the cloud, but its response was lost.
+    driver.upload(service.assets.path(plan["asset_id"]), asset["name"])
+    service.update(plan["root"], "UNKNOWN")
+    scheduler.tick(datetime.fromisoformat(plan["times"][0]) + timedelta(milliseconds=1))
+    scheduler.tick(datetime.fromisoformat(plan["times"][0]) + timedelta(seconds=1))
+    assert service.operation(plan["root"])["status"] == "COMPLETED"
+    assert policy.load(job["id"]) is None
+    assert driver.client.uploads == 1