Preserve uncertain-write evidence and safe transport diagnostics for v0.1.3 [skip ci]

README.md

@@ -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)

README.zh-CN.md

@@ -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)

docs/REAL_DEVICE_TEST.md

@@ -25,3 +25,19 @@
 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.

docs/deployment/GITEA_CD.md

@@ -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
@@ -366,13 +366,15 @@ complete audit succeeded.
 
 ### Production image retention
 
-After a deployment passes verification and its success journal is saved, Frame Flow
+Production uses `pigway-frame-flow:latest`. Builds first create a unique release
-removes obsolete `release-<commit>-<owner>` and `rollback-<owner>` images from the
+image, then tag it as latest before activation. The old image is pinned during
-configured production image repository. It retains the running version and the
+an update so a failed deployment can restore it. Only after the new version
-previous deployment baseline for one-step rollback. Images referenced by any
+passes verification are old release/rollback images and duplicate tags removed.
-container (including stopped containers), images with unrelated tags, and CI
+After successful deployment, only the current production image remains; there
-images are protected. Retention does not prune build cache, volumes, or data.
+is no retained historical image for manual rollback. Rebuild a historical
-Failed deployments do not run retention; cleanup failure produces a warning and
+revision before manually reverting a successful deployment.
-never rolls back a successfully verified deployment. Historical deployment
+
-journals remain available for audit, but only the latest rollback image is
+Images referenced by other containers (including stopped containers), images
-retained locally; older releases require rebuilding before use.
+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.

pigway_frame_flow/api/app.py

@@ -124,7 +124,7 @@ def create_app(settings=None, start_worker=True):
     app = FastAPI(
         title="Pigway Frame Flow API",
         docs_url=None,
-        version="0.1.1",
+        version="0.1.3",
         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.1",
+            version="0.1.3",
             mock=settings.mock,
             write_enabled=settings.write_enabled,
             api_key_configured=bool(settings.api_key)

pigway_frame_flow/display/service.py

@@ -7,6 +7,7 @@ from uuid import uuid4
 
 from pigway_frame_flow.display.cloud_library import CloudLibrary
 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,6 +50,7 @@ class DisplayService:
         self.lock = threading.RLock()
         self.retention = Retention(self)
         self.cloud_library = CloudLibrary(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.
@@ -169,8 +171,9 @@ class DisplayService:
         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:
+            for key in ("photo_attempts", "upload_intent"):
-                result = {**result, "photo_attempts": previous["photo_attempts"]}
+                if key in previous:
+                    result = {**result, key: previous[key]}
         self.db.execute(
             "UPDATE operations SET status=?,updated_at=?,result=COALESCE(?,result),error=? WHERE id=?",
             (
@@ -439,7 +442,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
@@ -627,6 +630,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")
@@ -690,22 +695,31 @@ 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='display' AND write_started=1"
         ):
             expected = json.loads(op["result"] or "{}").get("image_id")
-            if not expected:
+            intent = json.loads(op["result"] or "{}").get("upload_intent")
+            if not expected and not intent:
                 continue
             checked = time.monotonic()
             if checked - self.unknown_checked.get(op["id"], float("-inf")) < 60:
                 continue
             self.unknown_checked[op["id"]] = checked
             try:
-                device = self.registry.get(op["device_id"]).device(refresh=True)
+                driver = self.registry.get(op["device_id"])
+                device = driver.device(refresh=True)
+                if intent:
+                    expected = self.upload_reconciliation.check(op, driver, device)
                 if (
-                    device.current_content.id == expected
+                    expected
+                    and device.current_content.id == expected
                     and device.current_content.synced
                 ):
                     self.update(op["id"], "COMPLETED")
@@ -847,7 +861,20 @@ class DisplayService:
                 self.update(
                     id,
                     "COMPLETED" if confirmed else "FAILED",
-                    {"image_id": device.current_content.id},
+                    op["result"],
+                    error=None
+                    if confirmed
+                    else {
+                        "code": E.DISPLAY_NOT_CONFIRMED
+                        if op["action"] == "display"
+                        else E.WRITE_RESULT_UNKNOWN,
+                        "message": "Manually closed unconfirmed write; no write replayed",
+                        "request_id": op["request_id"],
+                        "details": {
+                            "observed_image_id": device.current_content.id,
+                            "previous_error": op["error"],
+                        },
+                    },
                 )
                 resolved = True
                 if confirmed and op["action"] == "display":

pigway_frame_flow/display/upload_reconciliation.py

@@ -0,0 +1,157 @@
+"""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:
+                return None
+            expected = candidates[0].image_id
+            data = driver.download(expected)
+            if (
+                len(data) != intent["size"]
+                or hashlib.sha256(data).hexdigest() != intent["sha256"]
+            ):
+                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

pigway_frame_flow/drivers/kokonna/client.py

@@ -51,13 +51,30 @@ class SafeTransport:
                 timeout=(5, 120 if endpoint == "/upload" else 30),
                 allow_redirects=False,
             )
-        except requests.RequestException:
+        except requests.RequestException as exc:
+            # Persist only fixed categories: exception text and download URLs may
+            # contain credentials. Unknown writes must still never be replayed.
+            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",
+            )
             raise FlowError(
                 E.WRITE_RESULT_UNKNOWN if write else E.CLOUD_UNAVAILABLE,
                 "Cloud write outcome is unknown"
                 if write
                 else "Device cloud is unavailable",
                 502,
+                details={"transport_error": category},
             ) from None
         if response.status_code != 200:
             code = (
@@ -73,6 +90,7 @@ class SafeTransport:
                 code,
                 "Device cloud request failed",
                 429 if code == E.RATE_LIMITED else 502,
+                details={"http_status": response.status_code},
             )
         if binary:
             if len(response.content) > 20 * 1024 * 1024:

pigway_frame_flow/errors.py

@@ -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)

pigway_frame_flow/extensions/webhooks.py

@@ -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"])
-            for event in db.all(
+            if not types:
-                "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"]),
-            ):
-                if event["event_type"] not in types:
                 continue
+            placeholders = ",".join("?" for _ in types)
+            for event in db.all(
+                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"]),
+            ):
                 db.execute(
                     "INSERT OR IGNORE INTO webhook_deliveries VALUES(?,?,'QUEUED',0,?,NULL)",
                     (sub["id"], event["id"], now_iso()),

pigway_frame_flow/models.py

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

pigway_frame_flow/scheduler/engine.py

@@ -147,6 +147,7 @@ class Scheduler:
                 "SELECT * FROM jobs WHERE enabled=1 AND next_run<=?",
                 (current.isoformat(),),
             ):
+                try:
                     job = JobInput.model_validate_json(row["payload"])
                     slot = row["next_run"]
                     if self.db.one(
@@ -156,10 +157,14 @@ class Scheduler:
                         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
+                    late = (
+                        current - datetime.fromisoformat(slot)
+                    ).total_seconds() > 300
                     operation_id = None
                     if not late:
-                    action = "restore" if job.source.type == "restore" else "display"
+                        action = (
+                            "restore" if job.source.type == "restore" else "display"
+                        )
                         payload = (
                             {}
                             if isinstance(job.source, RestoreContent)
@@ -203,6 +208,12 @@ class Scheduler:
                                 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(),)
             ):

pigway_frame_flow/web/index.html

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

pyproject.toml

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

scripts/production-deploy.py

@@ -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()
@@ -724,13 +735,18 @@ class Deploy:
         print("Scoped container logs (redacted):", text, flush=True)
 
     def cleanup_images(self):
-        """Retain container references and the verified deployment's rollback baseline."""
+        """Keep only latest after verification; protect unrelated container images."""
         try:
-            protected = {self.state["previous_image"]}
+            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)
@@ -746,17 +762,30 @@ class Deploy:
             for image in images:
                 tags = image.get("RepoTags") or []
                 # Shared images and unrelated tags are never cleanup targets.
-                if image["Id"] in protected or not tags:
+                if not tags:
                     continue
                 if not all(
                     re.fullmatch(
                         re.escape(repo)
-                        + r":(?:release-[0-9a-f]{40}-[0-9a-f]{8}|rollback-[0-9a-f]{32})",
+                        + 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)
@@ -776,6 +805,14 @@ class Deploy:
             # Refresh baseline under the daemon-level lock.
             previous = self.preflight()
             version = self.verify(previous)
+            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)))
@@ -813,7 +850,6 @@ class Deploy:
             self.state["status"] = "success"
             self.state_io(self.state)
             print("DEPLOY VERIFIED", flush=True)
