feat(server): add Stage A observability for multi-doc reuse

daser/server/__main__.py

@@ -24,6 +24,7 @@
 from daser.server.core import ServerCore
 from daser.server.doc_registry import DocRegistry
 from daser.server.http import HTTPServerConfig, VLLMClient, build_http_app
+from daser.server.http.metrics import MetricsCollector
 from daser.server.ipc import IPCServer
 from daser.server.metadata_store import MetadataStore
 
@@ -196,6 +197,11 @@ def _parse_args() -> argparse.Namespace:
         help="L1 memory-tier capacity for --transfer-mode=iouring. Defaults "
         "to min(1GiB, --l2-size).",
     )
+    parser.add_argument(
+        "--enable-metrics",
+        action="store_true",
+        help="Enable Prometheus metrics on the main HTTP server at /metrics.",
+    )
     return parser.parse_args()
 
 
@@ -481,7 +487,12 @@ async def run_server(args: argparse.Namespace) -> None:
     if cfg.transfer_mode != "gds":
         await ipc_server.initialize_transfer()
 
-    app = build_http_app(_build_http_config(args), core)
+    metrics_collector = MetricsCollector(enabled=True) if args.enable_metrics else None
+    app = build_http_app(
+        _build_http_config(args),
+        core,
+        metrics_collector=metrics_collector,
+    )
     uvicorn_config = uvicorn.Config(
         app=app,
         host=args.host,

daser/server/chunk_manager.py

@@ -57,6 +57,11 @@ def store(self) -> MetadataStore:
         """The underlying MetadataStore."""
         return self._store
 
+    @property
+    def total_slots(self) -> int:
+        """Total slot capacity of the ring buffer."""
+        return self._total_slots
+
     @property
     def free_slots(self) -> int:
         """Number of slots currently available without eviction.

daser/server/core.py

@@ -224,6 +224,7 @@ def __init__(
         self._late_evicted_commits = 0
         self._lookup_requests = 0
         self._lookup_hits = 0
+        self._eviction_counts: dict[str, int] = {}
         self._committed_chunk_keys: set[str] = set()
         self._commit_waiters: dict[str, set[asyncio.Future[None]]] = {}
 
@@ -242,12 +243,23 @@ async def rebuild_retrieval_index(self) -> None:
             await self._ri.insert(meta)
             self._committed_chunk_keys.add(meta.chunk_key)
 
-    async def lookup(self, tokens: list[int], model_id: str) -> list[ChunkInfo]:
+    async def lookup(
+        self,
+        tokens: list[int],
+        model_id: str,
+        record_access: bool = True,
+    ) -> list[ChunkInfo]:
         """Look up cached chunks for token IDs.
 
+        Hits update per-chunk access statistics by default. Read-only callers
+        such as diagnostics can pass ``record_access=False`` to avoid making an
+        inspection request look like production cache demand.
+
         Args:
             tokens: prompt token IDs.
             model_id: model identifier.
+            record_access: whether hits should update access counters and
+                aggregate lookup hit/request statistics.
 
         Returns:
             List of matching chunks, possibly empty.
@@ -256,9 +268,12 @@ async def lookup(self, tokens: list[int], model_id: str) -> list[ChunkInfo]:
             Performs no blocking I/O and should run on the server event loop.
         """
         matches = await self._ri.lookup(tokens, model_id)
-        self._lookup_requests += 1
-        if matches:
-            self._lookup_hits += 1
+        if record_access:
+            self._lookup_requests += 1
+            if matches:
+                self._lookup_hits += 1
+                for match in matches:
+                    self._cm.store.touch(match.meta.chunk_key)
         return [self._chunk_info(match) for match in matches]
 
     async def alloc_chunk(
@@ -432,6 +447,17 @@ async def commit_stats(self) -> dict[str, int]:
             "lookup_hits": self._lookup_hits,
         }
 
+    def eviction_stats(self) -> dict[str, int]:
+        """Return cumulative chunk eviction counters by reason.
+
+        Returns:
+            Mapping from eviction reason to cumulative count.
+
+        Async/thread-safety:
+            Reads in-memory counters on the server event loop.
+        """
+        return dict(self._eviction_counts)
+
     async def live_allocations(self, allocations: list[dict[str, Any]]) -> list[str]:
         """Return chunk keys that still own their allocated slot ranges.
 
@@ -499,6 +525,7 @@ async def evict_chunk(self, chunk_key: str) -> None:
         if meta is not None:
             self._mark_chunk_evicted_in_docs(meta)
             self._cm.store.remove(chunk_key)
+            self._record_eviction("explicit")
         self._committed_chunk_keys.discard(chunk_key)
         self._evicted_chunk_keys.add(chunk_key)
         logger.debug("[CORE] evict_chunk key=%s", chunk_key[:8])
@@ -636,6 +663,8 @@ async def delete_document(self, doc_id: str) -> DeleteDocumentResult:
             if self._detach_doc_from_chunk(doc_id, key):
                 await self._ri.remove(key)
                 chunks_evicted += 1
+        if chunks_evicted:
+            self._record_eviction("document_delete", chunks_evicted)
         logger.info(
             "[CORE] delete_document doc_id=%s chunks_evicted=%d",
             doc_id,
@@ -653,6 +682,7 @@ async def _drain_ring_evictions(self) -> None:
             await self._ri.remove(chunk_key)
             self._committed_chunk_keys.discard(chunk_key)
             self._evicted_chunk_keys.add(chunk_key)
+            self._record_eviction("ring")
             logger.debug("[CORE] removed auto-evicted chunk key=%s", chunk_key[:8])
 
     def _notify_commit_waiters(self, chunk_key: str) -> None:
@@ -681,6 +711,15 @@ def _mark_chunk_evicted_in_docs(self, meta: ChunkMeta) -> None:
         for doc_id in list(meta.doc_ids):
             registry.mark_chunk_evicted(doc_id, meta.chunk_key)
 
+    def _record_eviction(self, reason: str, count: int = 1) -> None:
+        """Increment the cumulative eviction counter.
+
+        Args:
+            reason: eviction reason label.
+            count: number of chunks evicted for this reason.
+        """
+        self._eviction_counts[reason] = self._eviction_counts.get(reason, 0) + count
+
     def _detach_doc_from_chunk(self, doc_id: str, chunk_key: str) -> bool:
         """Detach a document reference and remove unreferenced chunks.