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

Preserve uncertain-write evidence and safe transport diagnostics for v0.1.3 [skip ci]
Fix webhook starvation and isolate scheduled job failures for v0.1.2 [skip ci]
2026-10-16 12:25:16 +08:00 · 2026-10-16 10:30:08 +08:00 · 2026-10-16 09:48:51 +08:00 · 2026-10-14 13:10:33 +08:00 · 2026-10-12 21:12:44 +08:00 · 2026-10-12 20:51:08 +08:00
29 changed files with 1697 additions and 132 deletions
@@ -90,4 +90,4 @@ Tests mock cloud/device calls. CI also verifies an isolated Docker build, readin

 Cloud synchronization does not prove the physical pixels refreshed. KoKonna cloud uploads activate immediately, so upload-only uses local staging. Native slideshow membership editing is unavailable; avoid running native and Frame Flow playback simultaneously. Smart cropping is a saliency heuristic, not guaranteed face detection. The service uses one worker and is not designed for multiple replicas. Browser OAuth, AI daily-art scheduling, additional calendar recurrence features and more device drivers remain roadmap items.

-Related projects: [PIGWay Device Agent](https://tea.pigway.com/way/pigway-device-agent) · [PIGWay Device Console](https://tea.pigway.com/way/pigway-device-console) · [PIGWay Yahboom Cooling HAT](https://tea.pigway.com/way/pigway-yahboom-cooling-hat)
+Related projects: [PIGWay Raspberry Pi Agent](https://tea.pigway.com/way/pigway-raspberry-pi-agent) · [PIGWay Device Console](https://tea.pigway.com/way/pigway-device-console) · [PIGWay Yahboom Cooling HAT](https://tea.pigway.com/way/pigway-yahboom-cooling-hat)
@@ -90,4 +90,4 @@ KoKonna 是支持的设备集成。本项目与 KoKonna 无隶属关系，也不

 云端同步不等于已确认物理屏幕刷新。KoKonna 云端上传会立即切换，因此“仅上传”使用本地暂存。官方接口不支持编辑原生轮播成员，应避免原生轮播与 Frame Flow 播放同时运行。smart 裁切采用显著性启发式，不保证识别人脸。当前服务使用单 Worker，不面向多副本部署。浏览器 OAuth、AI 每日艺术调度、更多日历重复规则及设备驱动仍属于后续计划。

-相关项目：[PIGWay Device Agent](https://tea.pigway.com/way/pigway-device-agent) · [PIGWay Device Console](https://tea.pigway.com/way/pigway-device-console) · [PIGWay Yahboom Cooling HAT](https://tea.pigway.com/way/pigway-yahboom-cooling-hat)
+相关项目：[PIGWay Raspberry Pi Agent](https://tea.pigway.com/way/pigway-raspberry-pi-agent) · [PIGWay Device Console](https://tea.pigway.com/way/pigway-device-console) · [PIGWay Yahboom Cooling HAT](https://tea.pigway.com/way/pigway-yahboom-cooling-hat)
@@ -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. 有目标一致且同步完成的证据才记为成功。处理后恢复原计划，错过的任务不补发。
+
+仍可能因厂商无法给出明确写入结果而需要人工处理；本策略解决错误误判、可安全重试的暂时失败和长期阻塞缺乏解释的问题，不承诺网络故障时所有设备任务都能自动成功。
@@ -25,3 +25,23 @@
 The API cannot set coverId. Restoration guarantees a requested image switch when
 confirmed, not arbitrary cover metadata restoration. Physical screen refresh has
 no documented pixel telemetry; visual owner observation supplements cloud synced.
+
+### Upload response reconciliation
+
+Before an upload, the service persists its asset ID, exact SHA256/size, filename,
+existing gallery IDs and timestamp. When the response is lost, read-only queries
+look for one new image in the upload time window; only an exact downloaded-byte
+match identifies it. Ambiguous, missing or altered content remains locked for
+manual review. No upload or switch is replayed by reconciliation.
+
+A proven upload completes the display only when the device reports that exact
+image and `synced=true`. If a normal display remains on its previous synced
+image beyond the sync timeout plus five minutes, two observations at least one
+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)。自动重试只允许只读请求、连接建立超时或明确限流拒绝；读取超时等未知写入不会重发。
@@ -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.
@@ -107,7 +107,7 @@ been performed during implementation; those are first-run acceptance steps.
   an immutable rollback tag so normal dangling-image cleanup cannot remove it.
 4. Refuse a stale checked-out `main` revision that no longer equals remote main.
   Build an immutable `release-<commit>-<owner>` tag locally, without a registry
-   and without changing `latest` or CI image tags.
+   then point `latest` at the candidate; CI image tags are never changed.
 5. Persist a `prepared` journal with old/new runtime configs. Create an additive
   SQLite online backup under `/data/deploy-backups/<owner>.db`. No data/config
   directory or file is removed, replaced or recursively chowned.
@@ -266,7 +266,7 @@ as an Actions artifact. First bootstrap has no prior healthy version to roll bac
   `192.0.2.10:18787` binding (not loopback).
 4. Open `http://192.0.2.10:18787/api/v1/version`; its `commit` must equal the full
   workflow checkout commit. Check `/ready` returns HTTP 200. The immutable image tag
-   is `pigway-frame-flow:release-<commit>-<transaction>`; compare its image ID in
+   is `pigway-frame-flow:latest`; compare its image ID in
   Container Manager with the running container.
 5. Keep `PROD_DEPLOY_ENABLED=true`. Subsequent main pushes Verify then deploy using
   this same runtime and journal mechanism. The first update records bootstrap as
@@ -363,3 +363,18 @@ the audit client's normal HTTP freshness rules. It is not a permanent offline
 snapshot or a cross-job cache. Raw network exceptions are omitted because proxy
 URLs may contain credentials. Logs identify the advisory host and whether the
 complete audit succeeded.
+
+### Production image retention
+
+Production uses `pigway-frame-flow:latest`. Builds first create a unique release
+image, then tag it as latest before activation. The old image is pinned during
+an update so a failed deployment can restore it. Only after the new version
+passes verification are old release/rollback images and duplicate tags removed.
+After successful deployment, only the current production image remains; there
+is no retained historical image for manual rollback. Rebuild a historical
+revision before manually reverting a successful deployment.
+
+Images referenced by other containers (including stopped containers), images
+with unrelated tags, and CI resources remain protected. Cleanup never prunes
+volumes, data, or shared build cache. Cleanup failure is a warning and does not
+roll back a verified deployment. Failed deployments do not run cleanup.
@@ -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",
@@ -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.0",
+        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.0",
+            version="0.1.4",
            mock=settings.mock,
            write_enabled=settings.write_enabled,
            api_key_configured=bool(settings.api_key)
@@ -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
@@ -6,7 +6,9 @@ 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
 from pigway_frame_flow.errors import ErrorCode as E
 from pigway_frame_flow.errors import FlowError
@@ -49,7 +51,10 @@ 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] = {}
        # A killed write cannot be replayed safely; pre-write work can resume.
        db.execute(
            "UPDATE operations SET status='UNKNOWN', updated_at=? WHERE write_started=1 AND status IN ('UPLOADING','RENDERING')",
@@ -162,14 +167,16 @@ 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):
        if result:
            row = self.db.one("SELECT result FROM operations WHERE id=?", (id,))
            previous = json.loads(row["result"]) if row and row["result"] else {}
-            if "photo_attempts" in previous:
-                result = {**result, "photo_attempts": previous["photo_attempts"]}
+            for key in ("photo_attempts", "upload_intent"):
+                if key in previous:
+                    result = {**result, key: previous[key]}
        self.db.execute(
            "UPDATE operations SET status=?,updated_at=?,result=COALESCE(?,result),error=? WHERE id=?",
            (
@@ -180,6 +187,8 @@ class DisplayService:
                id,
            ),
        )
+        if status in ("COMPLETED", "FAILED"):
+            self.recovery.resolve(id)

    def blocked(self, device_id):
        return (
@@ -438,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
@@ -626,6 +635,8 @@ class DisplayService:
            raise FlowError(
                E.CONFLICT, "Temporary override requires an existing image", 409
            )
+        if asset:
+            self.upload_reconciliation.prepare(id, driver, asset, previous)
        self._start_write(id)
        if asset:
            self.update(id, "UPLOADING")
@@ -689,6 +700,63 @@ class DisplayService:
        )

    def sync(self):
+        with self.lock:
+            self._sync()
+
+    def _sync(self):
+        # Late acknowledgements only resolve a known display target. Never replay
+        # writes or infer success from gallery presence alone.
+        for op in self.db.all(
+            "SELECT * FROM operations WHERE status='UNKNOWN' AND action!='render-ai' AND write_started=1"
+        ):
+            expected = json.loads(op["result"] or "{}").get("image_id")
+            intent = json.loads(op["result"] or "{}").get("upload_intent")
+            checked = time.monotonic()
+            if checked - self.unknown_checked.get(op["id"], float("-inf")) < 60:
+                continue
+            self.unknown_checked[op["id"]] = checked
+            if op["action"] != "display" or (not expected and not intent):
+                self.recovery.observe(
+                    op,
+                    "missing_upload_evidence"
+                    if op["action"] == "display"
+                    else "manual_reconciliation_required",
+                )
+                continue
+            try:
+                driver = self.registry.get(op["device_id"])
+                device = driver.device(refresh=True)
+                if intent:
+                    expected = self.upload_reconciliation.check(op, driver, device)
+                    if (
+                        self.db.one(
+                            "SELECT status FROM operations WHERE id=?", (op["id"],)
+                        )["status"]
+                        == "FAILED"
+                    ):
+                        continue
+                if (
+                    expected
+                    and device.current_content.id == expected
+                    and device.current_content.synced
+                ):
+                    self.update(op["id"], "COMPLETED")
+                    self.cloud_library.confirm(op["id"])
+                    self.db.history(op, "COMPLETED")
+                    self.db.event("operation.completed", {"operation_id": op["id"]})
+                    self.unknown_checked.pop(op["id"], None)
+                elif expected:
+                    self.recovery.observe(
+                        op,
+                        "waiting_device_sync"
+                        if device.current_content.id == expected
+                        else "target_mismatch",
+                    )
+            except FlowError:
+                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()
@@ -821,9 +889,25 @@ 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":
+                    self.cloud_library.confirm(id)
+                    self.db.event("operation.completed", {"operation_id": id})
                self.db.history(
                    dict(
                        id=id,
@@ -0,0 +1,164 @@
+"""Resolve missing upload responses using read-only, exact-content evidence."""
+
+import hashlib
+import json
+from datetime import UTC, datetime, timedelta
+
+from pigway_frame_flow.errors import ErrorCode
+from pigway_frame_flow.storage.database import encode, now_iso
+
+
+class UploadReconciliation:
+    def __init__(self, service):
+        self.service = service
+
+    def prepare(self, operation_id, driver, asset, previous):
+        path = self.service.assets.path(asset["asset_id"])
+        self.service.update(
+            operation_id,
+            "RENDERING",
+            {
+                "rendered_asset_id": asset["asset_id"],
+                "upload_intent": {
+                    "name": asset["name"],
+                    "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
+                    "size": path.stat().st_size,
+                    "before_ids": [image.image_id for image in driver.images()],
+                    "previous_image_id": previous,
+                    "started_at": now_iso(),
+                },
+            },
+        )
+
+    def check(self, op, driver, device):
+        service = self.service
+        result = json.loads(op["result"] or "{}")
+        intent = result.get("upload_intent")
+        if not intent:
+            return None
+        expected = result.get("image_id")
+        if not expected:
+            start = datetime.fromisoformat(intent["started_at"])
+            candidates = []
+            for image in driver.images():
+                if (
+                    image.image_id in intent["before_ids"]
+                    or image.name != intent["name"]
+                ):
+                    continue
+                if image.size_bytes is not None and image.size_bytes != intent["size"]:
+                    continue
+                if not image.created_at:
+                    continue
+                try:
+                    created = datetime.fromisoformat(image.created_at)
+                    age = (created - start).total_seconds()
+                except (ValueError, TypeError):
+                    continue
+                if 0 <= age <= service.settings.sync_timeout + 300:
+                    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)
+            if (
+                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
+            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"],),
+            )
+            service.db.execute(
+                "INSERT OR REPLACE INTO image_sources VALUES(?,?,?,?,?)",
+                (
+                    op["device_id"],
+                    expected,
+                    result["rendered_asset_id"],
+                    op["id"],
+                    source["payload"] if source else encode({"type": "image"}),
+                ),
+            )
+            if not service.db.one(
+                "SELECT image_id FROM managed_cloud_uploads WHERE operation_id=?",
+                (op["id"],),
+            ):
+                service.cloud_library.register(
+                    op["device_id"], expected, result["rendered_asset_id"], op["id"]
+                )
+        if device.current_content.id == expected and device.current_content.synced:
+            options = json.loads(op["payload"]).get("options", {})
+            service.db.execute(
+                "UPDATE device_state SET managed_image_id=? WHERE device_id=?",
+                (expected, op["device_id"]),
+            )
+            if options.get("restore_previous"):
+                expires = datetime.fromisoformat(intent["started_at"]) + timedelta(
+                    seconds=options["duration"]
+                )
+                service.db.execute(
+                    "INSERT OR IGNORE INTO overrides VALUES(?,?,?,?,?,?)",
+                    (
+                        op["id"],
+                        op["device_id"],
+                        expected,
+                        options["priority"],
+                        expires.isoformat(),
+                        intent["started_at"],
+                    ),
+                )
+            return expected
+        elapsed = (
+            datetime.now(UTC) - datetime.fromisoformat(intent["started_at"])
+        ).total_seconds()
+        if (
+            result.get("upload_confirmed")
+            and not json.loads(op["payload"]).get("options", {}).get("restore_previous")
+            and elapsed > service.settings.sync_timeout + 300
+            and device.current_content.id == intent["previous_image_id"]
+            and device.current_content.synced
+        ):
+            # Upload is proven; display is not. Two separated stable observations
+            # close the failed display, without replaying either vendor write.
+            observed = result.get("unchanged_observed_at")
+            if not observed:
+                result["unchanged_observed_at"] = now_iso()
+                service.update(
+                    op["id"], "UNKNOWN", result, json.loads(op["error"] or "{}")
+                )
+            elif (
+                datetime.now(UTC) - datetime.fromisoformat(observed)
+            ).total_seconds() >= 60:
+                service.update(
+                    op["id"],
+                    "FAILED",
+                    result,
+                    {
+                        "code": ErrorCode.DISPLAY_NOT_CONFIRMED,
+                        "message": "Upload confirmed; frame remained on previous image",
+                        "request_id": op["request_id"],
+                        "details": {},
+                    },
+                )
+                service.db.history(op, "DISPLAY_NOT_CONFIRMED")
+                service.db.event(
+                    "operation.failed",
+                    {"operation_id": op["id"], "code": ErrorCode.DISPLAY_NOT_CONFIRMED},
+                )
+        elif result.get("unchanged_observed_at"):
+            result.pop("unchanged_observed_at")
+            service.update(op["id"], "UNKNOWN", result, json.loads(op["error"] or "{}"))
+        return expected
@@ -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"]

@@ -11,6 +11,7 @@ class ErrorCode(StrEnum):
    UNSUPPORTED_CAPABILITY = "UNSUPPORTED_CAPABILITY"
    RENDER_FAILED = "RENDER_FAILED"
    UPLOAD_FAILED = "UPLOAD_FAILED"
+    DISPLAY_NOT_CONFIRMED = "DISPLAY_NOT_CONFIRMED"
    WRITE_RESULT_UNKNOWN = "WRITE_RESULT_UNKNOWN"
    RATE_LIMITED = "RATE_LIMITED"
    INTERNAL_ERROR = "INTERNAL_ERROR"
@@ -22,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)
@@ -39,12 +39,13 @@ class Webhooks:
        db = self.service.db
        for sub in db.all("SELECT * FROM webhook_subscriptions WHERE enabled=1"):
            types = json.loads(sub["event_types"])
+            if not types:
+                continue
+            placeholders = ",".join("?" for _ in types)
            for event in db.all(
-                "SELECT * FROM events e WHERE created_at>=? AND NOT EXISTS (SELECT 1 FROM webhook_deliveries d WHERE d.event_id=e.id AND d.subscription_id=?) ORDER BY created_at LIMIT 500",
-                (sub["created_at"], sub["id"]),
+                f"SELECT * FROM events e WHERE created_at>=? AND event_type IN ({placeholders}) AND NOT EXISTS (SELECT 1 FROM webhook_deliveries d WHERE d.event_id=e.id AND d.subscription_id=?) ORDER BY created_at LIMIT 500",
+                (sub["created_at"], *types, sub["id"]),
            ):
-                if event["event_type"] not in types:
-                    continue
                db.execute(
                    "INSERT OR IGNORE INTO webhook_deliveries VALUES(?,?,'QUEUED',0,?,NULL)",
                    (sub["id"], event["id"], now_iso()),
@@ -137,6 +138,7 @@ def install_webhooks(app, service):
        allowed = {
            "operation.completed",
            "operation.failed",
+            "operation.blocked",
            "display.changed",
            "device.online",
            "device.offline",
@@ -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.0"
+    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.0"
+    version: str = "0.1.4"
    api_version: str = "v1"
    commit: str = "development"
    build_time: str = "unknown"
@@ -147,61 +147,72 @@ class Scheduler:
                "SELECT * FROM jobs WHERE enabled=1 AND next_run<=?",
                (current.isoformat(),),
            ):
-                job = JobInput.model_validate_json(row["payload"])
-                slot = row["next_run"]
-                if self.db.one(
-                    "SELECT * FROM job_runs WHERE job_id=? AND scheduled_at=?",
-                    (row["id"], slot),
-                ):
-                    continue
-                # Skip overdue tasks (>5 minutes): avoid an unexpected artwork
-                # overwrite after downtime. One-time tasks are retired too.
-                late = (current - datetime.fromisoformat(slot)).total_seconds() > 300
-                operation_id = None
-                if not late:
-                    action = "restore" if job.source.type == "restore" else "display"
-                    payload = (
-                        {}
-                        if isinstance(job.source, RestoreContent)
-                        else DisplayRequest(
-                            device_id=job.device_id,
-                            content=job.source,
-                            options=Options(
-                                duration=job.display_duration,
-                                restore_previous=job.restore_previous,
-                                priority=job.priority,
-                                cleanup_unstarred=job.cleanup_unstarred,
+                try:
+                    job = JobInput.model_validate_json(row["payload"])
+                    slot = row["next_run"]
+                    if self.db.one(
+                        "SELECT * FROM job_runs WHERE job_id=? AND scheduled_at=?",
+                        (row["id"], slot),
+                    ):
+                        continue
+                    # Skip overdue tasks (>5 minutes): avoid an unexpected artwork
+                    # overwrite after downtime. One-time tasks are retired too.
+                    late = (
+                        current - datetime.fromisoformat(slot)
+                    ).total_seconds() > 300
+                    operation_id = None
+                    if not late:
+                        action = (
+                            "restore" if job.source.type == "restore" else "display"
+                        )
+                        payload = (
+                            {}
+                            if isinstance(job.source, RestoreContent)
+                            else DisplayRequest(
+                                device_id=job.device_id,
+                                content=job.source,
+                                options=Options(
+                                    duration=job.display_duration,
+                                    restore_previous=job.restore_previous,
+                                    priority=job.priority,
+                                    cleanup_unstarred=job.cleanup_unstarred,
+                                ),
+                            ).model_dump(mode="json")
+                        )
+                        operation_id = self.service.submit(
+                            action,
+                            payload,
+                            job.device_id,
+                            "scheduler",
+                            "schedule-" + uuid4().hex,
+                            "job-" + row["id"] + "-" + slot,
+                        )
+                    following = next_run(job.schedule, current)
+                    enabled = bool(following)
+                    updated = job.model_copy(update={"enabled": enabled}).model_dump(
+                        mode="json"
+                    )
+                    with self.db.transaction() as tx:
+                        tx.execute(
+                            "INSERT OR IGNORE INTO job_runs VALUES(?,?,?)",
+                            (row["id"], slot, operation_id),
+                        )
+                        tx.execute(
+                            "UPDATE jobs SET payload=?,enabled=?,next_run=?,last_run=?,last_result=? WHERE id=?",
+                            (
+                                encode(updated),
+                                int(enabled),
+                                following,
+                                now_iso(),
+                                "SKIPPED_OVERDUE" if late else operation_id,
+                                row["id"],
                            ),
-                        ).model_dump(mode="json")
-                    )
-                    operation_id = self.service.submit(
-                        action,
-                        payload,
-                        job.device_id,
-                        "scheduler",
-                        "schedule-" + uuid4().hex,
-                        "job-" + row["id"] + "-" + slot,
-                    )
-                following = next_run(job.schedule, current)
-                enabled = bool(following)
-                updated = job.model_copy(update={"enabled": enabled}).model_dump(
-                    mode="json"
-                )
-                with self.db.transaction() as tx:
-                    tx.execute(
-                        "INSERT OR IGNORE INTO job_runs VALUES(?,?,?)",
-                        (row["id"], slot, operation_id),
-                    )
-                    tx.execute(
-                        "UPDATE jobs SET payload=?,enabled=?,next_run=?,last_run=?,last_result=? WHERE id=?",
-                        (
-                            encode(updated),
-                            int(enabled),
-                            following,
-                            now_iso(),
-                            "SKIPPED_OVERDUE" if late else operation_id,
-                            row["id"],
-                        ),
+                        )
+                except Exception:
+                    # Keep this slot pending for retry or the existing overdue policy.
+                    # A single device must not prevent other jobs or queue consumption.
+                    logging.getLogger("frame-flow").error(
+                        "Scheduled job failed; details suppressed"
                    )
            for row in self.db.all(
                "SELECT * FROM playlists WHERE next_run<=?", (current.isoformat(),)
@@ -269,6 +280,9 @@ class Scheduler:
    def run(self):
        while not self.stop_event.is_set():
            try:
+                # Resolve late sync before submitting due jobs: a write lock
+                # otherwise makes tick fail before reconciliation can run.
+                self.service.sync()
                self.tick()
                self.service.expire()
                self.service.cloud_library.sweep()
@@ -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,
@@ -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.0</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>
@@ -1,6 +1,6 @@
 [project]
 name = "pigway-frame-flow"
-version = "0.1.0"
+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]
@@ -37,6 +37,7 @@ class Deploy:
        self.lock_id = None
        self.state = None
        self.old_image = None
+        self.old_image_tag = ""
        self.secret_values = [
            self.env[key]
            for key in (
@@ -196,6 +197,7 @@ class Deploy:
        ):
            raise ValueError("Production user/restart policy differs from repository")
        self.old_image = old["Image"]
+        self.old_image_tag = old["Config"].get("Image", "")
        volume = self.run(
            "volume", "inspect", self.env["PROD_STATE_VOLUME"], check=False
        )
@@ -335,6 +337,12 @@ class Deploy:
    def rollback(self, previous):
        self.activate(previous)
        self.verify(previous)
+        self.run(
+            "image",
+            "tag",
+            previous["services"]["pigway-frame-flow"]["image"],
+            self.env["PROD_IMAGE_REPO"] + ":latest",
+        )
        print("ROLLBACK VERIFIED", flush=True)

    def secret_config(self):
@@ -390,7 +398,9 @@ class Deploy:
            command += ["--build-arg", key + "=" + value]
        self.run(*command, str(ROOT))
        service = config["services"]["pigway-frame-flow"]
-        service["image"] = image
+        latest = self.env["PROD_IMAGE_REPO"] + ":latest"
+        self.run("image", "tag", image, latest)
+        service["image"] = latest
        proposed = self.secret_config()
        service["environment"].update(proposed)
        for key in ("SYNOLOGY_PHOTOS_URL", "SYNOLOGY_PHOTOS_CERT_SHA256"):
@@ -664,6 +674,7 @@ class Deploy:
            self.state["status"] = "success"
            self.state_io(self.state)
            print("BOOTSTRAP VERIFIED", flush=True)
+            self.cleanup_images()
        except BaseException:
            print("BOOTSTRAP FAILED", flush=True)
            self.diagnostics()
@@ -723,6 +734,68 @@ class Deploy:
        text = re.sub(r"https?://[^\s\"'<>]+", "[REDACTED-URL]", text)
        print("Scoped container logs (redacted):", text, flush=True)

+    def cleanup_images(self):
+        """Keep only latest after verification; protect unrelated container images."""
+        try:
+            protected = set()
+            referenced_tags = set()
+            container_ids = self.run("ps", "-aq").stdout.decode().split()
+            if container_ids:
+                containers = json.loads(self.run("inspect", *container_ids).stdout)
+                protected.update(container["Image"] for container in containers)
+                referenced_tags.update(
+                    container.get("Config", {}).get("Image", "")
+                    for container in containers
+                )
+            repo = self.env["PROD_IMAGE_REPO"]
+            image_ids = (
+                self.run("image", "ls", "--no-trunc", "-q", repo)
+                .stdout.decode()
+                .split()
+            )
+            if not image_ids:
+                return
+            images = json.loads(
+                self.run("image", "inspect", *sorted(set(image_ids))).stdout
+            )
+            removed = 0
+            for image in images:
+                tags = image.get("RepoTags") or []
+                # Shared images and unrelated tags are never cleanup targets.
+                if not tags:
+                    continue
+                if not all(
+                    re.fullmatch(
+                        re.escape(repo)
+                        + r":(?:latest|release-[0-9a-f]{40}-[0-9a-f]{8}(?:-rollback)?|rollback-[0-9a-f]{32})",
+                        tag,
+                    )
+                    for tag in tags
+                ):
+                    continue
+                latest = repo + ":latest"
+                if image["Id"] in protected:
+                    # Current production uses latest. Other containers retain
+                    # their referenced tags and images, even when stopped.
+                    if latest not in tags:
+                        continue
+                    tags = [
+                        tag
+                        for tag in tags
+                        if tag != latest and tag not in referenced_tags
+                    ]
+                if not tags:
+                    continue
+                result = self.run("image", "rm", *tags, check=False)
+                if result.returncode:
+                    print("IMAGE RETENTION: referenced image retained", flush=True)
+                else:
+                    removed += 1
+            print(f"IMAGE RETENTION: removed {removed} obsolete images", flush=True)
+        except Exception:
+            # Cleanup failure must never roll back an already verified deployment.
+            print("IMAGE RETENTION WARNING: cleanup deferred", flush=True)
+
    def deploy(self):
        previous = self.preflight()
        self.lock()
@@ -732,7 +805,15 @@ class Deploy:
            # Refresh baseline under the daemon-level lock.
            previous = self.preflight()
            version = self.verify(previous)
-            pin = self.env["PROD_IMAGE_REPO"] + ":rollback-" + self.owner
+            release = self.old_image_tag.removesuffix("-rollback")
+            if re.fullmatch(
+                re.escape(self.env["PROD_IMAGE_REPO"])
+                + r":release-[0-9a-f]{40}-[0-9a-f]{8}",
+                release,
+            ):
+                pin = release + "-rollback"
+            else:
+                pin = self.env["PROD_IMAGE_REPO"] + ":rollback-" + self.owner
            self.run("image", "tag", self.old_image, pin)
            candidate = self.build(json.loads(json.dumps(previous)))
            self.state = {
@@ -791,6 +872,10 @@ class Deploy:
            if safe_unlock:
                self.unlock()

+        # The deployment lock itself references the old image. Release it
+        # before retention; other container references remain protected.
+        self.cleanup_images()
+
    def recover(self, manual=False):
        self.settings()
        lock = self.run(
@@ -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"
  }
 }
@@ -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]
@@ -152,3 +152,55 @@ def test_offline(service):
    id = submit(service)
    service.process_one()
    assert service.operation(id)["error"]["code"] == "DEVICE_OFFLINE"
+
+
+def test_late_sync_resolves_unknown_without_upload_retry(service, monkeypatch):
+    id = submit(service)
+    service.process_one()
+    service.update(id, "UNKNOWN")
+    driver = service.registry.get("living-room")
+    calls = []
+    device = driver.device
+    monkeypatch.setattr(driver, "device", lambda **kw: calls.append(kw) or device(**kw))
+    service.sync()
+    assert service.operation(id)["status"] == "COMPLETED"
+    assert driver.client.uploads == 1
+    assert calls == [{"refresh": True}]
+    assert not service.blocked("living-room")
+    assert service.db.one(
+        "SELECT confirmed_at FROM managed_cloud_uploads WHERE operation_id=?", (id,)
+    )["confirmed_at"]
+
+
+def test_unknown_wrong_target_stays_locked_and_queries_at_most_once_per_minute(
+    service, monkeypatch
+):
+    id = submit(service)
+    service.process_one()
+    service.update(id, "UNKNOWN")
+    driver = service.registry.get("living-room")
+    driver.client.display(1)
+    driver.invalidate()
+    calls = []
+    device = driver.device
+    monkeypatch.setattr(driver, "device", lambda **kw: calls.append(kw) or device(**kw))
+    service.sync()
+    service.sync()
+    assert service.operation(id)["status"] == "UNKNOWN"
+    assert service.blocked("living-room")
+    assert driver.client.uploads == 1
+    assert len(calls) == 1
+
+
+def test_unknown_matching_image_without_sync_stays_locked(service, monkeypatch):
+    id = submit(service)
+    service.process_one()
+    service.update(id, "UNKNOWN")
+    driver = service.registry.get("living-room")
+    device = driver.device()
+    device.current_content.synced = False
+    monkeypatch.setattr(driver, "device", lambda **kw: device)
+    service.sync()
+    assert service.operation(id)["status"] == "UNKNOWN"
+    assert service.blocked("living-room")
+    assert driver.client.uploads == 1
@@ -373,3 +373,30 @@ def test_ai_layout_sends_metadata_only_and_validates_response(
    assert b"test-ai-secret" not in calls[0]["body"]
    assert image.getvalue() not in calls[0]["body"]
    assert calls[0]["headers"]["Authorization"] == "Bearer test-ai-secret"
+
+
+def test_webhook_matching_event_after_unrelated_backlog(client, service, monkeypatch):
+    import json
+
+    service.settings.image_url_hosts.append("example.com")
+    response = client.post(
+        "/api/v1/webhooks",
+        json={
+            "callback_url": "https://example.com/events",
+            "event_types": ["operation.failed"],
+        },
+    )
+    assert response.status_code == 201
+    for _ in range(500):
+        service.db.event("display.changed", {})
+    service.db.event("operation.failed", {"operation_id": "expected"})
+    sent = []
+    monkeypatch.setattr(
+        service.assets, "fetch_url", lambda url, **kwargs: sent.append(kwargs) or b""
+    )
+    service.webhooks.tick()
+    service.webhooks.last = 0
+    service.webhooks.tick()
+    assert len(sent) == 1
+    assert json.loads(sent[0]["body"])["data"] == {"operation_id": "expected"}
+    assert len(service.db.all("SELECT * FROM webhook_deliveries")) == 1
@@ -98,6 +98,7 @@ def test_preflight_rejects_wrong_host_resources(tool, monkeypatch, change):
 def test_deploy_rolls_back_exact_baseline_without_deleting_data(
    tool, monkeypatch, failure
 ):
+    tool.old_image_tag = "pigway-frame-flow:release-" + "a" * 40 + "-" + "b" * 8
    old = {"services": {"pigway-frame-flow": {"image": "sha256:previous"}}}
    new = {"services": {"pigway-frame-flow": {"image": "candidate"}}}
    activated, calls, saved = [], [], []
@@ -111,6 +112,10 @@ def test_deploy_rolls_back_exact_baseline_without_deleting_data(
    monkeypatch.setattr(
        tool, "state_io", lambda state: saved.append(json.loads(json.dumps(state)))
    )
+    cleanup = []
+    monkeypatch.setattr(
+        tool, "cleanup_images", lambda: cleanup.append(saved[-1]["status"])
+    )
    verifies = []

    def verify(config):
@@ -125,13 +130,17 @@ def test_deploy_rolls_back_exact_baseline_without_deleting_data(
    if failure:
        with pytest.raises(RuntimeError):
            tool.deploy()
+        assert cleanup == []
        assert activated == [new, old]
        assert saved[0]["previous"] == old
        assert ("unlock" in calls) is (failure != "rollback")
    else:
        tool.deploy()
+        assert cleanup == ["success"]
+        assert "unlock" in calls
        assert activated == [new]
        assert saved[-1]["status"] == "success"
+        assert saved[-1]["rollback_tag"] == tool.old_image_tag + "-rollback"
    assert not any(
        isinstance(c, tuple) and ("down" in c or "prune" in c) for c in calls
    )
@@ -243,3 +252,64 @@ def test_lock_conflict_never_activates_production(tool, monkeypatch):
    monkeypatch.setattr(tool, "run", fail)
    with pytest.raises(RuntimeError):
        tool.deploy()
+
+
+def test_image_retention_keeps_latest_and_protects_other_containers(tool, monkeypatch):
+    tool.state = {"previous_image": "previous"}
+    release = "pigway-frame-flow:release-" + "a" * 40 + "-" + "b" * 8
+    rollback = "pigway-frame-flow:rollback-" + "c" * 32
+    tool.state["rollback_tag"] = release + "-rollback"
+    images = [
+        {"Id": "current", "RepoTags": ["pigway-frame-flow:latest", release]},
+        {"Id": "previous", "RepoTags": [release + "-rollback", release, rollback]},
+        {"Id": "stopped", "RepoTags": [release]},
+        {"Id": "obsolete", "RepoTags": [release, rollback]},
+        {"Id": "shared", "RepoTags": [release, "other:latest"]},
+        {"Id": "ci", "RepoTags": ["ffci-fixture:latest"]},
+    ]
+    calls = []
+
+    def run(*args, **kwargs):
+        calls.append(args)
+        if args == ("ps", "-aq"):
+            data = "container1 container2"
+        elif args[0] == "inspect":
+            data = json.dumps(
+                [
+                    {
+                        "Image": "current",
+                        "Config": {"Image": "pigway-frame-flow:latest"},
+                    },
+                    {"Image": "stopped"},
+                ]
+            )
+        elif args[:2] == ("image", "ls"):
+            data = "current previous stopped obsolete shared ci"
+        elif args[:2] == ("image", "inspect"):
+            data = json.dumps(images)
+        else:
+            data = ""
+        return SimpleNamespace(stdout=data.encode(), returncode=0)
+
+    monkeypatch.setattr(tool, "run", run)
+    tool.cleanup_images()
+    assert [c for c in calls if c[:2] == ("image", "rm")] == [
+        ("image", "rm", release),
+        ("image", "rm", release + "-rollback", release, rollback),
+        ("image", "rm", release, rollback),
+    ]
+
+
+def test_image_retention_failure_does_not_fail_verified_deploy(
+    tool, monkeypatch, capsys
+):
+    tool.state = {"previous_image": "previous"}
+
+    def unavailable(*args, **kwargs):
+        raise RuntimeError("private diagnostic")
+
+    monkeypatch.setattr(tool, "run", unavailable)
+    tool.cleanup_images()
+    output = capsys.readouterr().out
+    assert "cleanup deferred" in output
+    assert "private diagnostic" not in output
@@ -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")
@@ -75,3 +75,76 @@ def test_playlist(app, service):
 def test_invalid_cron():
    with pytest.raises(FlowError):
        next_run(Schedule(type="cron", cron="* * * * * *"))
+
+
+def test_worker_checks_late_sync_before_due_tasks(app, service, monkeypatch):
+    scheduler = app.state.scheduler
+    calls = []
+    monkeypatch.setattr(service, "sync", lambda: calls.append("sync"))
+
+    def tick():
+        calls.append("tick")
+        scheduler.stop_event.set()
+
+    monkeypatch.setattr(scheduler, "tick", tick)
+    monkeypatch.setattr(service, "expire", lambda: None)
+    monkeypatch.setattr(service.cloud_library, "sweep", lambda: None)
+    monkeypatch.setattr(service.local_retention, "sweep", lambda: None)
+    monkeypatch.setattr(service, "process_one", lambda: False)
+    scheduler.run()
+    assert calls[:2] == ["sync", "tick"]
+
+
+def test_blocked_job_keeps_retry_slot_without_blocking_worker(
+    app, service, monkeypatch
+):
+    scheduler = app.state.scheduler
+    service.registry.drivers["other-room"] = service.registry.get("living-room")
+    jobs = []
+    for device in ("living-room", "other-room"):
+        jobs.append(
+            scheduler.save_job(
+                JobInput(
+                    name=device,
+                    device_id=device,
+                    schedule={"type": "daily", "time": "07:00"},
+                    source={"type": "restore"},
+                )
+            )
+        )
+    current = datetime.now(UTC)
+    due = (current - timedelta(seconds=2)).isoformat()
+    service.db.execute("UPDATE jobs SET next_run=?", (due,))
+    unknown = service.submit("display", {}, "living-room", "test", "unknown")
+    service.db.execute(
+        "UPDATE operations SET status='UNKNOWN',write_started=1 WHERE id=?", (unknown,)
+    )
+    queued = service.submit("render", {}, "living-room", "test", "render")
+    calls = []
+    monkeypatch.setattr(service, "sync", lambda: None)
+    monkeypatch.setattr(service, "expire", lambda: calls.append("expire"))
+    monkeypatch.setattr(service.cloud_library, "sweep", lambda: None)
+    monkeypatch.setattr(service.local_retention, "sweep", lambda: None)
+
+    def process():
+        calls.append("process")
+        assert service.operation(queued)["status"] == "QUEUED"
+        scheduler.stop_event.set()
+        return False
+
+    monkeypatch.setattr(service, "process_one", process)
+    monkeypatch.setattr(
+        scheduler.stop_event, "wait", lambda timeout: scheduler.stop_event.set()
+    )
+    scheduler.run()
+    assert calls == ["expire", "process"]
+    blocked = service.db.one("SELECT * FROM jobs WHERE id=?", (jobs[0]["id"],))
+    assert blocked["next_run"] == due
+    assert blocked["last_run"] is None
+    assert len(service.db.all("SELECT * FROM job_runs")) == 1
+    assert service.db.one("SELECT * FROM job_runs")["job_id"] == jobs[1]["id"]
+    scheduler.tick(current + timedelta(minutes=6))
+    assert (
+        service.db.one("SELECT * FROM jobs WHERE id=?", (jobs[0]["id"],))["last_result"]
+        == "SKIPPED_OVERDUE"
+    )
@@ -0,0 +1,153 @@
+import json
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from test_display import submit
+
+from pigway_frame_flow.errors import ErrorCode, FlowError
+
+
+def lose_response(service, monkeypatch, change_display=True):
+    driver = service.registry.get("living-room")
+    upload = driver.upload
+
+    def uncertain(path, name):
+        upload(path, name)
+        if not change_display:
+            driver.client.display(1)
+        raise FlowError(
+            ErrorCode.WRITE_RESULT_UNKNOWN, "Cloud write outcome is unknown", 502
+        )
+
+    monkeypatch.setattr(driver, "upload", uncertain)
+    operation = submit(service)
+    service.process_one()
+    assert service.operation(operation)["status"] == "UNKNOWN"
+    return operation, driver
+
+
+def test_lost_upload_response_resolves_by_exact_bytes_without_reupload(
+    service, monkeypatch
+):
+    operation, driver = lose_response(service, monkeypatch)
+    service.sync()
+    assert service.operation(operation)["status"] == "COMPLETED"
+    assert service.operation(operation)["result"]["image_id"] == 2
+    assert driver.client.uploads == 1
+    assert not service.blocked("living-room")
+
+
+def test_same_name_wrong_bytes_does_not_unlock(service, monkeypatch):
+    operation, driver = lose_response(service, monkeypatch)
+    monkeypatch.setattr(driver, "download", lambda _: b"unrelated")
+    service.sync()
+    assert service.operation(operation)["status"] == "UNKNOWN"
+    assert service.blocked("living-room")
+    assert driver.client.uploads == 1
+
+
+def test_ambiguous_gallery_does_not_download_or_unlock(service, monkeypatch):
+    operation, driver = lose_response(service, monkeypatch)
+    driver.client.state["images"].append(dict(driver.client.state["images"][-1], id=3))
+    monkeypatch.setattr(driver, "download", lambda _: pytest.fail("Ambiguous download"))
+    service.sync()
+    assert service.operation(operation)["status"] == "UNKNOWN"
+
+
+def test_confirmed_upload_not_displayed_closes_failed_display_after_stable_checks(
+    service, monkeypatch
+):
+    operation, driver = lose_response(service, monkeypatch, change_display=False)
+    service.sync()
+    result = service.operation(operation)["result"]
+    assert result["upload_confirmed"]
+    start = datetime.fromisoformat(result["upload_intent"]["started_at"])
+    import pigway_frame_flow.display.upload_reconciliation as module
+
+    class Later(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return start + timedelta(seconds=service.settings.sync_timeout + 600)
+
+    monkeypatch.setattr(module, "datetime", Later)
+    service.unknown_checked.clear()
+    service.sync()
+    assert service.operation(operation)["status"] == "UNKNOWN"
+    result = service.operation(operation)["result"]
+    result["unchanged_observed_at"] = (
+        datetime.now(UTC) - timedelta(seconds=61)
+    ).isoformat()
+    service.db.execute(
+        "UPDATE operations SET result=? WHERE id=?", (json.dumps(result), operation)
+    )
+    service.unknown_checked.clear()
+    service.sync()
+    assert service.operation(operation)["status"] == "FAILED"
+    assert service.operation(operation)["error"]["code"] == "DISPLAY_NOT_CONFIRMED"
+    assert not service.blocked("living-room")
+    assert driver.client.uploads == 1
+
+
+def test_legacy_unknown_without_intent_remains_locked(service):
+    operation = submit(service)
+    service._start_write(operation)
+    service.update(operation, "UNKNOWN")
+    service.sync()
+    assert service.blocked("living-room")
+
+
+def test_recovered_upload_preserves_temporary_restore(service, monkeypatch):
+    driver = service.registry.get("living-room")
+    upload = driver.upload
+
+    def uncertain(path, name):
+        upload(path, name)
+        raise FlowError(ErrorCode.WRITE_RESULT_UNKNOWN, "Response lost", 502)
+
+    monkeypatch.setattr(driver, "upload", uncertain)
+    operation = submit(service, duration=60, restore=True)
+    service.process_one()
+    service.sync()
+    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"]
Author	SHA1	Message	Date
渡口浪人	a843d4de9b	Add documented KoKonna failure policy, bounded retries and durable recovery escalation [skip ci]	2026-10-16 12:25:16 +08:00
渡口浪人	d2f02d8054	Preserve uncertain-write evidence and safe transport diagnostics for v0.1.3 [skip ci]	2026-10-16 10:30:08 +08:00
渡口浪人	8a3c616b5f	Fix webhook starvation and isolate scheduled job failures for v0.1.2 [skip ci]	2026-10-16 09:48:51 +08:00
渡口浪人	e1b001a431	Reconcile lost upload responses without replaying frame writes	2026-10-14 13:10:33 +08:00
渡口浪人	e9d7b12c66	Release deployment lock before pruning old production image	2026-10-12 21:12:44 +08:00
渡口浪人	b46fbea4c9	Retain only the verified production latest image	2026-10-12 20:51:08 +08:00
渡口浪人	df1eba74e8	Keep one rollback tag derived from original release	2026-10-11 17:15:56 +08:00
渡口浪人	b41163965e	docs: clarify Raspberry Pi project names and integration scope	2026-10-11 16:29:30 +08:00
渡口浪人	aabbff8127	Release version 0.1.1 [skip ci]	2026-10-11 14:43:29 +08:00
渡口浪人	7cdf712829	Reconcile delayed frame sync before scheduled writes	2026-10-11 14:24:32 +08:00
渡口浪人	327fa53d7b	Retain current production image and one rollback baseline	2026-10-11 09:14:23 +08:00