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

tests/schema-contract.json

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

tests/test_client.py

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

tests/test_recovery.py

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