feat(schedulers): expose scheduler status and controls

CHANGELOG.md

@@ -79,6 +79,16 @@ plus bug fixes and the first real HTTP-layer test coverage.
   `ARC_CLIENT_SEND_INSTALLATION_ID=0` or `send_installation_id=False`; pin the
   ID in containers with `ARC_CLIENT_INSTALLATION_ID`. See README, "Privacy".
 
+- **Scheduler management** (`client.schedulers`, sync + async): inspect CQ and retention
+  scheduler status, reload active continuous-query schedules, and manually trigger
+  retention. The status route accepts authenticated tokens; control routes preserve
+  Arc administrator and enterprise-license gates. Retention triggers require
+  `confirm=True` because they may permanently delete data. Supported by Arc 26.09.3 and later.
+  On Arc 27.01.1 and later, a control refused because the license is missing, expired
+  or of a tier without the scheduler raises `ArcLicenseError` (HTTP 403); a scheduler
+  off in config is `ArcServerError` with `status_code == 503`, and a retention trigger
+  on a node that is not the primary writer, or while a cycle runs, is `ArcServerError`
+  with `status_code == 409`.
 - Arrow completion metadata in the existing `pyarrow.Table` schema for sync and
   async queries: policy cap and execution time are retained; failed, malformed,
   conflicting or unobservable announced completion metadata is rejected.

README.md

@@ -64,7 +64,7 @@ with ArcClient(host="localhost", token="your-token") as client:
 - **Databases**: List, create, delete databases and list measurements
 - **Management**: Retention policies, continuous queries, delete operations
 - **Enterprise**: RBAC, query governance, audit logs, query management, cluster,
-  compaction, storage tiering, backup/restore, and MQTT ingestion
+  compaction, scheduler controls, storage tiering, backup/restore, and MQTT ingestion
 - **Authentication**: Token management (create, rotate, revoke)
 
 ## Data Ingestion
