Fix webhook starvation and isolate scheduled job failures for v0.1.2 [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.2",
         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.2",
             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=?",
             (
@@ -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")

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/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"

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"])
+            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",
+                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"], sub["id"]),
+                (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()),

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.2"
 
 
 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.2"
     api_version: str = "v1"
     commit: str = "development"
     build_time: str = "unknown"

pigway_frame_flow/scheduler/engine.py

@@ -147,61 +147,72 @@ class Scheduler:
                 "SELECT * FROM jobs WHERE enabled=1 AND next_run<=?",
                 (current.isoformat(),),
             ):
-                job = JobInput.model_validate_json(row["payload"])
+                try:
-                slot = row["next_run"]
+                    job = JobInput.model_validate_json(row["payload"])
-                if self.db.one(
+                    slot = row["next_run"]
-                    "SELECT * FROM job_runs WHERE job_id=? AND scheduled_at=?",
+                    if self.db.one(
-                    (row["id"], slot),
+                        "SELECT * FROM job_runs WHERE job_id=? AND scheduled_at=?",
-                ):
+                        (row["id"], slot),
-                    continue
+                    ):
-                # Skip overdue tasks (>5 minutes): avoid an unexpected artwork
+                        continue
-                # overwrite after downtime. One-time tasks are retired too.
+                    # Skip overdue tasks (>5 minutes): avoid an unexpected artwork
-                late = (current - datetime.fromisoformat(slot)).total_seconds() > 300
+                    # overwrite after downtime. One-time tasks are retired too.
-                operation_id = None
+                    late = (
-                if not late:
+                        current - datetime.fromisoformat(slot)
-                    action = "restore" if job.source.type == "restore" else "display"
+                    ).total_seconds() > 300
-                    payload = (
+                    operation_id = None
-                        {}
+                    if not late:
-                        if isinstance(job.source, RestoreContent)
+                        action = (
-                        else DisplayRequest(
+                            "restore" if job.source.type == "restore" else "display"
-                            device_id=job.device_id,
+                        )
-                            content=job.source,
+                        payload = (
-                            options=Options(
+                            {}
-                                duration=job.display_duration,
+                            if isinstance(job.source, RestoreContent)
-                                restore_previous=job.restore_previous,
+                            else DisplayRequest(
-                                priority=job.priority,
+                                device_id=job.device_id,
-                                cleanup_unstarred=job.cleanup_unstarred,
+                                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")
+                        )
-                    )
+                except Exception:
-                    operation_id = self.service.submit(
+                    # Keep this slot pending for retry or the existing overdue policy.
-                        action,
+                    # A single device must not prevent other jobs or queue consumption.
-                        payload,
+                    logging.getLogger("frame-flow").error(
-                        job.device_id,
+                        "Scheduled job failed; details suppressed"
-                        "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"],
-                        ),
                     )
             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.2</small></div></aside><div class="workspace"><header><div><span class="eyebrow">PIGWAY FRAME FLOW</span><h1 id="page-title"></h1></div><div class="toolbar"><select id="device" aria-label="Device"></select><button id="lang" class="icon-button" aria-label="Change language">EN</button><button id="theme" class="icon-button" aria-label="Change theme">◐</button></div></header><main id="main"></main><footer id="footer"></footer></div><div id="toast" role="status" aria-live="polite"></div><dialog id="modal"><div id="modal-content"></div></dialog><script src="/ui/i18n.js"></script><script src="/ui/features.js"></script><script src="/ui/extensions.js"></script><script src="/ui/app.js"></script></body></html>

pyproject.toml

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