fix(management): align retention and continuous-query contracts

CHANGELOG.md

@@ -66,6 +66,9 @@ plus bug fixes and the first real HTTP-layer test coverage.
 
 ### Fixed
 
+- Retention and continuous-query clients accept Arc's list/null/flat response
+  shapes. Sparse updates remain sparse; explicit full replacements support
+  released servers, and CLEAR requests explicit null values on newer servers.
 - Buffered close waits for active writes, retains unacknowledged batches and
   surfaces failures without replay. Typed errors retain structured diagnostics.
 

src/arc_client/__init__.py

@@ -5,6 +5,7 @@
 # failure during package initialization.
 __version__ = "0.2.0"
 
+from arc_client._sentinels import CLEAR
 from arc_client.async_client import AsyncArcClient
 from arc_client.client import ArcClient
 from arc_client.compression import Compression
@@ -27,6 +28,7 @@
 )
 
 __all__ = [
+    "CLEAR",
     # Clients
     "ArcClient",
     "AsyncArcClient",

src/arc_client/_sentinels.py

@@ -0,0 +1,12 @@
+"""Explicit clearing for nullable fields in partial updates."""
+
+from enum import Enum
+
+
+class ClearType(Enum):
+    """Marker for an explicit JSON null, distinct from an omitted argument."""
+
+    CLEAR = "CLEAR"
+
+
+CLEAR = ClearType.CLEAR

src/arc_client/management/async_continuous_query.py

@@ -5,8 +5,9 @@
 from datetime import datetime
 from typing import Any, List, Optional, Union
 
+from arc_client._sentinels import CLEAR, ClearType
 from arc_client.config import ClientConfig
-from arc_client.exceptions import ArcError, ArcNotFoundError
+from arc_client.exceptions import ArcError, ArcNotFoundError, ArcValidationError
 from arc_client.http.async_http import AsyncHTTPClient
 from arc_client.models.continuous_query import (
     ContinuousQuery,
@@ -34,6 +35,8 @@ async def create(
         retention_days: Optional[int] = None,
         delete_source_after_days: Optional[int] = None,
         is_active: bool = True,
+        *,
+        tag_columns: Optional[List[str]] = None,
     ) -> ContinuousQuery:
         """Create a new continuous query."""
         payload = {
@@ -47,13 +50,14 @@ async def create(
             "retention_days": retention_days,
             "delete_source_after_days": delete_source_after_days,
             "is_active": is_active,
+            "tag_columns": tag_columns,
         }
         try:
             response = await self._http.post("/api/v1/continuous_queries", json=payload)
             data = response.json()
             if not data.get("success", True):
                 raise ArcError(data.get("error", "Failed to create CQ"))
-            return ContinuousQuery(**data.get("query", data))
+            return ContinuousQuery.from_response(data)
         except ArcError:
             raise
         except Exception as e:
@@ -72,8 +76,21 @@ async def list(
 
             response = await self._http.get("/api/v1/continuous_queries", params=params)
             data = response.json()
-            queries = data.get("queries", [])
-            return [ContinuousQuery(**q) for q in queries]
+            if data is None:
+                queries = []
+            elif isinstance(data, list):
+                queries = data
+            elif isinstance(data, dict) and "queries" in data:
+                queries = data["queries"]
+                if queries is None:
+                    queries = []
+            else:
+                raise ArcError("Malformed continuous query list: expected an array")
+            if not isinstance(queries, list):
+                raise ArcError("Malformed continuous query list: expected an array")
+            return [ContinuousQuery.from_response(q) for q in queries]
+        except ArcError:
+            raise
         except Exception as e:
             raise ArcError(f"Failed to list continuous queries: {e}") from e
 
@@ -87,7 +104,7 @@ async def get(self, query_id: int) -> ContinuousQuery:
                 if "not found" in error.lower():
                     raise ArcNotFoundError(f"CQ {query_id} not found")
                 raise ArcError(error)
-            return ContinuousQuery(**data.get("query", data))
+            return ContinuousQuery.from_response(data)
         except (ArcNotFoundError, ArcError):
             raise
         except Exception as e:
@@ -97,30 +114,85 @@ async def update(
         self,
         query_id: int,
         name: Optional[str] = None,
-        description: Optional[str] = None,
+        description: Union[str, ClearType, None] = None,
         query: Optional[str] = None,
         interval: Optional[str] = None,
-        retention_days: Optional[int] = None,
-        delete_source_after_days: Optional[int] = None,
+        retention_days: Union[int, ClearType, None] = None,
+        delete_source_after_days: Union[int, ClearType, None] = None,
         is_active: Optional[bool] = None,
+        *,
+        database: Optional[str] = None,
+        source_measurement: Optional[str] = None,
+        destination_measurement: Optional[str] = None,
+        tag_columns: Union[List[str], ClearType, None] = None,
     ) -> ContinuousQuery:
-        """Update a continuous query."""
+        """Send a partial update without reading or replacing unrelated fields.
+
+        None omits an argument, preserving existing caller behavior. CLEAR
+        sends JSON null for nullable fields. Servers with full-replacement
+        PUT semantics (including Arc 26.09.3) require replace() instead.
+        """
+        values = {
+            "name": name,
+            "description": description,
+            "query": query,
+            "interval": interval,
+            "retention_days": retention_days,
+            "delete_source_after_days": delete_source_after_days,
+            "is_active": is_active,
+            "database": database,
+            "source_measurement": source_measurement,
+            "destination_measurement": destination_measurement,
+            "tag_columns": tag_columns,
+        }
+        nullable = {"description", "tag_columns", "retention_days", "delete_source_after_days"}
         payload: dict[str, Any] = {}
-        if name is not None:
-            payload["name"] = name
-        if description is not None:
-            payload["description"] = description
-        if query is not None:
-            payload["query"] = query
-        if interval is not None:
-            payload["interval"] = interval
-        if retention_days is not None:
-            payload["retention_days"] = retention_days
-        if delete_source_after_days is not None:
-            payload["delete_source_after_days"] = delete_source_after_days
-        if is_active is not None:
-            payload["is_active"] = is_active
+        for key, value in values.items():
+            if value is CLEAR:
+                if key not in nullable:
+                    raise ArcValidationError(f"{key} may not be cleared")
+                payload[key] = None
+            elif value is not None:
+                payload[key] = value
+        return await self._put_definition(query_id, payload)
+
+    async def replace(
+        self,
+        query_id: int,
+        *,
+        name: str,
+        database: str,
+        source_measurement: str,
+        destination_measurement: str,
+        query: str,
+        interval: str,
+        description: Optional[str],
+        tag_columns: Optional[List[str]],
+        retention_days: Optional[int],
+        delete_source_after_days: Optional[int],
+        is_active: bool,
+    ) -> ContinuousQuery:
+        """Replace all mutable fields with one PUT on old or new servers.
+
+        Every field must be explicit. This operation intentionally overwrites
+        the definition; it does not fetch or merge a potentially stale copy.
+        """
+        payload = {
+            "name": name,
+            "database": database,
+            "source_measurement": source_measurement,
+            "destination_measurement": destination_measurement,
+            "query": query,
+            "interval": interval,
+            "description": description,
+            "tag_columns": tag_columns,
+            "retention_days": retention_days,
+            "delete_source_after_days": delete_source_after_days,
+            "is_active": is_active,
+        }
+        return await self._put_definition(query_id, payload)
 
+    async def _put_definition(self, query_id: int, payload: dict[str, Any]) -> ContinuousQuery:
         try:
             response = await self._http.put(f"/api/v1/continuous_queries/{query_id}", json=payload)
             data = response.json()
@@ -129,8 +201,8 @@ async def update(
                 if "not found" in error.lower():
                     raise ArcNotFoundError(f"CQ {query_id} not found")
                 raise ArcError(error)
-            return ContinuousQuery(**data.get("query", data))
-        except (ArcNotFoundError, ArcError):
+            return ContinuousQuery.from_response(data)
+        except ArcError:
             raise
         except Exception as e:
             raise ArcError(f"Failed to update continuous query: {e}") from e
@@ -186,6 +258,12 @@ async def get_executions(self, query_id: int, limit: int = 50) -> List[CQExecuti
             )
             data = response.json()
             executions = data.get("executions", [])
+            if executions is None:
+                executions = []
+            if not isinstance(executions, list):
+                raise ArcError("Malformed CQ history: expected an array")
             return [CQExecution(**e) for e in executions]
+        except ArcError:
+            raise
         except Exception as e:
             raise ArcError(f"Failed to get CQ executions: {e}") from e

src/arc_client/management/async_retention.py

@@ -2,10 +2,11 @@
 
 from __future__ import annotations
 
-from typing import Any, List, Optional
+from typing import Any, List, Optional, Union
 
+from arc_client._sentinels import CLEAR, ClearType
 from arc_client.config import ClientConfig
-from arc_client.exceptions import ArcError, ArcNotFoundError
+from arc_client.exceptions import ArcError, ArcNotFoundError, ArcValidationError
 from arc_client.http.async_http import AsyncHTTPClient
 from arc_client.models.retention import (
     ExecuteRetentionResponse,
@@ -55,8 +56,14 @@ async def list(self) -> List[RetentionPolicy]:
         try:
             response = await self._http.get("/api/v1/retention")
             data = response.json()
-            policies = data.get("policies", [])
+            policies = data.get("policies", data) if isinstance(data, dict) else data
+            if policies is None:
+                policies = []
+            if not isinstance(policies, list):
+                raise ArcError("Malformed retention policy list: expected an array")
             return [RetentionPolicy(**p) for p in policies]
+        except ArcError:
+            raise
         except Exception as e:
             raise ArcError(f"Failed to list retention policies: {e}") from e
 
@@ -81,23 +88,63 @@ async def update(
         policy_id: int,
         name: Optional[str] = None,
         retention_days: Optional[int] = None,
-        measurement: Optional[str] = None,
+        measurement: Union[str, ClearType, None] = None,
         buffer_days: Optional[int] = None,
         is_active: Optional[bool] = None,
+        *,
+        database: Optional[str] = None,
     ) -> RetentionPolicy:
-        """Update a retention policy."""
+        """Send a partial update without reading unrelated fields.
+
+        None omits an argument, preserving existing caller behavior. CLEAR
+        sends JSON null for measurement. Servers with full-replacement PUT
+        semantics (including Arc 26.09.3) require replace() instead.
+        """
+        values = {
+            "name": name,
+            "retention_days": retention_days,
+            "measurement": measurement,
+            "buffer_days": buffer_days,
+            "is_active": is_active,
+            "database": database,
+        }
         payload: dict[str, Any] = {}
-        if name is not None:
-            payload["name"] = name
-        if retention_days is not None:
-            payload["retention_days"] = retention_days
-        if measurement is not None:
-            payload["measurement"] = measurement
-        if buffer_days is not None:
-            payload["buffer_days"] = buffer_days
-        if is_active is not None:
-            payload["is_active"] = is_active
+        for key, value in values.items():
+            if value is CLEAR:
+                if key != "measurement":
+                    raise ArcValidationError(f"{key} may not be cleared")
+                payload[key] = None
+            elif value is not None:
+                payload[key] = value
+        return await self._put_definition(policy_id, payload)
 
+    async def replace(
+        self,
+        policy_id: int,
+        *,
+        name: str,
+        database: str,
+        retention_days: int,
+        measurement: Optional[str],
+        buffer_days: int,
+        is_active: bool,
+    ) -> RetentionPolicy:
+        """Replace all mutable fields with one PUT on old or new servers.
+
+        Every field must be explicit. This operation intentionally overwrites
+        the definition; it does not fetch or merge a potentially stale copy.
+        """
+        payload = {
+            "name": name,
+            "database": database,
+            "retention_days": retention_days,
+            "measurement": measurement,
+            "buffer_days": buffer_days,
+            "is_active": is_active,
+        }
+        return await self._put_definition(policy_id, payload)
+
+    async def _put_definition(self, policy_id: int, payload: dict[str, Any]) -> RetentionPolicy:
         try:
             response = await self._http.put(f"/api/v1/retention/{policy_id}", json=payload)
             data = response.json()
@@ -136,6 +183,8 @@ async def execute(
             response = await self._http.post(f"/api/v1/retention/{policy_id}/execute", json=payload)
             data = response.json()
             return ExecuteRetentionResponse(**data)
+        except ArcError:
+            raise
         except Exception as e:
             raise ArcError(f"Failed to execute retention policy: {e}") from e
 
@@ -148,6 +197,12 @@ async def get_executions(self, policy_id: int, limit: int = 50) -> List[Retentio
             )
             data = response.json()
             executions = data.get("executions", [])
+            if executions is None:
+                executions = []
+            if not isinstance(executions, list):
+                raise ArcError("Malformed retention execution history: expected an array")
             return [RetentionExecution(**e) for e in executions]
+        except ArcError:
+            raise
         except Exception as e:
             raise ArcError(f"Failed to get retention executions: {e}") from e