Reconcile delayed frame sync before scheduled writes

pigway_frame_flow/display/service.py

@@ -50,6 +50,7 @@ class DisplayService:
         self.retention = Retention(self)
         self.cloud_library = CloudLibrary(self)
         self.expiration_checked: dict[str, float] = {}
+        self.unknown_checked: dict[str, float] = {}
         # A killed write cannot be replayed safely; pre-write work can resume.
         db.execute(
             "UPDATE operations SET status='UNKNOWN', updated_at=? WHERE write_started=1 AND status IN ('UPLOADING','RENDERING')",
@@ -689,6 +690,31 @@ class DisplayService:
         )
 
     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:
+                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)
+                if (
+                    device.current_content.id == expected
+                    and device.current_content.synced
+                ):
+                    self.update(op["id"], "COMPLETED")
+                    self.cloud_library.confirm(op["id"])
+                    self.db.history(op, "COMPLETED")
+                    self.db.event("operation.completed", {"operation_id": op["id"]})
+                    self.unknown_checked.pop(op["id"], None)
+            except FlowError:
+                pass
         for op in self.db.all("SELECT * FROM operations WHERE status='SYNCING'"):
             try:
                 device = self.registry.get(op["device_id"]).device()
@@ -824,6 +850,9 @@ class DisplayService:
                     {"image_id": device.current_content.id},
                 )
                 resolved = True
+                if confirmed and op["action"] == "display":
+                    self.cloud_library.confirm(id)
+                    self.db.event("operation.completed", {"operation_id": id})
                 self.db.history(
                     dict(
                         id=id,

pigway_frame_flow/scheduler/engine.py

@@ -269,6 +269,9 @@ class Scheduler:
     def run(self):
         while not self.stop_event.is_set():
             try:
+                # Resolve late sync before submitting due jobs: a write lock
+                # otherwise makes tick fail before reconciliation can run.
+                self.service.sync()
                 self.tick()
                 self.service.expire()
                 self.service.cloud_library.sweep()

tests/test_display.py

@@ -152,3 +152,55 @@ def test_offline(service):
     id = submit(service)
     service.process_one()
     assert service.operation(id)["error"]["code"] == "DEVICE_OFFLINE"
+
+
+def test_late_sync_resolves_unknown_without_upload_retry(service, monkeypatch):
+    id = submit(service)
+    service.process_one()
+    service.update(id, "UNKNOWN")
+    driver = service.registry.get("living-room")
+    calls = []
+    device = driver.device
+    monkeypatch.setattr(driver, "device", lambda **kw: calls.append(kw) or device(**kw))
+    service.sync()
+    assert service.operation(id)["status"] == "COMPLETED"
+    assert driver.client.uploads == 1
+    assert calls == [{"refresh": True}]
+    assert not service.blocked("living-room")
+    assert service.db.one(
+        "SELECT confirmed_at FROM managed_cloud_uploads WHERE operation_id=?", (id,)
+    )["confirmed_at"]
+
+
+def test_unknown_wrong_target_stays_locked_and_queries_at_most_once_per_minute(
+    service, monkeypatch
+):
+    id = submit(service)
+    service.process_one()
+    service.update(id, "UNKNOWN")
+    driver = service.registry.get("living-room")
+    driver.client.display(1)
+    driver.invalidate()
+    calls = []
+    device = driver.device
+    monkeypatch.setattr(driver, "device", lambda **kw: calls.append(kw) or device(**kw))
+    service.sync()
+    service.sync()
+    assert service.operation(id)["status"] == "UNKNOWN"
+    assert service.blocked("living-room")
+    assert driver.client.uploads == 1
+    assert len(calls) == 1
+
+
+def test_unknown_matching_image_without_sync_stays_locked(service, monkeypatch):
+    id = submit(service)
+    service.process_one()
+    service.update(id, "UNKNOWN")
+    driver = service.registry.get("living-room")
+    device = driver.device()
+    device.current_content.synced = False
+    monkeypatch.setattr(driver, "device", lambda **kw: device)
+    service.sync()
+    assert service.operation(id)["status"] == "UNKNOWN"
+    assert service.blocked("living-room")
+    assert driver.client.uploads == 1

tests/test_scheduler.py

@@ -75,3 +75,21 @@ def test_playlist(app, service):
 def test_invalid_cron():
     with pytest.raises(FlowError):
         next_run(Schedule(type="cron", cron="* * * * * *"))
+
+
+def test_worker_checks_late_sync_before_due_tasks(app, service, monkeypatch):
+    scheduler = app.state.scheduler
+    calls = []
+    monkeypatch.setattr(service, "sync", lambda: calls.append("sync"))
+
+    def tick():
+        calls.append("tick")
+        scheduler.stop_event.set()
+
+    monkeypatch.setattr(scheduler, "tick", tick)
+    monkeypatch.setattr(service, "expire", lambda: None)
+    monkeypatch.setattr(service.cloud_library, "sweep", lambda: None)
+    monkeypatch.setattr(service.local_retention, "sweep", lambda: None)
+    monkeypatch.setattr(service, "process_one", lambda: False)
+    scheduler.run()
+    assert calls[:2] == ["sync", "tick"]