agent: add deterministic Service endpoint diagnosis

docs/agent.md

@@ -6,10 +6,13 @@ Press `Ctrl-A` to open the agent panel — a chat sidebar that answers questions
 about the cluster you are looking at.  The agent sees your current screen
 context (view, namespace, selected resource, active filter) and inspects the
 cluster through read-only tools: fetching manifests, logs, events, and resource
-listings, plus a compound `diagnose_pod` tool that gathers a broken pod's
+listings, plus compound diagnostic tools — `diagnose_pod` gathers a broken pod's
 container states, owner chain, warning events, and targeted log excerpts in a
-single deterministic call — projected evidence instead of raw YAML dumps, which
-is where small local models otherwise fail.  It can also drive the TUI itself — navigate views, apply filters,
+single deterministic call; `diagnose_service` checks whether a Service has
+current ready EndpointSlice endpoints, reporting structured versioned findings
+with explicit evidence gaps when EndpointSlice data is unavailable (e.g. RBAC
+denial) — projected evidence instead of raw YAML dumps, which is where small
+local models otherwise fail.  It can also drive the TUI itself — navigate views, apply filters,
 drill down, and open the log pane or describe screen — so "show me the crashing
 pod's logs" lands you in the actual log viewer instead of a text dump.
 Tool results are capped at 8,000 characters — manifests are shrunk