-            self.cleanup_images()
         except BaseException as exc:
             print("DEPLOY FAILED:", type(exc).__name__, flush=True)
             self.diagnostics()
@@ -836,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(

tests/test_client.py

@@ -118,3 +118,26 @@ def test_rate_budget():
         c.get_device()
     assert e.value.code == "RATE_LIMITED"
     assert s.request.call_count == 18
+
+
+@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 == "WRITE_RESULT_UNKNOWN"
+    assert error.value.details == {"transport_error": category}
+    assert "secret" not in str(error.value.details)
+    assert s.request.call_count == 1

tests/test_extensions.py

@@ -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

tests/test_production_deploy.py

@@ -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 = [], [], []
@@ -136,8 +137,10 @@ def test_deploy_rolls_back_exact_baseline_without_deleting_data(
     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
     )
@@ -251,13 +254,14 @@ def test_lock_conflict_never_activates_production(tool, monkeypatch):
         tool.deploy()
 
 
-def test_image_retention_protects_containers_rollback_and_other_tags(tool, monkeypatch):
+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": [release]},
+        {"Id": "current", "RepoTags": ["pigway-frame-flow:latest", release]},
-        {"Id": "previous", "RepoTags": [rollback]},
+        {"Id": "previous", "RepoTags": [release + "-rollback", release, rollback]},
         {"Id": "stopped", "RepoTags": [release]},
         {"Id": "obsolete", "RepoTags": [release, rollback]},
         {"Id": "shared", "RepoTags": [release, "other:latest"]},
@@ -270,7 +274,15 @@ def test_image_retention_protects_containers_rollback_and_other_tags(tool, monke
         if args == ("ps", "-aq"):
             data = "container1 container2"
         elif args[0] == "inspect":
-            data = json.dumps([{"Image": "current"}, {"Image": "stopped"}])
+            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"):
@@ -282,7 +294,9 @@ def test_image_retention_protects_containers_rollback_and_other_tags(tool, monke
     monkeypatch.setattr(tool, "run", run)
     tool.cleanup_images()
     assert [c for c in calls if c[:2] == ("image", "rm")] == [
-        ("image", "rm", release, rollback)
+        ("image", "rm", release),
+        ("image", "rm", release + "-rollback", release, rollback),
+        ("image", "rm", release, rollback),
     ]
 
 

tests/test_scheduler.py

@@ -93,3 +93,58 @@ def test_worker_checks_late_sync_before_due_tasks(app, service, monkeypatch):
     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"
+    )

tests/test_upload_reconciliation.py

@@ -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"]