feat(diagnostics): expose write protocol health and stats

CHANGELOG.md

@@ -65,6 +65,10 @@ plus bug fixes and the first real HTTP-layer test coverage.
   - Admin `logs`, `memstats`, `duckdb_memory` and
     `free_os_memory(confirm=True)`.
   - `auth.cache_stats()` and `auth.invalidate_cache()`.
+- **Write-protocol diagnostics** (`client.write_diagnostics`, sync + async):
+  inspect Line Protocol health and statistics, and retrieve MessagePack
+  protocol specifications and decoder/buffer statistics. Responses are raw
+  server-defined JSON; administrative flush remains out of scope.
 - **Edge sync administration** (`client.edge_sync`):
   - Hub: spoke registry and air-gap bundle import.
   - Spoke: status, ledger, sync, requeue/dismiss, and bundle export/ack.

README.md

@@ -375,6 +375,30 @@ async def main():
 asyncio.run(main())
 ```
 
+## Write Protocol Diagnostics
+
+Inspect the health and runtime counters of the Line Protocol and MessagePack
+writers. These read-only endpoints require an authenticated token.
+
+```python
+# Synchronous
+health = client.write_diagnostics.line_protocol_health()
+line_stats = client.write_diagnostics.line_protocol_stats()
+msgpack_spec = client.write_diagnostics.msgpack_spec()
+msgpack_stats = client.write_diagnostics.msgpack_stats()
+
+# Asynchronous equivalents
+async with AsyncArcClient(host="localhost", token="your-token") as async_client:
+    health = await async_client.write_diagnostics.line_protocol_health()
+    line_stats = await async_client.write_diagnostics.line_protocol_stats()
+    msgpack_spec = await async_client.write_diagnostics.msgpack_spec()
+    msgpack_stats = await async_client.write_diagnostics.msgpack_stats()
+```
+
+The methods return the server's JSON response unchanged. They are available on
+Arc 26.09.3 and later. The administrative Line Protocol flush endpoint
+is intentionally not included in this read-only diagnostics interface.
+
 ## Management Operations
 
 ### Retention Policies

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._write_diagnostics: Any = None
 
     def _get_http(self) -> AsyncHTTPClient:
         """Get or create the HTTP client."""
@@ -259,6 +260,19 @@ def backup(self) -> Any:
             self._backup = AsyncBackupClient(self._get_http(), self._config)
         return self._backup
 
+    @property
+    def write_diagnostics(self) -> Any:
+        """Get async read-only Line Protocol and MessagePack diagnostics."""
+        if self._write_diagnostics is None:
+            from arc_client.management.async_write_diagnostics import (
+                AsyncWriteDiagnosticsClient,
+            )
+
+            self._write_diagnostics = AsyncWriteDiagnosticsClient(
+                self._get_http(), self._config
+            )
+        return self._write_diagnostics
+
     @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._write_diagnostics: 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 write_diagnostics(self) -> Any:
+        """Get read-only Line Protocol and MessagePack diagnostics."""
+        if self._write_diagnostics is None:
+            from arc_client.management.write_diagnostics import WriteDiagnosticsClient
+
+            self._write_diagnostics = WriteDiagnosticsClient(self._get_http(), self._config)
+        return self._write_diagnostics
+
     @property
     def mqtt(self) -> Any:
         """Get the MQTT ingestion management client (enterprise)."""

src/arc_client/management/__init__.py

@@ -2,7 +2,8 @@
 
 Retention, continuous queries, delete, databases, and the enterprise surface
 (RBAC, governance, audit, query management, cluster, compaction, tiering,
-backup, MQTT) and operations (reconciliation, diagnostics, edge sync).
+backup, MQTT, write diagnostics) and operations (reconciliation, diagnostics,
+edge sync).
 """
 
 from arc_client.management.async_audit import AsyncAuditClient
@@ -21,6 +22,7 @@
 from arc_client.management.async_reconciliation import AsyncReconciliationClient
 from arc_client.management.async_retention import AsyncRetentionClient
 from arc_client.management.async_tiering import AsyncTieringClient
+from arc_client.management.async_write_diagnostics import AsyncWriteDiagnosticsClient
 from arc_client.management.audit import AuditClient
 from arc_client.management.backup import BackupClient
 from arc_client.management.cluster import ClusterClient
@@ -37,6 +39,7 @@
 from arc_client.management.reconciliation import ReconciliationClient
 from arc_client.management.retention import RetentionClient
 from arc_client.management.tiering import TieringClient
+from arc_client.management.write_diagnostics import WriteDiagnosticsClient
 
 __all__ = [
     # Core management
@@ -46,6 +49,8 @@
     "AsyncDeleteClient",
     "RetentionClient",
     "AsyncRetentionClient",
+    "WriteDiagnosticsClient",
+    "AsyncWriteDiagnosticsClient",
     "DatabaseClient",
     "AsyncDatabaseClient",
     # Enterprise

src/arc_client/management/async_write_diagnostics.py

@@ -0,0 +1,54 @@
+"""Asynchronous diagnostics for Arc write protocols.
+
+Wraps the read-only Line Protocol health/statistics and MessagePack
+specification/statistics endpoints. All endpoints require authentication.
+Responses are returned as raw JSON values because Arc does not promise a
+stable schema for these operator diagnostics.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from arc_client.config import ClientConfig
+from arc_client.exceptions import ArcError
+from arc_client.http.async_http import AsyncHTTPClient
+
+
+class AsyncWriteDiagnosticsClient:
+    """Inspect writer health and statistics; return raw, server-defined JSON."""
+
+    def __init__(self, http: AsyncHTTPClient, config: ClientConfig) -> None:
+        self._http = http
+        self._config = config
+
+    async def _json(self, path: str) -> Any:
+        try:
+            response = await self._http.get(path)
+            return response.json()
+        except ArcError:
+            raise
+        except Exception as e:
+            raise ArcError(f"Write diagnostics request failed (GET {path}): {e}") from e
+
+    async def line_protocol_health(self) -> Any:
+        """Return raw JSON health status for the Line Protocol writer."""
+        return await self._json("/api/v1/write/line-protocol/health")
+
+    async def line_protocol_stats(self) -> Any:
+        """Return raw JSON request, record, byte, and error counters for Line Protocol."""
+        return await self._json("/api/v1/write/line-protocol/stats")
+
+    async def msgpack_spec(self) -> Any:
+        """Return the raw JSON MessagePack write-protocol specification.
+
+        Arc's current response contains stale authentication metadata: it says
+        authentication is not implemented and lists only ``x-api-key``. The
+        endpoint does require authentication, so do not rely on that section
+        until the server-side specification is corrected.
+        """
+        return await self._json("/api/v1/write/msgpack/spec")
+
+    async def msgpack_stats(self) -> Any:
+        """Return raw JSON MessagePack decoder and Arrow buffer statistics."""
+        return await self._json("/api/v1/write/msgpack/stats")

src/arc_client/management/write_diagnostics.py

@@ -0,0 +1,54 @@
+"""Synchronous diagnostics for Arc write protocols.
+
+Wraps the read-only Line Protocol health/statistics and MessagePack
+specification/statistics endpoints. All endpoints require authentication.
+Responses are returned as raw JSON values because Arc does not promise a
+stable schema for these operator diagnostics.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from arc_client.config import ClientConfig
+from arc_client.exceptions import ArcError
+from arc_client.http.sync_http import SyncHTTPClient
+
+
+class WriteDiagnosticsClient:
+    """Inspect writer health and statistics; return raw, server-defined JSON."""
+
+    def __init__(self, http: SyncHTTPClient, config: ClientConfig) -> None:
+        self._http = http
+        self._config = config
+
+    def _json(self, path: str) -> Any:
+        try:
+            response = self._http.get(path)
+            return response.json()
+        except ArcError:
+            raise
+        except Exception as e:
+            raise ArcError(f"Write diagnostics request failed (GET {path}): {e}") from e
+
+    def line_protocol_health(self) -> Any:
+        """Return raw JSON health status for the Line Protocol writer."""
+        return self._json("/api/v1/write/line-protocol/health")
+
+    def line_protocol_stats(self) -> Any:
+        """Return raw JSON request, record, byte, and error counters for Line Protocol."""
+        return self._json("/api/v1/write/line-protocol/stats")
+
+    def msgpack_spec(self) -> Any:
+        """Return the raw JSON MessagePack write-protocol specification.
+
+        Arc's current response contains stale authentication metadata: it says
+        authentication is not implemented and lists only ``x-api-key``. The
+        endpoint does require authentication, so do not rely on that section
+        until the server-side specification is corrected.
+        """
+        return self._json("/api/v1/write/msgpack/spec")
+
+    def msgpack_stats(self) -> Any:
+        """Return raw JSON MessagePack decoder and Arrow buffer statistics."""
+        return self._json("/api/v1/write/msgpack/stats")

tests/unit/http/test_write_diagnostics.py

@@ -0,0 +1,87 @@
+"""HTTP contract tests for Line Protocol and MessagePack diagnostics."""
+
+from __future__ import annotations
+
+import pytest
+from pytest_httpx import HTTPXMock
+
+from arc_client import ArcClient, AsyncArcClient
+from arc_client.exceptions import ArcPermissionError, ArcServerError
+
+
+@pytest.mark.parametrize("async_mode", [False, True])
+async def test_write_protocol_diagnostics(httpx_mock: HTTPXMock, async_mode: bool) -> None:
+    expected = [
+        {"status": "healthy", "service": "line_protocol_writer"},
+        {
+            "status": "success",
+            "stats": {
+                "total_requests": 12,
+                "total_records": 34,
+                "total_bytes": 1024,
+                "total_bytes_gzipped": 512,
+                "total_errors": 1,
+            },
+        },
+        {"version": "2.0", "protocol": "MessagePack"},
+        {"decoder": {"records": 5}, "buffer": {"buffers": 2}},
+    ]
+    paths = [
+        "/api/v1/write/line-protocol/health",
+        "/api/v1/write/line-protocol/stats",
+        "/api/v1/write/msgpack/spec",
+        "/api/v1/write/msgpack/stats",
+    ]
+    for payload in expected:
+        httpx_mock.add_response(json=payload)
+
+    if async_mode:
+        async with AsyncArcClient(host="h", token="t", max_retries=0) as client:
+            diagnostics = client.write_diagnostics
+            results = [
+                await diagnostics.line_protocol_health(),
+                await diagnostics.line_protocol_stats(),
+                await diagnostics.msgpack_spec(),
+                await diagnostics.msgpack_stats(),
+            ]
+    else:
+        with ArcClient(host="h", token="t", max_retries=0) as client:
+            diagnostics = client.write_diagnostics
+            results = [
+                diagnostics.line_protocol_health(),
+                diagnostics.line_protocol_stats(),
+                diagnostics.msgpack_spec(),
+                diagnostics.msgpack_stats(),
+            ]
+
+    assert results == expected
+    requests = httpx_mock.get_requests()
+    assert [(request.method, request.url.path) for request in requests] == [
+        ("GET", path) for path in paths
+    ]
+    assert all(request.headers["Authorization"] == "Bearer t" for request in requests)
+
+
+@pytest.mark.parametrize("async_mode", [False, True])
+async def test_write_diagnostics_preserve_permission_and_server_errors(
+    httpx_mock: HTTPXMock, async_mode: bool
+) -> None:
+    httpx_mock.add_response(status_code=403, json={"error": "Authentication required"})
+    httpx_mock.add_response(status_code=503, json={"error": "Writer unavailable"})
+
+    if async_mode:
+        async with AsyncArcClient(host="h", token="t", max_retries=0) as client:
+            with pytest.raises(ArcPermissionError) as permission_error:
+                await client.write_diagnostics.line_protocol_health()
+            with pytest.raises(ArcServerError) as error:
+                await client.write_diagnostics.msgpack_stats()
+    else:
+        with ArcClient(host="h", token="t", max_retries=0) as client:
+            with pytest.raises(ArcPermissionError) as permission_error:
+                client.write_diagnostics.line_protocol_health()
+            with pytest.raises(ArcServerError) as error:
+                client.write_diagnostics.msgpack_stats()
+
+    assert permission_error.value.response_body == {"error": "Authentication required"}
+    assert error.value.status_code == 503
+    assert error.value.response_body == {"error": "Writer unavailable"}