@@ -309,8 +312,8 @@ Small models rarely volunteer the screen tools (`open_describe`,
 while the TUI sits idle. Agent follow mode mirrors each successful
 cluster read from a chat turn on screen, using the same mapping as MCP
 follow mode: `list_resources` navigates the view, `get_resource` /
-`get_events` / `diagnose_pod` open the describe view, and `get_logs`
-opens the live log pane.
+`get_events` / `diagnose_pod` / `diagnose_service` open the describe view,
+and `get_logs` opens the live log pane.
 
 Follow is **on by default**. Disable it in `config.yaml`:
 

docs/mcp.md

@@ -39,6 +39,14 @@ format:
   both halves; redacting first means the value is gone before there is
   anything to split. An MCP client sees the same masked report the model
   would.
+- **Service endpoint diagnosis** (`diagnose_service`) is deterministic
+  structured YAML: one Service GET and one EndpointSlice LIST, projected
+  into versioned findings with explicit evidence gaps. EndpointSlice RBAC
+  denials are surfaced as `gaps[].source == "endpointslices"` rather than
+  as an error, so the model can reason about incomplete evidence. The result
+  is bounded to the shared 8,000-character cap and is always parseable YAML.
+  Replacement detection considers only core-v1 Service owner references;
+  custom-controller owners do not invalidate manually managed slices.
 - **Logs, events, lists, single-pod diagnoses, and helm status** get only
   their own tool-specific shaping (scoping, formatting, size caps). They
   are **not** credential-pattern masked: a token printed into a pod's log
@@ -99,6 +107,7 @@ mirrors those reads in the TUI so you can watch the assistant work:
 | `get_resource`, `get_events` | describe pane on that object |
 | `get_logs` | log pane on that pod/container |
 | `diagnose_pod` | describe pane on the pod |
+| `diagnose_service` | Service describe pane |
 | `list_operators` | navigate to subscriptions |
 | `helm_list_releases` | navigate to the helm release browser |
 

src/korvid/core/service_analysis.py

@@ -0,0 +1,325 @@
+"""Deterministic Service-to-EndpointSlice analysis contract."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import Literal
+
+Severity = Literal["warning"]
+Confidence = Literal["high", "medium"]
+Outcome = Literal["healthy", "findings", "incomplete", "not_applicable"]
+
+_ANALYZER = "service.endpoints"
+_VERSION = "1"
+_RULE_VERSION = "1"
+
+
+@dataclass(frozen=True, slots=True)
+class ResourceIdentity:
+    """Stable identity for a Kubernetes resource."""
+
+    kind: str
+    namespace: str
+    name: str
+    uid: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class Evidence:
+    """One deterministic evidence item for a report."""
+
+    resource: ResourceIdentity
+    field: str
+    value: str
+
+
+@dataclass(frozen=True, slots=True)
+class EvidenceGap:
+    """A missing or untrusted evidence source."""
+
+    source: str
+    reason: str
+
+
+@dataclass(frozen=True, slots=True)
+class Finding:
+    """A versioned rule result for a single diagnostic finding."""
+
+    rule_id: str
+    rule_version: str
+    severity: Severity
+    confidence: Confidence
+    primary: ResourceIdentity
+    related: tuple[ResourceIdentity, ...]
+    evidence: tuple[Evidence, ...]
+    explanation: str
+    next_checks: tuple[str, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class ServiceSnapshot:
+    """Immutable Service input for the analyzer."""
+
+    identity: ResourceIdentity
+    service_type: str
+    selector: tuple[tuple[str, str], ...]
+
+
+@dataclass(frozen=True, slots=True)
+class EndpointSliceSnapshot:
+    """Immutable EndpointSlice input for the analyzer."""
+
+    identity: ResourceIdentity
+    service_name: str
+    service_owner_uids: tuple[str, ...]
+    address_type: str
+    endpoints: int
+    ready_endpoints: int
+
+
+@dataclass(frozen=True, slots=True)
+class AnalysisReport:
+    """Deterministic diagnostic output for one Service."""
+
+    analyzer: str
+    version: str
+    outcome: Outcome
+    primary: ResourceIdentity
+    findings: tuple[Finding, ...] = ()
+    evidence: tuple[Evidence, ...] = ()
+    gaps: tuple[EvidenceGap, ...] = ()
+
+    def as_document(self) -> dict[str, object]:
+        """Return a stable structured document view."""
+
+        return {
+            "analyzer": self.analyzer,
+            "version": self.version,
+            "outcome": self.outcome,
+            "primary": _resource_document(self.primary),
+            "findings": [_finding_document(finding) for finding in self.findings],
+            "evidence": [_evidence_document(item) for item in self.evidence],
+            "gaps": [_gap_document(item) for item in self.gaps],
+        }
+
+
+def analyze_service_endpoints(
+    service: ServiceSnapshot,
+    slices: Sequence[EndpointSliceSnapshot],
+    gap: EvidenceGap | None = None,
+) -> AnalysisReport:
+    """Analyze a Service against EndpointSlice snapshots."""
+
+    if service.service_type == "ExternalName":
+        return _not_applicable_report(service.identity)
+    if gap is not None:
+        return _incomplete_report(service.identity, (gap,))
+
+    matching = tuple(
+        sorted(
+            (
+                item
+                for item in slices
+                if item.service_name == service.identity.name
+                and item.identity.namespace == service.identity.namespace
+            ),
+            key=_slice_sort_key,
+        )
+    )
+    current, stale = _partition_current(service.identity.uid, matching)
+    if stale:
+        return _incomplete_report(
+            service.identity,
+            (EvidenceGap("endpointslices/stale-owner", _stale_reason(service, stale)),),
+        )
+    if not current:
+        return _no_slices_report(service)
+
+    evidence = _current_evidence(current)
+    if sum(item.ready_endpoints for item in current) == 0:
+        return _no_ready_report(service, current, evidence)
+    return AnalysisReport(
+        analyzer=_ANALYZER,
+        version=_VERSION,
+        outcome="healthy",
+        primary=service.identity,
+        evidence=evidence,
+    )
+
+
+def _partition_current(
+    service_uid: str,
+    slices: Sequence[EndpointSliceSnapshot],
+) -> tuple[tuple[EndpointSliceSnapshot, ...], tuple[EndpointSliceSnapshot, ...]]:
+    current: list[EndpointSliceSnapshot] = []
+    stale: list[EndpointSliceSnapshot] = []
+    for item in slices:
+        if item.service_owner_uids and (
+            not service_uid or service_uid not in item.service_owner_uids
+        ):
+            stale.append(item)
+        else:
+            current.append(item)
+    return tuple(current), tuple(stale)
+
+
+def _slice_sort_key(item: EndpointSliceSnapshot) -> tuple[str, str, str, str]:
+    return (
+        item.identity.namespace,
+        item.identity.name,
+        item.identity.uid,
+        item.address_type,
+    )
+
+
+def _current_evidence(slices: Sequence[EndpointSliceSnapshot]) -> tuple[Evidence, ...]:
+    items: list[Evidence] = []
+    for item in slices:
+        items.append(Evidence(item.identity, "endpoints.ready", str(item.ready_endpoints)))
+        items.append(Evidence(item.identity, "endpoints.total", str(item.endpoints)))
+        items.append(Evidence(item.identity, "endpoints.address_type", item.address_type))
+        if item.service_owner_uids:
+            items.append(
+                Evidence(
+                    item.identity,
+                    "endpoints.service_owner_uids",
+                    ",".join(item.service_owner_uids),
+                )
+            )
+    return tuple(items)
+
+
+def _confidence_for_healthy(
+    service: ServiceSnapshot, slices: Sequence[EndpointSliceSnapshot]
+) -> Confidence:
+    if not service.identity.uid:
+        return "medium"
+    if any(
+        service.identity.uid in item.service_owner_uids
+        for item in slices
+        if item.service_owner_uids
+    ):
+        return "high"
+    return "medium"
+
+
+def _no_slices_report(service: ServiceSnapshot) -> AnalysisReport:
+    finding = Finding(
+        rule_id="service.no_endpoint_slices",
+        rule_version=_RULE_VERSION,
+        severity="warning",
+        confidence="medium",
+        primary=service.identity,
+        related=(),
+        evidence=(),
+        explanation="No EndpointSlices matched the Service name.",
+        next_checks=(
+            "Confirm EndpointSlice discovery is available in the namespace.",
+            "Verify the Service selector or manual slice labels match the Service name.",
+        ),
+    )
+    return AnalysisReport(
+        analyzer=_ANALYZER,
+        version=_VERSION,
+        outcome="findings",
+        primary=service.identity,
+        findings=(finding,),
+    )
+
+
+def _no_ready_report(
+    service: ServiceSnapshot,
+    slices: Sequence[EndpointSliceSnapshot],
+    evidence: tuple[Evidence, ...],
+) -> AnalysisReport:
+    finding = Finding(
+        rule_id="service.no_ready_endpoints",
+        rule_version=_RULE_VERSION,
+        severity="warning",
+        confidence=_confidence_for_healthy(service, slices),
+        primary=service.identity,
+        related=tuple(item.identity for item in slices),
+        evidence=evidence,
+        explanation="Matching EndpointSlices exist, but none report ready endpoints.",
+        next_checks=(
+            "Inspect the matching EndpointSlices for readiness and address counts.",
+            "Check the backing Pods or manual slice payload for readiness issues.",
+        ),
+    )
+    return AnalysisReport(
+        analyzer=_ANALYZER,
+        version=_VERSION,
+        outcome="findings",
+        primary=service.identity,
+        findings=(finding,),
+        evidence=evidence,
+    )
+
+
+def _incomplete_report(
+    primary: ResourceIdentity,
+    gaps: tuple[EvidenceGap, ...],
+) -> AnalysisReport:
+    return AnalysisReport(
+        analyzer=_ANALYZER,
+        version=_VERSION,
+        outcome="incomplete",
+        primary=primary,
+        gaps=gaps,
+    )
+
+
+def _not_applicable_report(primary: ResourceIdentity) -> AnalysisReport:
+    return AnalysisReport(
+        analyzer=_ANALYZER,
+        version=_VERSION,
+        outcome="not_applicable",
+        primary=primary,
+    )
+
+
+def _stale_reason(service: ServiceSnapshot, stale: Sequence[EndpointSliceSnapshot]) -> str:
+    count = len(stale)
+    verb = "is" if count == 1 else "are"
+    if service.identity.uid:
+        return (
+            f"{count} EndpointSlice{'' if count == 1 else 's'} {verb} owned by a "
+            f"different Service UID than {service.identity.uid!r}."
+        )
+    return f"{count} EndpointSlice{'' if count == 1 else 's'} {verb} owned by a Service, but the Service UID is absent."
+
+
+def _resource_document(resource: ResourceIdentity) -> dict[str, str]:
+    return {
+        "kind": resource.kind,
+        "namespace": resource.namespace,
+        "name": resource.name,
+        "uid": resource.uid,
+    }
+
+
+def _evidence_document(item: Evidence) -> dict[str, object]:
+    return {
+        "resource": _resource_document(item.resource),
+        "field": item.field,
+        "value": item.value,
+    }
+
+
+def _gap_document(item: EvidenceGap) -> dict[str, str]:
+    return {"source": item.source, "reason": item.reason}
+
+
+def _finding_document(item: Finding) -> dict[str, object]:
+    return {
+        "rule_id": item.rule_id,
+        "rule_version": item.rule_version,
+        "severity": item.severity,
+        "confidence": item.confidence,
+        "primary": _resource_document(item.primary),
+        "related": [_resource_document(resource) for resource in item.related],
+        "evidence": [_evidence_document(evidence) for evidence in item.evidence],
+        "explanation": item.explanation,
+        "next_checks": list(item.next_checks),
+    }

src/korvid/evals/fake_kube.py

@@ -114,7 +114,7 @@ def _matches(self, manifest: dict[str, Any], meta: ResourceMeta, namespace: str
 
     async def list_objects(self, meta: ResourceMeta, namespace: str | None) -> list[GenericSummary]:
         return [
-            summary_for(meta.kind, manifest)
+            summary_for(meta.kind, manifest, group=meta.group)
             for manifest in self._objects
             if self._matches(manifest, meta, namespace)
         ]

src/korvid/k8s/client.py

@@ -177,7 +177,7 @@ def _pod_summary(self, manifest: dict[str, Any]) -> PodSummary:
 
     def _object_summary(self, meta: ResourceMeta, manifest: dict[str, Any]) -> GenericSummary:
         """summary_for + configured custom column values (issue #45)."""
-        summary = summary_for(meta.kind, manifest)
+        summary = summary_for(meta.kind, manifest, group=meta.group)
         columns = self._custom_columns.get(meta.plural)
         if not columns:
             return summary