@@ -401,6 +401,28 @@ print(f"Would delete {result.deleted_count} rows")
 result = client.retention.execute(policy.id, dry_run=False, confirm=True)
 ```
 
+### Scheduler inspection and controls
+
+Scheduler status is available to authenticated tokens. Reloading continuous-query
+schedules and manually triggering retention require an administrator token and
+the corresponding enterprise scheduler feature.
+
+```python
+status = client.schedulers.status()
+client.schedulers.reload_continuous_queries()
+
+# Runs all active retention policies immediately and may permanently delete data.
+client.schedulers.trigger_retention(confirm=True)
+```
+
+The retention operation refuses to send a request unless `confirm=True` is
+supplied. The async client exposes the same methods:
+
+```python
+async with AsyncArcClient(host="localhost", token="your-token") as client:
+    status = await client.schedulers.status()
+```
+
 ### Continuous Queries
 
 Aggregate and downsample data automatically:
@@ -513,6 +535,11 @@ client.compaction.trigger(database="metrics", measurement="cpu")
 client.compaction.cycles(limit=5)
 client.compaction.too_wide()
 
+# Scheduler inspection and explicit operator controls
+client.schedulers.status()  # requires an authenticated token
+client.schedulers.reload_continuous_queries()  # administrator + licensed feature
+# Retention may permanently delete data; explicit confirmation is required.
+client.schedulers.trigger_retention(confirm=True)
 client.tiering.set_policy("metrics", hot_only=True, hot_max_age_days=7)
 
 client.backup.create(databases=["metrics"])            # scoped backup

src/arc_client/async_client.py

@@ -101,6 +101,7 @@ def __init__(
         self._edge_sync: Any = None
         self._diagnostics: Any = None
         self._reconciliation: Any = None
+        self._schedulers: Any = None
 
     def _get_http(self) -> AsyncHTTPClient:
         """Get or create the HTTP client."""
@@ -259,6 +260,15 @@ def backup(self) -> Any:
             self._backup = AsyncBackupClient(self._get_http(), self._config)
         return self._backup
 
+    @property
+    def schedulers(self) -> Any:
+        """Get the async scheduler inspection and control client."""
+        if self._schedulers is None:
+            from arc_client.management.async_schedulers import AsyncSchedulerClient
+
+            self._schedulers = AsyncSchedulerClient(self._get_http(), self._config)
+        return self._schedulers
+
     @property
     def mqtt(self) -> Any:
         """Get the async MQTT ingestion management client (enterprise)."""

src/arc_client/client.py

@@ -101,6 +101,7 @@ def __init__(
         self._edge_sync: Any = None
         self._diagnostics: Any = None
         self._reconciliation: Any = None
+        self._schedulers: Any = None
 
     def _get_http(self) -> SyncHTTPClient:
         """Get or create the HTTP client."""
@@ -257,6 +258,15 @@ def backup(self) -> Any:
             self._backup = BackupClient(self._get_http(), self._config)
         return self._backup
 
+    @property
+    def schedulers(self) -> Any:
+        """Get the scheduler inspection and control client."""
+        if self._schedulers is None:
+            from arc_client.management.schedulers import SchedulerClient
+
+            self._schedulers = SchedulerClient(self._get_http(), self._config)
+        return self._schedulers
+
     @property
     def mqtt(self) -> Any:
         """Get the MQTT ingestion management client (enterprise)."""

src/arc_client/management/__init__.py

@@ -20,6 +20,7 @@
 from arc_client.management.async_rbac import AsyncRBACClient
 from arc_client.management.async_reconciliation import AsyncReconciliationClient
 from arc_client.management.async_retention import AsyncRetentionClient
+from arc_client.management.async_schedulers import AsyncSchedulerClient
 from arc_client.management.async_tiering import AsyncTieringClient
 from arc_client.management.audit import AuditClient
 from arc_client.management.backup import BackupClient
@@ -36,6 +37,7 @@
 from arc_client.management.rbac import RBACClient
 from arc_client.management.reconciliation import ReconciliationClient
 from arc_client.management.retention import RetentionClient
+from arc_client.management.schedulers import SchedulerClient
 from arc_client.management.tiering import TieringClient
 
 __all__ = [
@@ -46,6 +48,8 @@
     "AsyncDeleteClient",
     "RetentionClient",
     "AsyncRetentionClient",
+    "SchedulerClient",
+    "AsyncSchedulerClient",
     "DatabaseClient",
     "AsyncDatabaseClient",
     # Enterprise

src/arc_client/management/async_schedulers.py

@@ -0,0 +1,104 @@
+"""Asynchronous scheduler inspection and control for Arc.
+
+Wraps ``/api/v1/schedulers``. Status is available to authenticated tokens;
+control operations require an administrator token and the corresponding
+enterprise scheduler feature. Since Arc 27.01.1 a missing, expired or
+insufficient license is HTTP 403, raised as ``ArcLicenseError``; HTTP 503
+means the scheduler is off in config or failed to start, and ``status()``
+gives the reason. Earlier Arc versions answer 503 in both cases.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from arc_client.config import ClientConfig
+from arc_client.exceptions import (
+    ArcError,
+    ArcLicenseError,
+    ArcPermissionError,
+    ArcValidationError,
+)
+from arc_client.http.async_http import AsyncHTTPClient
+from arc_client.management.delete import DELETE_TIMEOUT_FLOOR
+
+
+class AsyncSchedulerClient:
+    """Inspect scheduler status and perform explicit scheduler controls.
+
+    Methods return raw JSON values because scheduler status and counters are
+    server-defined operator diagnostics, not a stable typed response schema.
+    """
+
+    def __init__(self, http: AsyncHTTPClient, config: ClientConfig) -> None:
+        self._http = http
+        self._config = config
+
+    async def _json(self, method: str, path: str, **kwargs: Any) -> Any:
+        try:
+            response = await getattr(self._http, method)(path, **kwargs)
+            return response.json()
+        except ArcPermissionError as e:
+            # Arc refuses a scheduler control with 403 when the license is
+            # missing, expired or of a tier without the scheduler. Only the
+            # "Enterprise license required" form matches the generic 403
+            # license detection; the tier and refused-reload forms name the
+            # license without that phrase.
+            message = str(e)
+            if "license" in message.lower():
+                detail = message.split("Permission denied: ", 1)[-1]
+                raise ArcLicenseError(f"Enterprise license required: {detail}") from e
+            raise
+        except ArcError:
+            raise
+        except Exception as e:
+            raise ArcError(f"Scheduler request failed ({method.upper()} {path}): {e}") from e
+
+    async def status(self) -> Any:
+        """Return the status of the CQ and retention schedulers.
+
+        Available with any authenticated token. Returns raw JSON with each
+        scheduler's enabled state and, when unavailable, its reason.
+        """
+        return await self._json("get", "/api/v1/schedulers/")
+
+    async def reload_continuous_queries(self) -> Any:
+        """Reload active CQ schedules (administrator and licensed feature required).
+
+        Returns the raw JSON response from Arc. A license that is missing,
+        expired or does not include the scheduler raises ``ArcLicenseError``
+        (HTTP 403). HTTP 503 means the scheduler is off in config or failed to
+        start; call ``status()`` for the reason.
+        """
+        return await self._json("post", "/api/v1/schedulers/cq/reload")
+
+    async def trigger_retention(self, confirm: bool = False) -> Any:
+        """Run all active retention policies immediately (administrator only).
+
+        This can delete data according to the configured retention policies.
+        Pass ``confirm=True`` to explicitly authorize the immediate run. The
+        request allows at least 90 seconds because Arc runs every active policy
+        synchronously. In a cluster, send it to the primary writer: another node
+        refuses it, as Arc does while a retention cycle is already running.
+        Since Arc 27.01.1 both refusals are HTTP 409, raised as
+        ``ArcServerError`` with ``status_code == 409``; earlier versions
+        answer HTTP 500.
+
+        Returns the raw JSON response from Arc. A license that is missing,
+        expired or does not include the scheduler raises ``ArcLicenseError``
+        (HTTP 403). HTTP 503 means the scheduler is off in config or failed to
+        start; call ``status()`` for the reason.
+
+        Raises:
+            ArcValidationError: If ``confirm`` is not True.
+        """
+        if not confirm:
+            raise ArcValidationError(
+                "Refusing to trigger retention without confirm=True. "
+                "This may permanently delete data."
+            )
+        return await self._json(
+            "post",
+            "/api/v1/schedulers/retention/trigger",
+            timeout=max(self._config.timeout, DELETE_TIMEOUT_FLOOR),
+        )

src/arc_client/management/schedulers.py

@@ -0,0 +1,104 @@
+"""Synchronous scheduler inspection and control for Arc.
+
+Wraps ``/api/v1/schedulers``. Status is available to authenticated tokens;
+control operations require an administrator token and the corresponding
+enterprise scheduler feature. Since Arc 27.01.1 a missing, expired or
+insufficient license is HTTP 403, raised as ``ArcLicenseError``; HTTP 503
+means the scheduler is off in config or failed to start, and ``status()``
+gives the reason. Earlier Arc versions answer 503 in both cases.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from arc_client.config import ClientConfig
+from arc_client.exceptions import (
+    ArcError,
+    ArcLicenseError,
+    ArcPermissionError,
+    ArcValidationError,
+)
+from arc_client.http.sync_http import SyncHTTPClient
+from arc_client.management.delete import DELETE_TIMEOUT_FLOOR
+
+
+class SchedulerClient:
+    """Inspect scheduler status and perform explicit scheduler controls.
+
+    Methods return raw JSON values because scheduler status and counters are
+    server-defined operator diagnostics, not a stable typed response schema.
+    """
+
+    def __init__(self, http: SyncHTTPClient, config: ClientConfig) -> None:
+        self._http = http
+        self._config = config
+
+    def _json(self, method: str, path: str, **kwargs: Any) -> Any:
+        try:
+            response = getattr(self._http, method)(path, **kwargs)
+            return response.json()
+        except ArcPermissionError as e:
+            # Arc refuses a scheduler control with 403 when the license is
+            # missing, expired or of a tier without the scheduler. Only the
+            # "Enterprise license required" form matches the generic 403
+            # license detection; the tier and refused-reload forms name the
+            # license without that phrase.
+            message = str(e)
+            if "license" in message.lower():
+                detail = message.split("Permission denied: ", 1)[-1]
+                raise ArcLicenseError(f"Enterprise license required: {detail}") from e
+            raise
+        except ArcError:
+            raise
+        except Exception as e:
+            raise ArcError(f"Scheduler request failed ({method.upper()} {path}): {e}") from e
+
+    def status(self) -> Any:
+        """Return the status of the CQ and retention schedulers.
+
+        Available with any authenticated token. Returns raw JSON with each
+        scheduler's enabled state and, when unavailable, its reason.
+        """
+        return self._json("get", "/api/v1/schedulers/")
+
+    def reload_continuous_queries(self) -> Any:
+        """Reload active CQ schedules (administrator and licensed feature required).
+
+        Returns the raw JSON response from Arc. A license that is missing,
+        expired or does not include the scheduler raises ``ArcLicenseError``
+        (HTTP 403). HTTP 503 means the scheduler is off in config or failed to
+        start; call ``status()`` for the reason.
+        """
+        return self._json("post", "/api/v1/schedulers/cq/reload")
+
+    def trigger_retention(self, confirm: bool = False) -> Any:
+        """Run all active retention policies immediately (administrator only).
+
+        This can delete data according to the configured retention policies.
+        Pass ``confirm=True`` to explicitly authorize the immediate run. The
+        request allows at least 90 seconds because Arc runs every active policy
+        synchronously. In a cluster, send it to the primary writer: another node
+        refuses it, as Arc does while a retention cycle is already running.
+        Since Arc 27.01.1 both refusals are HTTP 409, raised as
+        ``ArcServerError`` with ``status_code == 409``; earlier versions
+        answer HTTP 500.
+
+        Returns the raw JSON response from Arc. A license that is missing,
+        expired or does not include the scheduler raises ``ArcLicenseError``
+        (HTTP 403). HTTP 503 means the scheduler is off in config or failed to
+        start; call ``status()`` for the reason.
+
+        Raises:
+            ArcValidationError: If ``confirm`` is not True.
+        """
+        if not confirm:
+            raise ArcValidationError(
+                "Refusing to trigger retention without confirm=True. "
+                "This may permanently delete data."
+            )
+        return self._json(
+            "post",
+            "/api/v1/schedulers/retention/trigger",
+            timeout=max(self._config.timeout, DELETE_TIMEOUT_FLOOR),
+        )