fix(lease): ignore benign observer idle replies

packages/mosaic/framework/tools/lease-broker/receipt-observer-client.py

@@ -22,13 +22,23 @@ from typing import Final
 MAX_FRAME: Final = 64 * 1024
 BROKER_TIMEOUT_SECONDS: Final = 1.5
 MAX_TRANSCRIPT_BYTES: Final = 4 * 1024 * 1024
+BENIGN_OBSERVATION_UNAVAILABLE_CODE: Final = "OBSERVATION_UNAVAILABLE"
+
+
+def reject_duplicate_json_keys(pairs: list[tuple[str, object]]) -> dict[str, object]:
+    value: dict[str, object] = {}
+    for key, item in pairs:
+        if key in value:
+            raise ValueError("duplicate observer JSON key")
+        value[key] = item
+    return value
 
 
 def read_json(stream: object) -> dict[str, object]:
     raw = getattr(stream, "buffer", stream).read(MAX_FRAME + 1)
     if not isinstance(raw, bytes) or len(raw) > MAX_FRAME:
         raise ValueError("invalid observer input")
-    value = json.loads(raw)
+    value = json.loads(raw, object_pairs_hook=reject_duplicate_json_keys)
     if not isinstance(value, dict):
         raise ValueError("invalid observer input")
     return value
@@ -100,9 +110,9 @@ def observer_request(socket_path: Path, request: dict[str, object]) -> dict[str,
             if not chunk:
                 break
             response.extend(chunk)
-    if len(response) > MAX_FRAME or not response.endswith(b"\n"):
+    if len(response) > MAX_FRAME or response.count(b"\n") != 1 or not response.endswith(b"\n"):
         raise ValueError("invalid observer reply")
-    value = json.loads(response)
+    value = json.loads(response[:-1], object_pairs_hook=reject_duplicate_json_keys)
     if not isinstance(value, dict):
         raise ValueError("invalid observer reply")
     return value
@@ -133,10 +143,18 @@ def main(argv: Sequence[str] | None = None, *, environ: Mapping[str, str] | None
             "runtime": arguments.runtime,
             "latest_assistant_message": message,
         })
-    except (KeyError, OSError, ValueError, json.JSONDecodeError) as error:
+    except (KeyError, OSError, RecursionError, ValueError, json.JSONDecodeError) as error:
         print(f"Mosaic receipt observer refused: {error}", file=sys.stderr)
         return 2
-    return 0 if reply == {"ok": True} else 2
+    if set(reply) == {"ok"} and reply.get("ok") is True:
+        return 0
+    if (
+        set(reply) == {"ok", "code"}
+        and reply.get("ok") is False
+        and reply.get("code") == BENIGN_OBSERVATION_UNAVAILABLE_CODE
+    ):
+        return 0
+    return 2
 
 
 if __name__ == "__main__":