Add documented KoKonna failure policy, bounded retries and durable recovery escalation [skip ci]

docs/KOKONNA_RECOVERY.md

@@ -0,0 +1,61 @@
+# KoKonna 错误、重试与阻塞处理
+
+核对日期：2026-10-16。依据：[KoKonna OpenAPI 官方文档](https://kokonna.art/pages/openapi)、[Requests 异常定义](https://requests.readthedocs.io/en/latest/_modules/requests/exceptions/)。本页区分厂商契约与 Frame Flow 的本地策略；没有文档保证的行为不作为自动重发依据。
+
+## 官方接口契约
+
+- 设备 API Key 使用 Bearer 鉴权；单设备每分钟上限 20 次，请求超限返回 429。
+- 上传成功返回新图片 ID 和计数，并立即激活新图；接口不提供幂等键或上传状态查询契约。
+- 删除可能改变当前图片。切换与删除都不能视为天然可重放。
+- 404 可表示方法不存在；图片下载时也可表示目标图片不存在。
+- 500 既可能是通用服务故障，也可能明确表示设备密钥无效、图片不存在或图片 ID 参数错误。实现只匹配官方明确列出的签名，其他 500 不猜测是否产生副作用。
+- 上传请求体上限 50 MB；本服务原图限制仍为 20 MiB，并保留像素上限和 base64 检查。
+- 官方文档未承诺服务处理时长、超时后的取消、事务回滚或 Retry-After 返回；不能靠等待固定时间证明写入没有执行。
+- Requests 将连接建立超时定义为可安全重试；读取超时不提供相同保证。一般连接异常也不能推定请求未发送。
+
+## 已落实的本地策略
+
+| 结果 | 处理 | 自动重试 | 设备阻塞 |
+| --- | --- | --- | --- |
+| 未配置密钥 | CONFIG_REQUIRED | 无 | 无 |
+| 明确的设备密钥无效 | CONFIG_REQUIRED，保存固定错误分类 | 无 | 无 |
+| 明确的图片不存在 | NOT_FOUND | 无 | 无 |
+| 明确的图片 ID 参数错误 | INVALID_REQUEST | 无 | 无 |
+| 明确的方法不存在 | UNSUPPORTED_CAPABILITY | 无 | 无 |
+| 其他 HTTP 4xx 拒绝 | 保存状态码，结束失败 | 除 429 外无 | 无 |
+| 本地请求预算耗尽 | 等待预算窗口 | 同一请求最多 3 次尝试 | 不创建 UNKNOWN |
+| HTTP 429 | RATE_LIMITED，尊重服务端冷却时间 | 最多 3 次尝试 | 耗尽后失败，不创建 UNKNOWN |
+| ConnectTimeout | 确定请求未发送 | 读写均最多 3 次尝试 | 耗尽后失败，不创建 UNKNOWN |
+| 只读查询的临时网络错误、通用 5xx、无效 JSON | CLOUD_UNAVAILABLE | 最多 3 次尝试 | 不创建 UNKNOWN |
+| TLS 异常 | 保存固定分类 | 无 | 写请求仍保守标为 UNKNOWN |
+| 写请求读取超时、一般连接异常、通用 5xx | 无法确定写入结果 | 不重发，只读核对 | UNKNOWN，保留保护 |
+| 写请求无效响应或无有效上传 ID | 无法确认成功 | 不重发，只读核对 | UNKNOWN，保留保护 |
+
+“最多 3 次”包含首次请求，重试发生在同一传输请求内，不重新渲染、不重新执行整个操作。普通退避为 1、2 秒；每次实际请求都计入本地 18 次/分钟预算。429 优先使用 Retry-After 秒数或 HTTP 日期；缺失/无效时按 60 秒。单次等待不超过 60 秒，超过此值保留冷却并结束当前尝试，不提前请求。没有无限循环，也不在重启后重放不确定的写入。
+
+操作错误的 details 保存 attempts、retryable、write_outcome，以及 transport_error / http_status / failure_kind 等固定分类。不会保存异常原文、完整下载 URL、带密钥的厂商消息或响应正文。是否重试由结果类别决定，不能仅依据 HTTP 500。
+
+## 未知写入的核对与升级
+
+上传前记录资产、哈希、字节数、名称、上传前图库和时间。响应丢失后只读查找并校验唯一候选；图片相同且设备已同步才能记为成功。已证明上传、但设备持续停留原图时，沿用两次稳定观察的失败关闭规则。
+
+没有候选、候选不唯一、内容不符、设备未同步、查询失败或缺少历史证据，都有独立恢复原因。图库没有目标图片不等于厂商保证从未接受请求，因此不自动把它改为失败或重传。
+
+`GET /api/v1/operations/{operation_id}` 的 `recovery` 返回：
+
+- `state`：checking / manual_review / resolved。
+- `reason`：gallery_match_missing、gallery_match_ambiguous、content_mismatch、waiting_device_sync、target_mismatch、read_unavailable、reconciliation_error、missing_upload_evidence 或 manual_reconciliation_required。
+- `first_checked_at`、`last_checked_at`、`checks`、`requires_manual_review`。
+
+只读核对每条操作至多每分钟一次。操作持续未知超过 sync_timeout + 300 秒（默认约 15 分钟）时，持久化一次 `operation.blocked` 事件和 `MANUAL_REVIEW_REQUIRED` 历史记录；服务重启、重复检查或原因改变均不会重复升级同一条未解决操作。既有检查继续运行，不自动取消设备保护。
+
+管理员可在 Webhooks 中订阅 `operation.blocked`；未配置订阅不会自动向外发送通知。事件和 API 记录始终保留，但它们不能替代已配置的通知渠道。单个任务失败、单条未知操作的元数据异常均不应中断其他任务处理。
+
+## 人工处理
+
+1. 根据 operation_id 查看原始错误、恢复原因、当前设备和图库；不点击重新执行来猜测结果。
+2. 调用现有 reconcile 接口并传 `acknowledge=false` 进行核对。
+3. 无法证明成功、但管理员确认可以结束旧操作时，传 `acknowledge=true`：结束为失败，保留原始目标、上传意图和此前错误，记录观察到的图片与人工处理历史。该接口不会重发设备写入。
+4. 有目标一致且同步完成的证据才记为成功。处理后恢复原计划，错过的任务不补发。
+
+仍可能因厂商无法给出明确写入结果而需要人工处理；本策略解决错误误判、可安全重试的暂时失败和长期阻塞缺乏解释的问题，不承诺网络故障时所有设备任务都能自动成功。

docs/REAL_DEVICE_TEST.md

@@ -41,3 +41,7 @@ minute apart record `DISPLAY_NOT_CONFIRMED` and unblock future scheduled work.
 Temporary overrides retain their restore behavior and are not automatically
 closed as failed. Legacy operations without durable upload evidence require
 manual acknowledgement; absence from the gallery alone never proves failure.
+
+### 错误与有限重试
+
+完整错误矩阵、重试边界、持久化恢复状态和人工处理见 [KoKonna 恢复策略](KOKONNA_RECOVERY.md)。自动重试只允许只读请求、连接建立超时或明确限流拒绝；读取超时等未知写入不会重发。

docs/RESEARCH.md

@@ -1,6 +1,7 @@
 # KoKonna API research / 接口调查
 
-Checked 2026-10-01 against https://kokonna.art/pages/openapi.
+Rechecked 2026-10-16 against https://kokonna.art/pages/openapi.
+错误分类、重试和未知写入处理见 [KoKonna 恢复策略](KOKONNA_RECOVERY.md)。
 Base: https://api.kokonna.art/openapi. No fallback to legacy galaxyguide.cn.
 
 | Official endpoint | Frame Flow capability |
@@ -29,7 +30,7 @@ headroom at 18. Device state has a 60-second cache, gallery is loaded on demand.
 Uploads automatically activate a new image; there is no upload-without-activation
 flag. Gallery-only uploads are therefore local staging, not a fictitious cloud
 operation. Maximum upstream request size is 50 MB; this service accepts 20 MiB
-image files with bounded decoded pixels. No non-idempotent cloud write is retried.
+image files with bounded decoded pixels. Only proven-unsent connections or explicit rate-limit rejections may retry a write, within a three-attempt budget. Ambiguous writes are never replayed.
 Cloud acknowledgement is not physical refresh confirmation; /device synced and
 imageId drive the operation state. Unconfirmed or interrupted writes become UNKNOWN
 and block later device writes until read-only reconciliation/explicit acknowledgement.

docs/openapi.json

@@ -2,7 +2,7 @@
   "openapi": "3.1.0",
   "info": {
     "title": "Pigway Frame Flow API",
-    "version": "0.1.0"
+    "version": "0.1.4"
   },
   "paths": {
     "/api/v1/devices/{device_id}/widgets/selection": {
@@ -6397,7 +6397,15 @@
           "content": {
             "application/json": {
               "schema": {
-                "$ref": "#/components/schemas/TemplateContent"
+                "anyOf": [
+                  {
+                    "$ref": "#/components/schemas/TemplateContent"
+                  },
+                  {
+                    "$ref": "#/components/schemas/AutoCollageContent"
+                  }
+                ],
+                "title": "Payload"
               }
             }
           }
@@ -14298,6 +14306,87 @@
         ],
         "title": "AssetBatch"
       },
+      "AutoCollageContent": {
+        "properties": {
+          "type": {
+            "type": "string",
+            "const": "auto_collage",
+            "title": "Type"
+          },
+          "source_id": {
+            "anyOf": [
+              {
+                "type": "string",
+                "pattern": "^[a-z0-9-]{1,60}$"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Source Id"
+          },
+          "count": {
+            "type": "integer",
+            "maximum": 12.0,
+            "minimum": 1.0,
+            "title": "Count",
+            "default": 4
+          },
+          "title": {
+            "type": "string",
+            "maxLength": 100,
+            "title": "Title",
+            "default": ""
+          },
+          "message": {
+            "type": "string",
+            "maxLength": 500,
+            "title": "Message",
+            "default": ""
+          },
+          "orientation": {
+            "type": "string",
+            "enum": [
+              "auto",
+              "portrait",
+              "landscape"
+            ],
+            "title": "Orientation",
+            "default": "auto"
+          },
+          "style": {
+            "type": "string",
+            "enum": [
+              "grid",
+              "hero",
+              "poster"
+            ],
+            "title": "Style",
+            "default": "grid"
+          },
+          "use_ai": {
+            "type": "boolean",
+            "title": "Use Ai",
+            "default": true
+          },
+          "prompt": {
+            "type": "string",
+            "maxLength": 1500,
+            "title": "Prompt",
+            "default": ""
+          },
+          "profile": {
+            "$ref": "#/components/schemas/Profile",
+            "default": "color"
+          }
+        },
+        "additionalProperties": false,
+        "type": "object",
+        "required": [
+          "type"
+        ],
+        "title": "AutoCollageContent"
+      },
       "Body_import_images_api_v1_layout_images_post": {
         "properties": {
           "files": {
@@ -14778,6 +14867,9 @@
               {
                 "$ref": "#/components/schemas/PhotoSourceContent"
               },
+              {
+                "$ref": "#/components/schemas/AutoCollageContent"
+              },
               {
                 "$ref": "#/components/schemas/ProviderContent"
               },
@@ -14789,6 +14881,7 @@
             "discriminator": {
               "propertyName": "type",
               "mapping": {
+                "auto_collage": "#/components/schemas/AutoCollageContent",
                 "html": "#/components/schemas/FutureContent",
                 "image": "#/components/schemas/ImageContent",
                 "image_url": "#/components/schemas/URLContent",
@@ -14852,6 +14945,7 @@
           "UNSUPPORTED_CAPABILITY",
           "RENDER_FAILED",
           "UPLOAD_FAILED",
+          "DISPLAY_NOT_CONFIRMED",
           "WRITE_RESULT_UNKNOWN",
           "RATE_LIMITED",
           "INTERNAL_ERROR",
@@ -15062,7 +15156,7 @@
           "version": {
             "type": "string",
             "title": "Version",
-            "default": "0.1.0"
+            "default": "0.1.4"
           }
         },
         "type": "object",
@@ -15324,6 +15418,9 @@
                   {
                     "$ref": "#/components/schemas/PhotoSourceContent"
                   },
+                  {
+                    "$ref": "#/components/schemas/AutoCollageContent"
+                  },
                   {
                     "$ref": "#/components/schemas/ProviderContent"
                   },
@@ -15334,6 +15431,7 @@
                 "discriminator": {
                   "propertyName": "type",
                   "mapping": {
+                    "auto_collage": "#/components/schemas/AutoCollageContent",
                     "html": "#/components/schemas/FutureContent",
                     "image": "#/components/schemas/ImageContent",
                     "image_url": "#/components/schemas/URLContent",
@@ -15469,6 +15567,9 @@
                   {
                     "$ref": "#/components/schemas/PhotoSourceContent"
                   },
+                  {
+                    "$ref": "#/components/schemas/AutoCollageContent"
+                  },
                   {
                     "$ref": "#/components/schemas/ProviderContent"
                   },
@@ -15479,6 +15580,7 @@
                 "discriminator": {
                   "propertyName": "type",
                   "mapping": {
+                    "auto_collage": "#/components/schemas/AutoCollageContent",
                     "html": "#/components/schemas/FutureContent",
                     "image": "#/components/schemas/ImageContent",
                     "image_url": "#/components/schemas/URLContent",
@@ -15787,6 +15889,16 @@
       },
       "Operation": {
         "properties": {
+          "recovery": {
+            "anyOf": [
+              {
+                "$ref": "#/components/schemas/RecoveryState"
+              },
+              {
+                "type": "null"
+              }
+            ]
+          },
           "operation_id": {
             "type": "string",
             "title": "Operation Id"
@@ -16808,6 +16920,49 @@
         ],
         "title": "Reconciliation"
       },
+      "RecoveryState": {
+        "properties": {
+          "state": {
+            "type": "string",
+            "enum": [
+              "checking",
+              "manual_review",
+              "resolved"
+            ],
+            "title": "State"
+          },
+          "reason": {
+            "type": "string",
+            "title": "Reason"
+          },
+          "first_checked_at": {
+            "type": "string",
+            "title": "First Checked At"
+          },
+          "last_checked_at": {
+            "type": "string",
+            "title": "Last Checked At"
+          },
+          "checks": {
+            "type": "integer",
+            "title": "Checks"
+          },
+          "requires_manual_review": {
+            "type": "boolean",
+            "title": "Requires Manual Review"
+          }
+        },
+        "type": "object",
+        "required": [
+          "state",
+          "reason",
+          "first_checked_at",
+          "last_checked_at",
+          "checks",
+          "requires_manual_review"
+        ],
+        "title": "RecoveryState"
+      },
       "Region": {
         "properties": {
           "asset_id": {
@@ -17178,7 +17333,7 @@
           "version": {
             "type": "string",
             "title": "Version",
-            "default": "0.1.0"
+            "default": "0.1.4"
           },
           "api_version": {
             "type": "string",

pigway_frame_flow/api/app.py

@@ -124,7 +124,7 @@ def create_app(settings=None, start_worker=True):
     app = FastAPI(
         title="Pigway Frame Flow API",
         docs_url=None,
-        version="0.1.2",
+        version="0.1.4",
         lifespan=lifespan,
         responses=error_responses,
     )
@@ -1000,7 +1000,7 @@ def create_app(settings=None, start_worker=True):
     )
     def settings_view():
         return SettingsView(
-            version="0.1.2",
+            version="0.1.4",
             mock=settings.mock,
             write_enabled=settings.write_enabled,
             api_key_configured=bool(settings.api_key)

pigway_frame_flow/display/recovery.py

@@ -0,0 +1,73 @@
+"""Durable, read-only recovery progress for writes with uncertain outcomes."""
+
+from datetime import UTC, datetime
+from uuid import uuid4
+
+from pigway_frame_flow.storage.database import encode, now_iso
+
+
+class RecoveryMonitor:
+    def __init__(self, service):
+        self.service, self.db = service, service.db
+
+    def observe(self, op, reason):
+        timestamp = now_iso()
+        age = (
+            datetime.now(UTC) - datetime.fromisoformat(op["created_at"])
+        ).total_seconds()
+        review = age >= self.service.settings.sync_timeout + 300
+        with self.db.transaction() as tx:
+            previous = tx.execute(
+                "SELECT * FROM operation_recovery WHERE operation_id=?", (op["id"],)
+            ).fetchone()
+            # Once escalated, wall-clock corrections must not clear the alert.
+            review = review or bool(previous and previous["state"] == "manual_review")
+            state = "manual_review" if review else "checking"
+            tx.execute(
+                "INSERT INTO operation_recovery VALUES(?,?,?,?,?,1) ON CONFLICT(operation_id) DO UPDATE SET state=excluded.state,reason=excluded.reason,last_checked_at=excluded.last_checked_at,checks=operation_recovery.checks+1",
+                (op["id"], state, reason, timestamp, timestamp),
+            )
+            if review and (not previous or previous["state"] != "manual_review"):
+                tx.execute(
+                    "INSERT INTO events VALUES(?,?,?,?)",
+                    (
+                        uuid4().hex,
+                        "operation.blocked",
+                        encode(
+                            {
+                                "operation_id": op["id"],
+                                "device_id": op["device_id"],
+                                "reason": reason,
+                                "requires_manual_review": True,
+                            }
+                        ),
+                        timestamp,
+                    ),
+                )
+                tx.execute(
+                    "INSERT INTO history(client_id,request_id,operation_id,device_id,action,result,timestamp) VALUES(?,?,?,?,?,?,?)",
+                    (
+                        op["client_id"],
+                        op["request_id"],
+                        op["id"],
+                        op["device_id"],
+                        op["action"],
+                        "MANUAL_REVIEW_REQUIRED",
+                        timestamp,
+                    ),
+                )
+
+    def resolve(self, operation_id):
+        self.db.execute(
+            "UPDATE operation_recovery SET state='resolved' WHERE operation_id=? AND state!='resolved'",
+            (operation_id,),
+        )
+
+    def info(self, operation_id):
+        row = self.db.one(
+            "SELECT state,reason,first_checked_at,last_checked_at,checks FROM operation_recovery WHERE operation_id=?",
+            (operation_id,),
+        )
+        if row:
+            row["requires_manual_review"] = row["state"] == "manual_review"
+        return row

pigway_frame_flow/display/service.py

@@ -6,6 +6,7 @@ from datetime import UTC, datetime, timedelta
 from uuid import uuid4
 
 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
@@ -50,6 +51,7 @@ class DisplayService:
         self.lock = threading.RLock()
         self.retention = Retention(self)
         self.cloud_library = CloudLibrary(self)
+        self.recovery = RecoveryMonitor(self)
         self.upload_reconciliation = UploadReconciliation(self)
         self.expiration_checked: dict[str, float] = {}
         self.unknown_checked: dict[str, float] = {}
@@ -165,6 +167,7 @@ class DisplayService:
             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):
@@ -184,6 +187,8 @@ class DisplayService:
                 id,
             ),
         )
+        if status in ("COMPLETED", "FAILED"):
+            self.recovery.resolve(id)
 
     def blocked(self, device_id):
         return (
@@ -442,7 +447,7 @@ class DisplayService:
                         "code": exc.code,
                         "message": exc.message,
                         "request_id": op["request_id"],
-                        "details": {},
+                        "details": exc.details,
                     }
                     self.update(
                         op["id"], "UNKNOWN" if uncertain else "FAILED", error=error
@@ -702,21 +707,34 @@ class DisplayService:
         # 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='display' AND write_started=1"
+            "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")
-            if not expected and not intent:
-                continue
             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)
                 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
@@ -727,8 +745,18 @@ class DisplayService:
                     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:
-                pass
+                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()
@@ -861,7 +889,20 @@ class DisplayService:
                 self.update(
                     id,
                     "COMPLETED" if confirmed else "FAILED",
-                    {"image_id": device.current_content.id},
+                    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":

pigway_frame_flow/display/upload_reconciliation.py

@@ -59,6 +59,12 @@ class UploadReconciliation:
                     candidates.append(image)
             # Ambiguous matches never release a write lock.
             if len(candidates) != 1:
+                service.recovery.observe(
+                    op,
+                    "gallery_match_missing"
+                    if not candidates
+                    else "gallery_match_ambiguous",
+                )
                 return None
             expected = candidates[0].image_id
             data = driver.download(expected)
@@ -66,6 +72,7 @@ class UploadReconciliation:
                 len(data) != intent["size"]
                 or hashlib.sha256(data).hexdigest() != intent["sha256"]
             ):
+                service.recovery.observe(op, "content_mismatch")
                 return None
             result["image_id"] = expected
             result["upload_confirmed"] = True

pigway_frame_flow/drivers/kokonna/client.py

@@ -1,7 +1,7 @@
 """Hardened adapter over the MIT-0 guooscar 1.2.0 device/upload methods.
 
 No UI, scheduler, renderer, environment lookup, or constructor cloud calls.
-Never retry a write; a timeout may have occurred after cloud acceptance.
+Retry only read-only requests or writes proven unsent/rejected; never replay ambiguous writes.
 """
 
 import threading
@@ -19,28 +19,85 @@ BASE_URL = "https://api.kokonna.art/openapi"
 
 
 class SafeTransport:
+    MAX_ATTEMPTS = 3
+    MAX_RETRY_DELAY = 60
+
     def __init__(self, key, session=None):
         self.key = key
         self.session = session or requests.Session()
         self.calls: deque[float] = deque()
         self.lock = threading.Lock()
+        self.cooldown_until = 0.0
 
     def request(self, endpoint, payload=None, write=False, binary=False):
-        if not self.key:
-            raise FlowError(E.CONFIG_REQUIRED, "KoKonna configuration required", 503)
+        for attempt in range(1, self.MAX_ATTEMPTS + 1):
+            try:
+                return self._request_once(endpoint, payload, write, binary)
+            except FlowError as exc:
+                exc.details["attempts"] = attempt
+                delay = max(
+                    2 ** (attempt - 1), exc.details.get("retry_after_seconds", 0)
+                )
+                if (
+                    not exc.details.get("retryable")
+                    or attempt == self.MAX_ATTEMPTS
+                    or delay > self.MAX_RETRY_DELAY
+                ):
+                    if exc.details.get("retryable"):
+                        exc.details["retry_exhausted"] = attempt == self.MAX_ATTEMPTS
+                        exc.details["retry_deferred"] = delay > self.MAX_RETRY_DELAY
+                    raise
+                time.sleep(delay)
+        raise AssertionError("Retry loop exhausted")
+
+    def _reserve(self):
         with self.lock:
             current = time.monotonic()
             while self.calls and current - self.calls[0] >= 60:
                 self.calls.popleft()
-            # Reserve capacity beneath the official 20/minute ceiling.
+            delay = max(0.0, self.cooldown_until - current)
             if len(self.calls) >= 18:
-                raise FlowError(E.RATE_LIMITED, "Device request budget exhausted", 429)
+                delay = max(delay, 60 - (current - self.calls[0]))
+            if delay:
+                raise FlowError(
+                    E.RATE_LIMITED,
+                    "Device request budget exhausted",
+                    429,
+                    details={
+                        "retryable": True,
+                        "retry_after_seconds": delay,
+                        "write_outcome": "not_sent",
+                        "failure_kind": "local_rate_limit",
+                    },
+                )
             self.calls.append(current)
-        url = BASE_URL + endpoint
+
+    @staticmethod
+    def _retry_after(value):
+        from datetime import UTC, datetime
+        from email.utils import parsedate_to_datetime
+
+        try:
+            seconds = float(value)
+        except (ValueError, TypeError):
+            try:
+                seconds = (
+                    parsedate_to_datetime(value) - datetime.now(UTC)
+                ).total_seconds()
+            except (ValueError, TypeError, OverflowError):
+                return 60.0
+        import math
+
+        return max(0.0, seconds) if math.isfinite(seconds) else 60.0
+
+    def _request_once(self, endpoint, payload, write, binary):
+        if not self.key:
+            raise FlowError(E.CONFIG_REQUIRED, "KoKonna configuration required", 503)
+        self._reserve()
         try:
             response = self.session.request(
                 "GET" if binary else "POST",
-                url,
+                BASE_URL + endpoint,
                 headers={}
                 if binary
                 else {
@@ -51,49 +108,140 @@ class SafeTransport:
                 timeout=(5, 120 if endpoint == "/upload" else 30),
                 allow_redirects=False,
             )
-        except requests.RequestException:
+        except requests.RequestException as exc:
+            # Never retain exception text or a URL: binary URLs contain the key.
+            category = next(
+                (
+                    label
+                    for kind, label in (
+                        (requests.ConnectTimeout, "connect_timeout"),
+                        (requests.ReadTimeout, "read_timeout"),
+                        (requests.exceptions.SSLError, "tls_error"),
+                        (requests.ConnectionError, "connection_error"),
+                        (requests.Timeout, "timeout"),
+                    )
+                    if isinstance(exc, kind)
+                ),
+                "request_error",
+            )
+            not_sent = isinstance(exc, requests.ConnectTimeout)
+            uncertain = write and not not_sent
             raise FlowError(
-                E.WRITE_RESULT_UNKNOWN if write else E.CLOUD_UNAVAILABLE,
+                E.WRITE_RESULT_UNKNOWN if uncertain else E.CLOUD_UNAVAILABLE,
                 "Cloud write outcome is unknown"
-                if write
+                if uncertain
                 else "Device cloud is unavailable",
                 502,
+                details={
+                    "transport_error": category,
+                    "retryable": (not write and category != "tls_error") or not_sent,
+                    "write_outcome": "unknown"
+                    if uncertain
+                    else ("not_sent" if write else "read_only"),
+                },
             ) from None
-        if response.status_code != 200:
-            code = (
-                E.RATE_LIMITED
-                if response.status_code == 429
-                else (
-                    E.WRITE_RESULT_UNKNOWN
-                    if write and response.status_code >= 500
-                    else E.CLOUD_UNAVAILABLE
-                )
-            )
-            raise FlowError(
-                code,
-                "Device cloud request failed",
-                429 if code == E.RATE_LIMITED else 502,
-            )
-        if binary:
-            if len(response.content) > 20 * 1024 * 1024:
-                raise FlowError(E.INVALID_REQUEST, "Cloud image exceeds size limit")
-            return response.content
         try:
-            value = response.json()
-            if not isinstance(value, dict):
-                raise ValueError()
-            # Cloud errors may echo credentials: never return raw messages.
-            if value.get("success") is False or (
-                "message" in value and "id" not in value
+            value = None
+            if not binary or response.status_code != 200:
+                try:
+                    value = response.json()
+                except ValueError:
+                    pass
+            message = value.get("message") if isinstance(value, dict) else None
+            status = response.status_code
+            # Only exact documented business failures establish rejection. Generic
+            # 5xx responses may arrive after a write; those remain UNKNOWN.
+            business = None
+            if status in (404, 500) and isinstance(message, str):
+                if message == "image not found":
+                    business = (
+                        E.NOT_FOUND,
+                        "Cloud image not found",
+                        404,
+                        "image_not_found",
+                    )
+                elif status == 500 and message == "imageId is empty":
+                    business = (
+                        E.INVALID_REQUEST,
+                        "Cloud image ID is invalid",
+                        400,
+                        "invalid_image_id",
+                    )
+                elif status == 500 and message.startswith("can not find robot "):
+                    business = (
+                        E.CONFIG_REQUIRED,
+                        "Device key does not identify a frame",
+                        503,
+                        "device_key_invalid",
+                    )
+                elif status == 404 and message.startswith("can not find method "):
+                    business = (
+                        E.UNSUPPORTED_CAPABILITY,
+                        "Cloud method not found",
+                        422,
+                        "method_not_found",
+                    )
+            if business:
+                code, safe_message, local_status, kind = business
+                raise FlowError(
+                    code,
+                    safe_message,
+                    local_status,
+                    details={
+                        "http_status": status,
+                        "failure_kind": kind,
+                        "retryable": False,
+                        "write_outcome": "rejected" if write else "read_only",
+                    },
+                )
+            if status != 200:
+                limited = status == 429
+                rejected = 400 <= status < 500
+                uncertain = write and not rejected
+                details = {
+                    "http_status": status,
+                    "retryable": limited or (not write and status >= 500),
+                    "write_outcome": "unknown"
+                    if uncertain
+                    else ("rejected" if write else "read_only"),
+                }
+                if limited:
+                    delay = self._retry_after(response.headers.get("Retry-After"))
+                    details["retry_after_seconds"] = delay
+                    with self.lock:
+                        self.cooldown_until = max(
+                            self.cooldown_until, time.monotonic() + delay
+                        )
+                raise FlowError(
+                    E.RATE_LIMITED
+                    if limited
+                    else (E.WRITE_RESULT_UNKNOWN if uncertain else E.CLOUD_UNAVAILABLE),
+                    "Device cloud request failed",
+                    429 if limited else 502,
+                    details=details,
+                )
+            if binary:
+                if len(response.content) > 20 * 1024 * 1024:
+                    raise FlowError(E.INVALID_REQUEST, "Cloud image exceeds size limit")
+                return response.content
+            if (
+                not isinstance(value, dict)
+                or value.get("success") is False
+                or ("message" in value and "id" not in value)
             ):
-                raise ValueError()
+                raise FlowError(
+                    E.WRITE_RESULT_UNKNOWN if write else E.CLOUD_UNAVAILABLE,
+                    "Invalid device cloud response",
+                    502,
+                    details={
+                        "failure_kind": "invalid_response",
+                        "retryable": not write,
+                        "write_outcome": "unknown" if write else "read_only",
+                    },
+                )
             return value
-        except ValueError:
-            raise FlowError(
-                E.WRITE_RESULT_UNKNOWN if write else E.CLOUD_UNAVAILABLE,
-                "Invalid device cloud response",
-                502,
-            ) from None
+        finally:
+            response.close()
 
 
 class KokonnaClient:
@@ -118,7 +266,7 @@ class KokonnaClient:
     def upload(self, path, name):
         self.guard()
         value = guooscar_protocol.upload(self.transport, path.read_bytes(), name)
-        if not isinstance(value.get("id"), int) or value["id"] <= 0:
+        if type(value.get("id")) is not int or value["id"] <= 0:
             raise FlowError(E.WRITE_RESULT_UNKNOWN, "Upload ID not confirmed", 502)
         return value["id"]
 

pigway_frame_flow/errors.py

@@ -23,6 +23,7 @@ class ErrorCode(StrEnum):
 
 
 class FlowError(Exception):
-    def __init__(self, code: ErrorCode, message: str, status: int = 400):
+    def __init__(self, code: ErrorCode, message: str, status: int = 400, details=None):
         self.code, self.message, self.status = code, message, status
+        self.details = details or {}
         super().__init__(message)

pigway_frame_flow/extensions/webhooks.py

@@ -138,6 +138,7 @@ def install_webhooks(app, service):
         allowed = {
             "operation.completed",
             "operation.failed",
+            "operation.blocked",
             "display.changed",
             "device.online",
             "device.offline",

pigway_frame_flow/models.py

@@ -177,7 +177,17 @@ class OperationResult(BaseModel):
     deferred: bool = False
 
 
+class RecoveryState(BaseModel):
+    state: Literal["checking", "manual_review", "resolved"]
+    reason: str
+    first_checked_at: str
+    last_checked_at: str
+    checks: int
+    requires_manual_review: bool
+
+
 class Operation(BaseModel):
+    recovery: RecoveryState | None = None
     operation_id: str
     status: OperationStatus
     device_id: str
@@ -336,7 +346,7 @@ class SettingsView(BaseModel):
 
 class Health(BaseModel):
     status: str
-    version: str = "0.1.2"
+    version: str = "0.1.4"
 
 
 class Readiness(BaseModel):
@@ -366,7 +376,7 @@ class RestoreRequest(StrictModel):
 
 class Version(BaseModel):
     product: str = "Pigway Frame Flow"
-    version: str = "0.1.2"
+    version: str = "0.1.4"
     api_version: str = "v1"
     commit: str = "development"
     build_time: str = "unknown"

pigway_frame_flow/storage/database.py

@@ -31,6 +31,7 @@ class Database:
         CREATE TABLE IF NOT EXISTS image_sources (device_id TEXT, image_id INTEGER, asset_id TEXT, operation_id TEXT, payload TEXT NOT NULL, PRIMARY KEY(device_id,image_id));
         CREATE TABLE IF NOT EXISTS app_config (key TEXT PRIMARY KEY, payload TEXT NOT NULL);
         CREATE TABLE IF NOT EXISTS playlists (id TEXT PRIMARY KEY, payload TEXT, position INTEGER, next_run TEXT);
+        CREATE TABLE IF NOT EXISTS operation_recovery (operation_id TEXT PRIMARY KEY, state TEXT NOT NULL, reason TEXT NOT NULL, first_checked_at TEXT NOT NULL, last_checked_at TEXT NOT NULL, checks INTEGER NOT NULL);
         CREATE TABLE IF NOT EXISTS operations (
             id TEXT PRIMARY KEY, status TEXT NOT NULL, action TEXT NOT NULL,
             payload TEXT NOT NULL, device_id TEXT NOT NULL, client_id TEXT NOT NULL,

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"><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.2</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"></script><script src="/ui/features.js"></script><script src="/ui/extensions.js"></script><script src="/ui/app.js"></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.4</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"></script><script src="/ui/features.js"></script><script src="/ui/extensions.js"></script><script src="/ui/app.js"></script></body></html>

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pigway-frame-flow"
-version = "0.1.2"
+version = "0.1.4"
 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]

tests/schema-contract.json

@@ -613,6 +613,16 @@
   },
   "Operation": {
     "properties": {
+      "recovery": {
+        "anyOf": [
+          {
+            "$ref": "#/components/schemas/RecoveryState"
+          },
+          {
+            "type": "null"
+          }
+        ]
+      },
       "operation_id": {
         "type": "string",
         "title": "Operation Id"
@@ -1422,5 +1432,48 @@
       "type"
     ],
     "title": "AutoCollageContent"
+  },
+  "RecoveryState": {
+    "properties": {
+      "state": {
+        "type": "string",
+        "enum": [
+          "checking",
+          "manual_review",
+          "resolved"
+        ],
+        "title": "State"
+      },
+      "reason": {
+        "type": "string",
+        "title": "Reason"
+      },
+      "first_checked_at": {
+        "type": "string",
+        "title": "First Checked At"
+      },
+      "last_checked_at": {
+        "type": "string",
+        "title": "Last Checked At"
+      },
+      "checks": {
+        "type": "integer",
+        "title": "Checks"
+      },
+      "requires_manual_review": {
+        "type": "boolean",
+        "title": "Requires Manual Review"
+      }
+    },
+    "type": "object",
+    "required": [
+      "state",
+      "reason",
+      "first_checked_at",
+      "last_checked_at",
+      "checks",
+      "requires_manual_review"
+    ],
+    "title": "RecoveryState"
   }
 }

tests/test_client.py

@@ -1,3 +1,4 @@
+from types import SimpleNamespace
 from unittest.mock import Mock
 
 import pytest
@@ -7,11 +8,28 @@ from pigway_frame_flow.drivers.kokonna.client import KokonnaClient
 from pigway_frame_flow.errors import FlowError
 
 
+@pytest.fixture(autouse=True)
+def transport_clock(monkeypatch):
+    import pigway_frame_flow.drivers.kokonna.client as module
+
+    clock = SimpleNamespace(now=0.0, waits=[])
+
+    def sleep(seconds):
+        clock.waits.append(seconds)
+        clock.now += seconds
+
+    monkeypatch.setattr(
+        module, "time", SimpleNamespace(monotonic=lambda: clock.now, sleep=sleep)
+    )
+    return clock
+
+
 def session(body=None, status=200):
     s = Mock()
     s.request.return_value.status_code = status
     s.request.return_value.json.return_value = body or {}
     s.request.return_value.content = b"image"
+    s.request.return_value.headers = {}
     return s
 
 
@@ -88,7 +106,7 @@ def test_timeout_redaction_no_retry(write):
         c.display(1) if write else c.get_device()
     assert "secret" not in str(e.value)
     assert e.value.code == ("WRITE_RESULT_UNKNOWN" if write else "CLOUD_UNAVAILABLE")
-    assert s.request.call_count == 1
+    assert s.request.call_count == (1 if write else 3)
 
 
 @pytest.mark.parametrize("status", [401, 429, 500, 530])
@@ -109,12 +127,187 @@ def test_invalid_upload_id(tmp_path):
     assert e.value.code == "WRITE_RESULT_UNKNOWN"
 
 
-def test_rate_budget():
+def test_rate_budget(transport_clock):
     s = session({"nickname": "x"})
     c = KokonnaClient("secret", session=s)
     for _ in range(18):
         c.get_device()
-    with pytest.raises(FlowError) as e:
-        c.get_device()
-    assert e.value.code == "RATE_LIMITED"
-    assert s.request.call_count == 18
+    c.get_device()
+    assert s.request.call_count == 19
+    assert transport_clock.waits == [60]
+
+
+@pytest.mark.parametrize(
+    "exception,category",
+    [
+        (requests.ConnectTimeout, "connect_timeout"),
+        (requests.ReadTimeout, "read_timeout"),
+        (requests.exceptions.SSLError, "tls_error"),
+        (requests.ConnectionError, "connection_error"),
+        (requests.Timeout, "timeout"),
+        (requests.RequestException, "request_error"),
+    ],
+)
+def test_transport_failure_records_safe_category_without_replay(exception, category):
+    s = session()
+    s.request.side_effect = exception("secret https://host/secret")
+    c = KokonnaClient("secret", True, s)
+    with pytest.raises(FlowError) as error:
+        c.display(1)
+    assert error.value.code == (
+        "CLOUD_UNAVAILABLE" if category == "connect_timeout" else "WRITE_RESULT_UNKNOWN"
+    )
+    assert error.value.details["transport_error"] == category
+    assert error.value.details["attempts"] == (
+        3 if category == "connect_timeout" else 1
+    )
+    assert "secret" not in str(error.value.details)
+    assert s.request.call_count == (3 if category == "connect_timeout" else 1)
+
+
+@pytest.mark.parametrize("write", [False, True])
+def test_connect_timeout_retries_then_succeeds(write, transport_clock):
+    s = session({"success": True})
+    response = s.request.return_value
+    s.request.side_effect = [requests.ConnectTimeout("private"), response]
+    c = KokonnaClient("private", True, s)
+    c.display(1) if write else c.get_device()
+    assert s.request.call_count == 2
+    assert transport_clock.waits == [1]
+
+
+@pytest.mark.parametrize(
+    "message,code,kind",
+    [
+        ("image not found", "NOT_FOUND", "image_not_found"),
+        ("imageId is empty", "INVALID_REQUEST", "invalid_image_id"),
+        ("can not find robot private", "CONFIG_REQUIRED", "device_key_invalid"),
+    ],
+)
+def test_documented_500_rejection_does_not_lock_or_retry(message, code, kind):
+    s = session({"message": message}, 500)
+    c = KokonnaClient("private", True, s)
+    with pytest.raises(FlowError) as error:
+        c.display(1)
+    assert error.value.code == code
+    assert error.value.details["write_outcome"] == "rejected"
+    assert error.value.details["failure_kind"] == kind
+    assert "private" not in str(error.value.details)
+    assert s.request.call_count == 1
+
+
+@pytest.mark.parametrize("status", [500, 502, 503, 530])
+def test_generic_5xx_read_retries_but_write_never_replays(status):
+    for write in [False, True]:
+        s = session({"message": "unknown failure private"}, status)
+        c = KokonnaClient("private", True, s)
+        with pytest.raises(FlowError) as error:
+            c.display(1) if write else c.get_device()
+        assert s.request.call_count == (1 if write else 3)
+        assert error.value.code == (
+            "WRITE_RESULT_UNKNOWN" if write else "CLOUD_UNAVAILABLE"
+        )
+        assert "private" not in str(error.value.details)
+
+
+def test_rate_limit_honors_retry_after_and_closes_response(transport_clock):
+    s = session({"message": "rate limited"}, 429)
+    first = s.request.return_value
+    first.headers = {"Retry-After": "7"}
+    second = session({"success": True}).request.return_value
+    s.request.side_effect = [first, second]
+    KokonnaClient("private", True, s).display(1)
+    assert s.request.call_count == 2
+    assert transport_clock.waits == [7]
+    first.close.assert_called_once()
+    second.close.assert_called_once()
+
+
+def test_long_retry_after_preserves_cooldown_without_early_request(transport_clock):
+    s = session({"message": "rate limited"}, 429)
+    s.request.return_value.headers = {"Retry-After": "120"}
+    c = KokonnaClient("private", True, s)
+    for _ in range(2):
+        with pytest.raises(FlowError) as error:
+            c.display(1)
+        assert error.value.code == "RATE_LIMITED"
+    assert s.request.call_count == 1
+    assert transport_clock.waits == []
+
+
+@pytest.mark.parametrize("header", [None, "bad", "nan", "inf"])
+def test_invalid_retry_after_defaults_to_one_minute(header):
+    from pigway_frame_flow.drivers.kokonna.client import SafeTransport
+
+    assert SafeTransport._retry_after(header) == 60
+
+
+def test_retry_after_accepts_http_date():
+    from datetime import UTC, datetime, timedelta
+    from email.utils import format_datetime
+
+    from pigway_frame_flow.drivers.kokonna.client import SafeTransport
+
+    future = format_datetime(datetime.now(UTC) + timedelta(seconds=30), usegmt=True)
+    assert 28 <= SafeTransport._retry_after(future) <= 30
+
+
+def test_invalid_write_response_is_never_retried():
+    s = session({"success": False, "message": "private"})
+    with pytest.raises(FlowError) as error:
+        KokonnaClient("private", True, s).display(1)
+    assert error.value.code == "WRITE_RESULT_UNKNOWN"
+    assert error.value.details["failure_kind"] == "invalid_response"
+    assert s.request.call_count == 1
+
+
+@pytest.mark.parametrize(
+    "failure,expected",
+    [
+        ("connect_timeout", "CLOUD_UNAVAILABLE"),
+        ("business_rejection", "CONFIG_REQUIRED"),
+        ("rate_limit", "RATE_LIMITED"),
+    ],
+)
+def test_proven_upload_failures_end_failed_without_device_lock(
+    service, monkeypatch, failure, expected
+):
+    from test_display import submit
+
+    s = session({"message": "can not find robot private"}, 500)
+    if failure == "rate_limit":
+        s.request.return_value.status_code = 429
+    if failure == "connect_timeout":
+        s.request.side_effect = requests.ConnectTimeout("private")
+    c = KokonnaClient("private", True, s)
+    monkeypatch.setattr(service.registry.get("living-room"), "upload", c.upload)
+    operation = submit(service)
+    service.process_one()
+    op = service.operation(operation)
+    assert op["status"] == "FAILED"
+    assert op["error"]["code"] == expected
+    assert not service.blocked("living-room")
+
+
+@pytest.mark.parametrize("value", [True, False, 0, -1, "12", None])
+def test_invalid_upload_id_never_becomes_confirmed(tmp_path, value):
+    path = tmp_path / "image.jpg"
+    path.write_bytes(b"jpeg")
+    s = session({"id": value})
+    with pytest.raises(FlowError) as error:
+        KokonnaClient("private", True, s).upload(path, "image.jpg")
+    assert error.value.code == "WRITE_RESULT_UNKNOWN"
+    assert s.request.call_count == 1
+
+
+def test_ambiguous_failure_after_safe_rejection_stops_retrying(transport_clock):
+    s = session({"message": "limited"}, 429)
+    first = s.request.return_value
+    first.headers = {"Retry-After": "1"}
+    s.request.side_effect = [first, requests.ReadTimeout("private")]
+    with pytest.raises(FlowError) as error:
+        KokonnaClient("private", True, s).display(1)
+    assert error.value.code == "WRITE_RESULT_UNKNOWN"
+    assert error.value.details["attempts"] == 2
+    assert s.request.call_count == 2
+    assert transport_clock.waits == [1]

tests/test_recovery.py

@@ -0,0 +1,109 @@
+from datetime import UTC, datetime, timedelta
+
+from test_display import submit
+
+from pigway_frame_flow.display.service import DisplayService
+from pigway_frame_flow.errors import ErrorCode, FlowError
+
+
+def unknown_upload(service, monkeypatch):
+    def fail(*args):
+        raise FlowError(ErrorCode.WRITE_RESULT_UNKNOWN, "Response lost", 502)
+
+    monkeypatch.setattr(service.registry.get("living-room"), "upload", fail)
+    operation = submit(service)
+    service.process_one()
+    return operation
+
+
+def test_missing_upload_is_visible_without_unsafe_unlock(client, service, monkeypatch):
+    operation = unknown_upload(service, monkeypatch)
+    service.sync()
+    recovery = client.get("/api/v1/operations/" + operation).json()["recovery"]
+    assert recovery["reason"] == "gallery_match_missing"
+    assert recovery["state"] == "checking"
+    assert not recovery["requires_manual_review"]
+    assert service.blocked("living-room")
+    assert service.registry.get("living-room").client.uploads == 0
+
+
+def test_persistent_block_escalates_once_survives_restart_and_preserves_evidence(
+    service, monkeypatch
+):
+    operation = unknown_upload(service, monkeypatch)
+    original = service.operation(operation)
+    service.db.execute(
+        "UPDATE operations SET created_at=? WHERE id=?",
+        (
+            (datetime.now(UTC) - timedelta(hours=1)).isoformat(),
+            operation,
+        ),
+    )
+    service.sync()
+    first = service.operation(operation)["recovery"]
+    assert first["state"] == "manual_review"
+    assert first["reason"] == "gallery_match_missing"
+    service.unknown_checked.clear()
+    service.sync()
+    restarted = DisplayService(
+        service.db, service.settings, service.registry, service.assets, service.renderer
+    )
+    restarted.sync()
+    assert (
+        len(service.db.all("SELECT * FROM events WHERE event_type='operation.blocked'"))
+        == 1
+    )
+    assert (
+        service.db.one(
+            "SELECT COUNT(*) AS n FROM history WHERE result='MANUAL_REVIEW_REQUIRED'"
+        )["n"]
+        == 1
+    )
+    assert service.operation(operation)["recovery"]["checks"] == 3
+    assert service.operation(operation)["result"] == original["result"]
+    service.reconcile(operation, acknowledge=True)
+    assert service.operation(operation)["recovery"]["state"] == "resolved"
+    assert not service.operation(operation)["recovery"]["requires_manual_review"]
+    assert not service.blocked("living-room")
+
+
+def test_missing_legacy_evidence_escalates_without_cloud_write(service):
+    operation = submit(service)
+    service._start_write(operation)
+    service.update(operation, "UNKNOWN")
+    service.sync()
+    assert (
+        service.operation(operation)["recovery"]["reason"] == "missing_upload_evidence"
+    )
+    assert service.blocked("living-room")
+
+
+def test_delete_unknown_requires_manual_review_without_replay(service):
+    operation = service.submit(
+        "delete", {"image_id": 1}, "living-room", "admin", "test"
+    )
+    service._start_write(operation)
+    service.update(operation, "UNKNOWN")
+    service.sync()
+    assert (
+        service.operation(operation)["recovery"]["reason"]
+        == "manual_reconciliation_required"
+    )
+    assert service.blocked("living-room")
+    assert service.registry.get("living-room").images()[0].image_id == 1
+
+
+def test_malformed_reconciliation_does_not_abort_other_background_work(
+    service, monkeypatch
+):
+    operation = unknown_upload(service, monkeypatch)
+
+    def invalid_metadata(*args):
+        raise ValueError("vendor metadata with private data")
+
+    monkeypatch.setattr(service.upload_reconciliation, "check", invalid_metadata)
+    service.sync()
+    op = service.operation(operation)
+    assert op["recovery"]["reason"] == "reconciliation_error"
+    assert "private" not in str(op["recovery"])
+    assert service.blocked("living-room")

tests/test_upload_reconciliation.py

@@ -111,3 +111,43 @@ def test_recovered_upload_preserves_temporary_restore(service, monkeypatch):
     assert service.operation(operation)["status"] == "COMPLETED"
     assert service.top("living-room")["id"] == operation
     assert service.state("living-room")["managed_image_id"] == 2
+
+
+def test_manual_close_preserves_unknown_upload_evidence(service, monkeypatch):
+    operation, driver = lose_response(service, monkeypatch, change_display=False)
+    before = service.operation(operation)
+    assert before["result"].get("image_id") is None
+    result = service.reconcile(operation, acknowledge=True)
+    after = service.operation(operation)
+    assert result["resolved"]
+    assert after["status"] == "FAILED"
+    assert after["result"] == before["result"]
+    assert after["error"]["code"] == "DISPLAY_NOT_CONFIRMED"
+    assert after["error"]["details"]["previous_error"] == before["error"]
+    assert after["error"]["details"]["observed_image_id"] == 1
+    assert not service.blocked("living-room")
+    assert driver.client.uploads == 1
+    assert (
+        service.db.one("SELECT * FROM history ORDER BY id DESC LIMIT 1")["result"]
+        == "ACKNOWLEDGED"
+    )
+
+
+def test_upload_transport_diagnostics_persist_on_operation(service, monkeypatch):
+    driver = service.registry.get("living-room")
+
+    def fail(*args):
+        raise FlowError(
+            ErrorCode.WRITE_RESULT_UNKNOWN,
+            "Cloud write outcome is unknown",
+            502,
+            details={"transport_error": "read_timeout"},
+        )
+
+    monkeypatch.setattr(driver, "upload", fail)
+    operation = submit(service)
+    service.process_one()
+    op = service.operation(operation)
+    assert op["status"] == "UNKNOWN"
+    assert op["error"]["details"] == {"transport_error": "read_timeout"}
+    assert op["result"]["upload_intent"]