diff --git a/docs/agent.md b/docs/agent.md index 9016a810..198eb9f0 100644 --- a/docs/agent.md +++ b/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`: diff --git a/docs/mcp.md b/docs/mcp.md index 8e034a8c..889c74b9 100644 --- a/docs/mcp.md +++ b/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 | diff --git a/src/korvid/core/service_analysis.py b/src/korvid/core/service_analysis.py new file mode 100644 index 00000000..1223425c --- /dev/null +++ b/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), + } diff --git a/src/korvid/evals/fake_kube.py b/src/korvid/evals/fake_kube.py index d636a511..a01f0cf4 100644 --- a/src/korvid/evals/fake_kube.py +++ b/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) ] diff --git a/src/korvid/k8s/client.py b/src/korvid/k8s/client.py index 085bf38d..7acf3493 100644 --- a/src/korvid/k8s/client.py +++ b/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 diff --git a/src/korvid/k8s/models.py b/src/korvid/k8s/models.py index 4b049162..83a7d51e 100644 --- a/src/korvid/k8s/models.py +++ b/src/korvid/k8s/models.py @@ -214,6 +214,15 @@ def _owner_uids(meta: dict[str, Any]) -> tuple[str, ...]: return tuple(str(ref["uid"]) for ref in (meta.get("ownerReferences") or []) if ref.get("uid")) +def _service_owner_uids(meta: dict[str, Any]) -> tuple[str, ...]: + """UIDs from ownerReferences where kind=='Service' and apiVersion=='v1' (core Service only).""" + return tuple( + str(ref["uid"]) + for ref in (meta.get("ownerReferences") or []) + if ref.get("uid") and ref.get("kind") == "Service" and ref.get("apiVersion") == "v1" + ) + + def _labels(meta: dict[str, Any]) -> tuple[tuple[str, str], ...]: """`metadata.labels` as a hashable tuple for frozen summaries (issue #44).""" labels = meta.get("labels") @@ -327,6 +336,48 @@ def from_manifest(cls, kind: str, manifest: dict[str, Any]) -> PackageManifestSu ) +@dataclass(frozen=True) +class EndpointSliceSummary(GenericSummary): + """EndpointSlice summary with service readiness fields.""" + + service_name: str = "" + address_type: str = "" + endpoints: int = 0 + ready_endpoints: int = 0 + #: UIDs from ownerReferences with kind=='Service' and apiVersion=='v1' only. + #: Used for service-replacement / stale-owner checks (PR #212). + service_owner_uids: tuple[str, ...] = () + + @classmethod + def from_manifest(cls, kind: str, manifest: dict[str, Any]) -> EndpointSliceSummary: + base = GenericSummary.from_manifest(kind, manifest) + meta = manifest.get("metadata") or {} + raw_endpoints = manifest.get("endpoints") + endpoints = raw_endpoints if isinstance(raw_endpoints, list) else [] + ready_endpoints = sum(1 for item in endpoints if _endpoint_is_ready(item)) + return cls( + **vars(base), + service_name=str(dict(base.labels).get("kubernetes.io/service-name", "")), + address_type=str(manifest.get("addressType") or ""), + endpoints=len(endpoints), + ready_endpoints=ready_endpoints, + service_owner_uids=_service_owner_uids(meta), + ) + + +def _endpoint_is_ready(item: Any) -> bool: + """True when an EndpointSlice endpoint is ready or omits the ready flag.""" + + if not isinstance(item, dict): + return False + conditions = item.get("conditions") + if conditions is None: + return True + if not isinstance(conditions, dict): + return False + return conditions.get("ready") is not False + + @dataclass(frozen=True) class OLMSubscriptionSummary(GenericSummary): """OLM Subscription (operators.coreos.com) - an installed operator.""" @@ -373,6 +424,7 @@ def from_manifest(cls, kind: str, manifest: dict[str, Any]) -> CSVSummary: #: OLM's API group; other groups also define kinds named "Subscription", so #: the dispatch below checks the manifest's apiVersion, not just the kind. +_DISCOVERY_GROUP_PREFIX = "discovery.k8s.io/" _OLM_GROUP_PREFIX = "operators.coreos.com/" _PACKAGES_GROUP_PREFIX = "packages.operators.coreos.com/" @@ -416,12 +468,33 @@ def from_pod_manifest(cls, kind: str, manifest: dict[str, Any]) -> PodListSummar ) -def summary_for(kind: str, manifest: dict[str, Any]) -> GenericSummary: - """Build the richest summary available for *kind* (ReplicaSet gets history fields).""" +def summary_for(kind: str, manifest: dict[str, Any], *, group: str | None = None) -> GenericSummary: + """Build the richest summary available for *kind* (ReplicaSet gets history fields). + + Args: + kind: The Kubernetes kind name. + manifest: The raw object manifest. + group: Authoritative API group from the resource discovery metadata. + When provided, it takes precedence over the manifest's `apiVersion` + for group-sensitive dispatch (e.g. EndpointSlice). When absent the + existing `apiVersion`-prefix fallback is used so direct callers and + tests that do not have ResourceMeta continue to work unchanged. + """ if kind == "Pod": return PodListSummary.from_pod_manifest(kind, manifest) if kind == "ReplicaSet": return ReplicaSetSummary.from_manifest(kind, manifest) + if kind == "EndpointSlice": + # Use the authoritative group when available to avoid misclassifying + # LIST items that omit apiVersion/TypeMeta (native K8s behaviour). + is_discovery = ( + group == "discovery.k8s.io" + if group is not None + else str(manifest.get("apiVersion") or "").startswith(_DISCOVERY_GROUP_PREFIX) + ) + if is_discovery: + return EndpointSliceSummary.from_manifest(kind, manifest) + return GenericSummary.from_manifest(kind, manifest) api_version = str(manifest.get("apiVersion") or "") if kind == "PackageManifest" and api_version.startswith(_PACKAGES_GROUP_PREFIX): return PackageManifestSummary.from_manifest(kind, manifest) diff --git a/src/korvid/tools/executor.py b/src/korvid/tools/executor.py index 8cdef1e5..960b1c54 100644 --- a/src/korvid/tools/executor.py +++ b/src/korvid/tools/executor.py @@ -17,11 +17,19 @@ redact_document, redact_text, ) +from korvid.core.service_analysis import ( + EndpointSliceSnapshot, + EvidenceGap, + ResourceIdentity, + ServiceSnapshot, + analyze_service_endpoints, +) from korvid.k8s.discovery import ResourceMeta from korvid.k8s.errors import ApiStatusError from korvid.k8s.helm import HelmReleaseSummary, HelmRevisionSummary from korvid.k8s.models import ( CSVSummary, + EndpointSliceSummary, GenericSummary, OLMSubscriptionSummary, PackageManifestSummary, @@ -254,6 +262,14 @@ def _package_facts(s: PackageManifestSummary) -> str: ) +def _endpoint_slice_facts(s: EndpointSliceSummary) -> str: + return ( + f"service={_clamp(s.service_name) or '?'}" + f" ready={s.ready_endpoints}/{s.endpoints}" + f" address_type={_clamp(s.address_type) or '?'}" + ) + + def _generic_facts(s: GenericSummary) -> str: return f"desired={s.desired}" if s.desired is not None else "" @@ -289,6 +305,7 @@ def _helm_revision_facts(s: HelmRevisionSummary) -> str: OLMSubscriptionSummary: _subscription_facts, CSVSummary: _csv_facts, PackageManifestSummary: _package_facts, + EndpointSliceSummary: _endpoint_slice_facts, # Helm's synthetic kinds are not reachable through list_resources today # (a follow-up adds a helm listing tool), but the contract keeps their # facts registered so that tool renders release status on day one. @@ -1035,16 +1052,34 @@ async def _get_events(self, args: dict[str, Any]) -> str: "PersistentVolumeClaim": ResourceMeta( "PersistentVolumeClaim", "persistentvolumeclaims", "", "v1", True ), + "Service": ResourceMeta("Service", "services", "", "v1", True), + "EndpointSlice": ResourceMeta( + "EndpointSlice", "endpointslices", "discovery.k8s.io", "v1", True + ), } def _meta_for_kind_name(self, kind_name: str) -> ResourceMeta | None: """Discovery metadata for an API kind name (e.g. ``"ReplicaSet"``), - falling back to fixed metadata for the stable built-in kinds.""" + falling back to fixed metadata for the stable built-in kinds. + + For kinds tracked in `_DIAGNOSE_BUILTIN_METAS` only discovered metadata + whose group matches the builtin's group is considered authoritative; a + same-kind CRD from a different group is skipped so it cannot shadow the + stable builtin (e.g. a CRD named ``EndpointSlice`` in ``example.io`` + must not displace ``discovery.k8s.io`` EndpointSlice lookups). + """ + builtin = self._DIAGNOSE_BUILTIN_METAS.get(kind_name) discovered = next( - (m for m in self._aliases.values() if m.kind == kind_name and not m.synthetic), + ( + m + for m in self._aliases.values() + if m.kind == kind_name + and not m.synthetic + and (builtin is None or m.group == builtin.group) + ), None, ) - return discovered or self._DIAGNOSE_BUILTIN_METAS.get(kind_name) + return discovered or builtin async def _diagnose_owner_chain(self, namespace: str, pod: dict[str, Any]) -> str: """``Deployment api (via ReplicaSet api-6f)`` — best-effort, never raises.""" @@ -1536,6 +1571,80 @@ async def _diagnose_workload(self, args: dict[str, Any]) -> ToolOutcome: workload = await self._kube.get_object(meta, namespace, name) return await self._diagnose_deployment(namespace, name, workload) + def _require_diagnose_meta(self, kind_name: str) -> ResourceMeta: + """Return discovery metadata for `kind_name`, falling back to built-ins. + + Raises: + ValueError: when neither discovery nor the built-in table knows + the kind — the diagnostic cannot proceed without an API path. + """ + meta = self._meta_for_kind_name(kind_name) + if meta is None: + raise ValueError(f"{kind_name} API was not discovered") + return meta + + async def _diagnose_service(self, args: dict[str, Any]) -> str: + """One-GET/one-LIST service endpoint diagnosis (issue #191).""" + name = _reject_slash_name(str(args["service"]), "service") + namespace = _reject_slash_name(str(args["namespace"]), "namespace") + service_meta = self._require_diagnose_meta("Service") + slice_meta = self._require_diagnose_meta("EndpointSlice") + manifest = await self._kube.get_object(service_meta, namespace, name) + service = _service_snapshot(manifest, namespace, name) + try: + summaries = await self._kube.list_objects(slice_meta, namespace) + except ApiStatusError as exc: + report = analyze_service_endpoints( + service, + (), + EvidenceGap("endpointslices", _api_gap_reason(exc)), + ) + else: + slices = tuple( + _endpoint_slice_snapshot(item) + for item in summaries + if isinstance(item, EndpointSliceSummary) + ) + report = analyze_service_endpoints(service, slices) + return dump_bounded_yaml(report.as_document(), MAX_RESULT_CHARS) + + +def _service_snapshot(manifest: dict[str, Any], namespace: str, name: str) -> ServiceSnapshot: + """Build a ``ServiceSnapshot`` from a raw Service manifest.""" + meta = manifest.get("metadata") or {} + uid = str(meta.get("uid") or "") + spec = manifest.get("spec") or {} + service_type = str(spec.get("type") or "ClusterIP") + selector_map = spec.get("selector") + if isinstance(selector_map, dict): + selector: tuple[tuple[str, str], ...] = tuple( + sorted((str(k), str(v)) for k, v in selector_map.items()) + ) + else: + selector = () + return ServiceSnapshot( + identity=ResourceIdentity("Service", namespace, name, uid), + service_type=service_type, + selector=selector, + ) + + +def _endpoint_slice_snapshot(item: EndpointSliceSummary) -> EndpointSliceSnapshot: + """Build an ``EndpointSliceSnapshot`` from an ``EndpointSliceSummary``.""" + return EndpointSliceSnapshot( + identity=ResourceIdentity("EndpointSlice", item.namespace, item.name, item.uid), + service_name=item.service_name, + service_owner_uids=item.service_owner_uids, + address_type=item.address_type, + endpoints=item.endpoints, + ready_endpoints=item.ready_endpoints, + ) + + +def _api_gap_reason(exc: ApiStatusError) -> str: + """Human-readable gap reason from an API status error.""" + return f"HTTP {exc.status}: {exc.reason}" + def _mask_manifest(manifest: dict[str, Any]) -> tuple[dict[str, Any], list[RedactionRecord]]: """Strip managedFields, then redact the whole document recursively. diff --git a/src/korvid/tools/follow.py b/src/korvid/tools/follow.py index 3212b623..94609657 100644 --- a/src/korvid/tools/follow.py +++ b/src/korvid/tools/follow.py @@ -34,6 +34,7 @@ "helm_list_releases", "diagnose_pod", "diagnose_workload", + "diagnose_service", } ) @@ -136,10 +137,22 @@ async def _mirror(ui: UIBridge, tool: str, args: Mapping[str, Any]) -> str | Non # the first-container resolution, and the fuller pane is the more # useful thing to watch. return await ui.agent_open_logs(pod, namespace, _str_or_none(args.get("container"))) + return await _mirror_diagnose(ui, tool, args, namespace) + + +async def _mirror_diagnose( + ui: UIBridge, tool: str, args: Mapping[str, Any], namespace: str | None +) -> str | None: + """Mirror a diagnose_* cluster read to the appropriate describe pane.""" if tool == "diagnose_pod": # The registry schema names the target 'pod' (matching get_logs). pod = _str_or_none(args.get("pod")) if pod is None: return None return await ui.agent_open_describe("pods", pod, namespace) + if tool == "diagnose_service": + service = _str_or_none(args.get("service")) + if service is None: + return None + return await ui.agent_open_describe("services", service, namespace) return None diff --git a/src/korvid/tools/registry.py b/src/korvid/tools/registry.py index 34c225e7..0e160571 100644 --- a/src/korvid/tools/registry.py +++ b/src/korvid/tools/registry.py @@ -625,6 +625,35 @@ def mcp_tool_schemas(*, write_proposals: bool = False) -> list[dict[str, Any]]: }, }, ), + ToolDef( + name="diagnose_service", + effect="cluster_read", + dispatch="_diagnose_service", + surfaces=_ALL_SURFACES, + result_format="structured_yaml", + schema={ + "type": "function", + "function": { + "name": "diagnose_service", + "description": ( + "Deterministically check whether a Service has current ready " + "EndpointSlice endpoints. Returns versioned findings and explicit " + "evidence gaps; prefer this when traffic cannot reach a Service." + ), + "parameters": { + "type": "object", + "properties": { + "service": {"type": "string", "description": "Service name."}, + "namespace": { + "type": "string", + "description": "Kubernetes namespace containing the Service.", + }, + }, + "required": ["service", "namespace"], + }, + }, + }, + ), ToolDef( name="navigate", effect="ui_only", diff --git a/tests/agent/test_runtime.py b/tests/agent/test_runtime.py index 07dca703..03062c33 100644 --- a/tests/agent/test_runtime.py +++ b/tests/agent/test_runtime.py @@ -73,6 +73,10 @@ async def collect( return [e async for e in runtime.run_turn(text, screen_context)] +def _read_tools_request_ceiling(non_tool_request_budget: int) -> int: + return len(json.dumps(READ_TOOLS, separators=(",", ":"))) + non_tool_request_budget + + async def test_text_only_turn() -> None: p = ScriptedProvider([[{"type": "text_delta", "text": "hi"}, {"type": "done"}]]) events = await collect(AgentRuntime(p, EchoExecutor()), "hello") @@ -1515,7 +1519,7 @@ async def test_over_ceiling_request_drops_the_oldest_turn_and_still_reaches_the_ provider, EchoExecutor(), max_history_chars=40_000, - max_request_chars=12_000, + max_request_chars=_read_tools_request_ceiling(10_000), ) first = await collect(runtime, "a" * 6_000) @@ -1647,12 +1651,14 @@ async def test_estimated_prompt_cost_reflects_the_history_actually_sent() -> Non provider, EchoExecutor(), max_history_chars=40_000, - max_request_chars=12_000, + max_request_chars=_read_tools_request_ceiling(10_000), ) - await collect(runtime, "a" * 6_000) + first = await collect(runtime, "a" * 6_000) + assert not [event for event in first if isinstance(event, AgentError)] first_total_in = runtime.total_tokens[0] - await collect(runtime, "b" * 6_000) + second = await collect(runtime, "b" * 6_000) + assert not [event for event in second if isinstance(event, AgentError)] sent_chars = len(json.dumps(provider.calls[1], ensure_ascii=False)) second_turn_in = runtime.total_tokens[0] - first_total_in diff --git a/tests/core/test_service_analysis.py b/tests/core/test_service_analysis.py new file mode 100644 index 00000000..8686f658 --- /dev/null +++ b/tests/core/test_service_analysis.py @@ -0,0 +1,236 @@ +"""Tests for deterministic Service-to-EndpointSlice analysis.""" + +from __future__ import annotations + +from korvid.core.service_analysis import ( + EndpointSliceSnapshot, + EvidenceGap, + ResourceIdentity, + ServiceSnapshot, + analyze_service_endpoints, +) + + +def _service( + *, + kind: str = "Service", + namespace: str = "default", + name: str = "web", + uid: str = "", + service_type: str = "ClusterIP", + selector: tuple[tuple[str, str], ...] = (("app", "web"),), +) -> ServiceSnapshot: + return ServiceSnapshot( + identity=ResourceIdentity(kind=kind, namespace=namespace, name=name, uid=uid), + service_type=service_type, + selector=selector, + ) + + +def _slice( + *, + kind: str = "EndpointSlice", + namespace: str = "default", + name: str = "web-1", + uid: str = "", + service_name: str = "web", + service_owner_uids: tuple[str, ...] = (), + address_type: str = "IPv4", + endpoints: int = 1, + ready_endpoints: int = 1, +) -> EndpointSliceSnapshot: + return EndpointSliceSnapshot( + identity=ResourceIdentity(kind=kind, namespace=namespace, name=name, uid=uid), + service_name=service_name, + service_owner_uids=service_owner_uids, + address_type=address_type, + endpoints=endpoints, + ready_endpoints=ready_endpoints, + ) + + +def test_ready_current_slice_is_healthy() -> None: + report = analyze_service_endpoints( + _service(uid="svc-1"), + (_slice(service_owner_uids=("svc-1",), endpoints=2, ready_endpoints=1),), + ) + assert report.outcome == "healthy" + assert report.findings == () + assert report.evidence[0].field == "endpoints.ready" + + +def test_no_current_slices_is_a_versioned_finding() -> None: + report = analyze_service_endpoints(_service(uid="svc-1"), ()) + assert report.outcome == "findings" + assert report.findings[0].rule_id == "service.no_endpoint_slices" + assert report.findings[0].rule_version == "1" + + +def test_unavailable_slice_evidence_never_reports_healthy() -> None: + gap = EvidenceGap(source="endpointslices", reason="forbidden (HTTP 403)") + report = analyze_service_endpoints(_service(), (), gap) + assert report.outcome == "incomplete" + assert report.gaps == (gap,) + + +def test_replaced_service_slice_is_stale_not_healthy() -> None: + report = analyze_service_endpoints( + _service(uid="new-uid"), + (_slice(service_owner_uids=("old-uid",), endpoints=1, ready_endpoints=1),), + ) + assert report.outcome == "incomplete" + assert report.findings == () + assert report.gaps[0].source == "endpointslices/stale-owner" + assert ( + report.gaps[0].reason + == "1 EndpointSlice is owned by a different Service UID than 'new-uid'." + ) + + +def test_owned_slice_with_missing_service_uid_is_stale() -> None: + report = analyze_service_endpoints( + _service(), + (_slice(service_owner_uids=("other",), endpoints=1, ready_endpoints=1),), + ) + assert report.outcome == "incomplete" + assert report.findings == () + assert report.gaps[0].source == "endpointslices/stale-owner" + assert ( + report.gaps[0].reason + == "1 EndpointSlice is owned by a Service, but the Service UID is absent." + ) + + +def test_stale_owner_reason_uses_plural_grammar() -> None: + report = analyze_service_endpoints( + _service(uid="new-uid"), + ( + _slice(name="web-1", uid="slice-1", service_owner_uids=("old-uid",)), + _slice(name="web-2", uid="slice-2", service_owner_uids=("old-uid",)), + ), + ) + assert report.outcome == "incomplete" + assert report.gaps[0].reason + assert ( + report.gaps[0].reason + == "2 EndpointSlices are owned by a different Service UID than 'new-uid'." + ) + + +def test_matching_requires_namespace_and_service_name() -> None: + report = analyze_service_endpoints( + _service(uid="svc-1", namespace="default"), + (_slice(namespace="other", service_owner_uids=("svc-1",), endpoints=1, ready_endpoints=1),), + ) + assert report.outcome == "findings" + assert report.findings[0].rule_id == "service.no_endpoint_slices" + + +def test_current_slices_without_ready_endpoints_warn() -> None: + report = analyze_service_endpoints( + _service(uid="svc-1"), + (_slice(service_owner_uids=("svc-1",), endpoints=2, ready_endpoints=0),), + ) + assert report.findings[0].rule_id == "service.no_ready_endpoints" + + +def test_selectorless_unowned_slice_is_valid_manual_evidence() -> None: + report = analyze_service_endpoints( + _service(uid="svc-1", selector=()), + (_slice(service_owner_uids=(), endpoints=1, ready_endpoints=1),), + ) + assert report.outcome == "healthy" + + +def test_external_name_is_not_applicable() -> None: + report = analyze_service_endpoints(_service(service_type="ExternalName"), ()) + assert report.outcome == "not_applicable" + + +def test_document_uses_stable_public_keys() -> None: + document = analyze_service_endpoints(_service(), ()).as_document() + assert tuple(document) == ( + "analyzer", + "version", + "outcome", + "primary", + "findings", + "evidence", + "gaps", + ) + + +# -- service_owner_uids stale-check tests (PR #212 fix) --------------------- + + +def _slice_with_svc( + *, + namespace: str = "default", + name: str = "web-1", + uid: str = "", + service_name: str = "web", + service_owner_uids: tuple[str, ...] = (), + address_type: str = "IPv4", + endpoints: int = 1, + ready_endpoints: int = 1, +) -> EndpointSliceSnapshot: + """Helper for tests that need to set service_owner_uids directly.""" + return EndpointSliceSnapshot( + identity=ResourceIdentity(kind="EndpointSlice", namespace=namespace, name=name, uid=uid), + service_name=service_name, + service_owner_uids=service_owner_uids, + address_type=address_type, + endpoints=endpoints, + ready_endpoints=ready_endpoints, + ) + + +def test_custom_controller_owner_only_with_ready_endpoints_is_healthy() -> None: + """An EndpointSlice owned only by a custom CRD controller (no Service ref) must + not be flagged stale. The stale check applies only to core/v1 Service owners.""" + report = analyze_service_endpoints( + _service(uid="svc-1"), + (_slice_with_svc(service_owner_uids=(), endpoints=1, ready_endpoints=1),), + ) + assert report.outcome == "healthy" + + +def test_mismatching_service_owner_uid_yields_incomplete() -> None: + """When service_owner_uids contains a UID that differs from the current Service, + the slice is stale and the outcome must be incomplete.""" + report = analyze_service_endpoints( + _service(uid="new-svc-uid"), + (_slice_with_svc(service_owner_uids=("old-svc-uid",), endpoints=1, ready_endpoints=1),), + ) + assert report.outcome == "incomplete" + assert report.gaps[0].source == "endpointslices/stale-owner" + + +def test_mixed_custom_and_service_owner_uses_only_service_refs_for_stale_check() -> None: + """When a slice has both a custom-controller ref and a Service ref, only the + Service UID matters for the stale decision.""" + # Matching Service UID → healthy even though custom UID is unrelated + report_healthy = analyze_service_endpoints( + _service(uid="svc-1"), + ( + _slice_with_svc( + service_owner_uids=("svc-1",), + endpoints=1, + ready_endpoints=1, + ), + ), + ) + assert report_healthy.outcome == "healthy" + + # Mismatching Service UID → stale, even though custom UID happens to be there + report_stale = analyze_service_endpoints( + _service(uid="svc-new"), + ( + _slice_with_svc( + service_owner_uids=("svc-old",), + endpoints=1, + ready_endpoints=1, + ), + ), + ) + assert report_stale.outcome == "incomplete" diff --git a/tests/k8s/test_client.py b/tests/k8s/test_client.py index d32ff4d1..3fe8c392 100644 --- a/tests/k8s/test_client.py +++ b/tests/k8s/test_client.py @@ -763,6 +763,45 @@ async def test_list_objects_raises_api_status_error() -> None: await client.list_objects(meta, "default") +async def test_list_objects_endpointslice_without_api_version_returns_typed_summary() -> None: + """Native LIST items often omit apiVersion/TypeMeta. When the ResourceMeta + group is 'discovery.k8s.io', _object_summary must still produce + EndpointSliceSummary with correct ready counts (issue #191 fix).""" + from korvid.k8s.models import EndpointSliceSummary + + client = KubeClient() + meta = ResourceMeta("EndpointSlice", "endpointslices", "discovery.k8s.io", "v1", True) + raw_item: dict[str, Any] = { + # No apiVersion / kind — this is how the Kubernetes API server returns LIST items. + "metadata": { + "name": "api-x1", + "namespace": "shop", + "labels": {"kubernetes.io/service-name": "api"}, + "ownerReferences": [{"uid": "svc-1"}], + }, + "addressType": "IPv4", + "endpoints": [ + {"conditions": {"ready": True}}, + {"conditions": {}}, + {"conditions": {"ready": False}}, + ], + } + list_resp = {"items": [raw_item]} + request_json_mock = AsyncMock(return_value=list_resp) + + with ( + patch.object(client, "_api", MagicMock()), + patch.object(client, "_request_json", request_json_mock), + ): + summaries = await client.list_objects(meta, "shop") + + assert len(summaries) == 1 + summary = summaries[0] + assert isinstance(summary, EndpointSliceSummary) + assert summary.ready_endpoints == 2 + assert summary.endpoints == 3 + + # Write operations (issue #16) ------------------------------------------------- diff --git a/tests/k8s/test_models.py b/tests/k8s/test_models.py index 511b9500..f796c8aa 100644 --- a/tests/k8s/test_models.py +++ b/tests/k8s/test_models.py @@ -5,6 +5,7 @@ from korvid.k8s.models import ( CSVSummary, + EndpointSliceSummary, GenericSummary, OLMSubscriptionSummary, PackageManifestSummary, @@ -244,6 +245,69 @@ def test_generic_summary_tolerates_non_mapping_spec() -> None: assert GenericSummary.from_manifest("Widget", manifest).desired is None +def test_endpoint_slice_summary_counts_nil_ready_as_ready() -> None: + summary = summary_for( + "EndpointSlice", + { + "apiVersion": "discovery.k8s.io/v1", + "kind": "EndpointSlice", + "metadata": { + "name": "api-x1", + "namespace": "shop", + "labels": {"kubernetes.io/service-name": "api"}, + "ownerReferences": [{"uid": "svc-1"}], + }, + "addressType": "IPv4", + "endpoints": [ + {"conditions": {"ready": True}}, + {"conditions": {}}, + {"conditions": {"ready": False}}, + ], + }, + ) + assert isinstance(summary, EndpointSliceSummary) + assert summary.service_name == "api" + assert summary.address_type == "IPv4" + assert summary.endpoints == 3 + assert summary.ready_endpoints == 2 + + +def test_same_named_endpoint_slice_crd_stays_generic() -> None: + summary = summary_for( + "EndpointSlice", + {"apiVersion": "example.io/v1", "metadata": {"name": "custom"}}, + ) + assert type(summary) is GenericSummary + + +def test_endpoint_slice_summary_ignores_malformed_endpoints() -> None: + summary = summary_for( + "EndpointSlice", + { + "apiVersion": "discovery.k8s.io/v1", + "metadata": {"name": "api-x1", "namespace": "shop"}, + "endpoints": "oops", + }, + ) + assert isinstance(summary, EndpointSliceSummary) + assert summary.endpoints == 0 + assert summary.ready_endpoints == 0 + + +def test_endpoint_slice_summary_treats_non_mapping_conditions_as_not_ready() -> None: + summary = summary_for( + "EndpointSlice", + { + "apiVersion": "discovery.k8s.io/v1", + "metadata": {"name": "api-x1", "namespace": "shop"}, + "endpoints": [{"conditions": "oops"}], + }, + ) + assert isinstance(summary, EndpointSliceSummary) + assert summary.endpoints == 1 + assert summary.ready_endpoints == 0 + + def test_pod_summary_owner_uids() -> None: manifest: dict[str, Any] = { "metadata": { @@ -316,6 +380,44 @@ def test_summary_for_falls_back_to_generic() -> None: assert not isinstance(summary, ReplicaSetSummary) +# --------------------------------------------------------------------------- +# summary_for: authoritative group kwarg (issue #191 fix) +# --------------------------------------------------------------------------- + + +def test_summary_for_endpointslice_without_api_version_dispatched_by_group() -> None: + """A LIST item that omits apiVersion becomes EndpointSliceSummary when the + authoritative group is provided by the caller (e.g. KubeClient._object_summary).""" + manifest: dict[str, Any] = { + "metadata": { + "name": "api-x1", + "namespace": "shop", + "labels": {"kubernetes.io/service-name": "api"}, + "ownerReferences": [{"uid": "svc-1"}], + }, + "addressType": "IPv4", + "endpoints": [ + {"conditions": {"ready": True}}, + {"conditions": {}}, + ], + } + summary = summary_for("EndpointSlice", manifest, group="discovery.k8s.io") + assert isinstance(summary, EndpointSliceSummary) + assert summary.service_name == "api" + assert summary.ready_endpoints == 2 + + +def test_summary_for_endpointslice_explicit_non_discovery_group_stays_generic() -> None: + """When an authoritative non-discovery group is provided, even a manifest that + claims apiVersion='discovery.k8s.io/v1' must remain GenericSummary.""" + manifest: dict[str, Any] = { + "apiVersion": "discovery.k8s.io/v1", + "metadata": {"name": "custom"}, + } + summary = summary_for("EndpointSlice", manifest, group="example.io") + assert type(summary) is GenericSummary + + def test_age_5m() -> None: gs = GenericSummary(name="x", namespace="ns", kind="Pod", created="2024-01-01T12:00:00Z") now = datetime(2024, 1, 1, 12, 5, 0, tzinfo=UTC) @@ -1231,3 +1333,72 @@ def test_summaries_default_to_no_labels() -> None: pod = PodSummary.from_manifest({"metadata": {"name": "p"}, "spec": {}, "status": {}}) assert gs.labels == () assert pod.labels == () + + +# -- EndpointSliceSummary.service_owner_uids projection (PR #212) ------------ + + +def test_endpoint_slice_summary_service_owner_uids_filters_to_core_service() -> None: + """Mixed ownerReferences: generic owner_uids contains all UIDs, but + service_owner_uids contains only refs whose kind=='Service' and apiVersion=='v1'.""" + manifest: dict[str, Any] = { + "apiVersion": "discovery.k8s.io/v1", + "kind": "EndpointSlice", + "metadata": { + "name": "api-x1", + "namespace": "shop", + "labels": {"kubernetes.io/service-name": "api"}, + "ownerReferences": [ + # core/v1 Service — should appear in service_owner_uids + {"kind": "Service", "apiVersion": "v1", "uid": "svc-uid-1"}, + # custom CRD controller — must NOT appear in service_owner_uids + {"kind": "MeshController", "apiVersion": "mesh.example.io/v1", "uid": "crd-uid-2"}, + ], + }, + "addressType": "IPv4", + "endpoints": [{"conditions": {"ready": True}}], + } + summary = summary_for("EndpointSlice", manifest) + assert isinstance(summary, EndpointSliceSummary) + # Generic owner_uids carries all UIDs (unchanged relation behaviour) + assert set(summary.owner_uids) == {"svc-uid-1", "crd-uid-2"} + # service_owner_uids carries only the core/v1 Service UID + assert summary.service_owner_uids == ("svc-uid-1",) + + +def test_endpoint_slice_summary_wrong_api_version_service_excluded() -> None: + """A 'Service' ownerRef with a non-core apiVersion must be excluded from + service_owner_uids (it might be a CRD impersonating the name 'Service').""" + manifest: dict[str, Any] = { + "apiVersion": "discovery.k8s.io/v1", + "kind": "EndpointSlice", + "metadata": { + "name": "api-x2", + "namespace": "shop", + "ownerReferences": [ + {"kind": "Service", "apiVersion": "custom.io/v1", "uid": "fake-svc-uid"}, + ], + }, + "addressType": "IPv4", + "endpoints": [], + } + summary = summary_for("EndpointSlice", manifest) + assert isinstance(summary, EndpointSliceSummary) + # Generic owner_uids still carries the UID + assert "fake-svc-uid" in summary.owner_uids + # service_owner_uids must exclude the malformed/wrong-apiVersion ref + assert summary.service_owner_uids == () + + +def test_endpoint_slice_summary_no_owner_refs_gives_empty_service_owner_uids() -> None: + """EndpointSlice without ownerReferences has empty service_owner_uids.""" + manifest: dict[str, Any] = { + "apiVersion": "discovery.k8s.io/v1", + "kind": "EndpointSlice", + "metadata": {"name": "manual-slice", "namespace": "shop"}, + "addressType": "IPv4", + "endpoints": [{"conditions": {"ready": True}}], + } + summary = summary_for("EndpointSlice", manifest) + assert isinstance(summary, EndpointSliceSummary) + assert summary.service_owner_uids == () diff --git a/tests/tools/test_executor.py b/tests/tools/test_executor.py index 8bc2128d..64a184cb 100644 --- a/tests/tools/test_executor.py +++ b/tests/tools/test_executor.py @@ -11,10 +11,10 @@ import korvid.tools.executor as executor_module from korvid.core.secrets import MASK_PLACEHOLDER -from korvid.k8s.discovery import PODS_META +from korvid.k8s.discovery import PODS_META, ResourceMeta from korvid.k8s.errors import ApiStatusError from korvid.k8s.logs import LogLine -from korvid.k8s.models import summary_for +from korvid.k8s.models import EndpointSliceSummary, GenericSummary, summary_for from korvid.tools.executor import ( MAX_RESULT_CHARS, READ_TOOLS, @@ -54,6 +54,7 @@ def test_read_tools_schema_names() -> None: "helm_list_releases", "diagnose_pod", "diagnose_workload", + "diagnose_service", ] @@ -1187,7 +1188,7 @@ async def list_objects(self, meta: Any, namespace: str | None) -> list[Any]: metadata = obj.get("metadata") or {} if meta.namespaced and namespace is not None and metadata.get("namespace") != namespace: continue - summaries.append(summary_for(meta.kind, obj)) + summaries.append(summary_for(meta.kind, obj, group=meta.group)) return summaries async def list_events_for( @@ -3072,3 +3073,252 @@ async def test_a_bounded_produced_manifest_survives_the_strict_reader() -> None: loaded = load_structured_document(outcome.text) assert loaded["kind"] == "CompositeApp" + + +# -- diagnose_service tests (issue #191) ------------------------------------ + + +class ServiceDiagnosisKube: + """Records cluster calls; returns scripted service manifest and slices.""" + + def __init__( + self, + service: dict[str, Any], + slices: list[GenericSummary] | None = None, + list_error: ApiStatusError | None = None, + ) -> None: + self._service = service + self._slices: list[GenericSummary] = slices or [] + self._list_error = list_error + self.calls: list[tuple[str, ...]] = [] + + async def get_object(self, meta: Any, namespace: str | None, name: str) -> dict[str, Any]: + self.calls.append(("get", meta.kind, str(namespace), name)) + return self._service + + async def list_objects(self, meta: Any, namespace: str | None) -> list[GenericSummary]: + self.calls.append(("list", meta.kind, str(namespace))) + if self._list_error is not None: + raise self._list_error + return self._slices + + +def _service_manifest(uid: str = "") -> dict[str, Any]: + return { + "apiVersion": "v1", + "kind": "Service", + "metadata": {"name": "api", "namespace": "shop", "uid": uid}, + "spec": {"type": "ClusterIP", "selector": {"app": "api"}}, + } + + +def _endpoint_slice_summary( + name: str = "api-abc", + owner_uids: tuple[str, ...] = (), + service_owner_uids: tuple[str, ...] | None = None, + ready_endpoints: int = 0, +) -> EndpointSliceSummary: + # When service_owner_uids is not given, mirror owner_uids for backwards compat + # in existing tests (those tests use a single same-kind Service UID). + resolved_service_owner_uids = owner_uids if service_owner_uids is None else service_owner_uids + return EndpointSliceSummary( + name=name, + namespace="shop", + kind="EndpointSlice", + created="", + uid="", + owner_uids=owner_uids, + labels=(("kubernetes.io/service-name", "api"),), + service_name="api", + address_type="IPv4", + endpoints=1, + ready_endpoints=ready_endpoints, + service_owner_uids=resolved_service_owner_uids, + ) + + +def _svc_executor(kube: Any) -> ToolExecutor: + return ToolExecutor(kube, {}) + + +def test_diagnose_service_is_a_shared_structured_read_tool() -> None: + definition = TOOLS_BY_NAME["diagnose_service"] + assert definition.effect == "cluster_read" + assert definition.result_format == "structured_yaml" + assert definition.surfaces == frozenset({"full_agent", "small_agent", "mcp"}) + + +@pytest.mark.asyncio +async def test_diagnose_service_gets_once_and_lists_once() -> None: + kube = ServiceDiagnosisKube( + service=_service_manifest(uid="svc-1"), + slices=[_endpoint_slice_summary(owner_uids=("svc-1",), ready_endpoints=1)], + ) + text = await _svc_executor(kube).execute( + "diagnose_service", {"service": "api", "namespace": "shop"} + ) + document = load_structured_document(text) + assert document["outcome"] == "healthy" + assert kube.calls == [ + ("get", "Service", "shop", "api"), + ("list", "EndpointSlice", "shop"), + ] + + +@pytest.mark.asyncio +async def test_diagnose_service_projects_rbac_denial_as_gap() -> None: + kube = ServiceDiagnosisKube( + service=_service_manifest(uid="svc-1"), + list_error=ApiStatusError(403, "Forbidden", ""), + ) + document = load_structured_document( + await _svc_executor(kube).execute( + "diagnose_service", {"service": "api", "namespace": "shop"} + ) + ) + assert document["outcome"] == "incomplete" + assert document["gaps"][0]["source"] == "endpointslices" + + +@pytest.mark.asyncio +async def test_diagnose_service_rejects_composite_name_before_cluster_io() -> None: + kube = ServiceDiagnosisKube(service=_service_manifest()) + text = await _svc_executor(kube).execute( + "diagnose_service", {"service": "shop/api", "namespace": "shop"} + ) + assert text.startswith("ERROR:") + assert kube.calls == [] + + +@pytest.mark.asyncio +async def test_diagnose_service_ignores_untyped_rows() -> None: + kube = ServiceDiagnosisKube( + service=_service_manifest(uid="svc-1"), + slices=[GenericSummary("other", "shop", "Other", "")], + ) + document = load_structured_document( + await _svc_executor(kube).execute( + "diagnose_service", {"service": "api", "namespace": "shop"} + ) + ) + assert document["findings"][0]["rule_id"] == "service.no_endpoint_slices" + + +@pytest.mark.asyncio +async def test_diagnose_service_result_remains_bounded_yaml() -> None: + kube = ServiceDiagnosisKube( + service=_service_manifest(uid="svc-1"), + slices=[ + _endpoint_slice_summary( + name=f"api-{index:05d}", + owner_uids=("svc-1",), + ready_endpoints=1, + ) + for index in range(2_000) + ], + ) + text = await _svc_executor(kube).execute( + "diagnose_service", {"service": "api", "namespace": "shop"} + ) + assert len(text) <= MAX_RESULT_CHARS + document = load_structured_document(text) + assert document["outcome"] == "healthy" + assert "elided" in text + + +# --------------------------------------------------------------------------- +# _meta_for_kind_name: CRD from wrong group must not shadow the builtin +# --------------------------------------------------------------------------- + + +def test_meta_for_kind_name_ignores_crd_endpointslice_with_wrong_group() -> None: + """A same-kind CRD alias inserted before the real EndpointSlice must not + be selected; _meta_for_kind_name must return the discovery.k8s.io builtin.""" + crd_alias = ResourceMeta("EndpointSlice", "endpointslices", "example.io", "v1alpha1", True) + executor = ToolExecutor(FakeKube(), {"endpointslices": crd_alias}) # type: ignore[arg-type] # minimal fake + meta = executor._meta_for_kind_name("EndpointSlice") + assert meta is not None + assert meta.group == "discovery.k8s.io" + + +# --------------------------------------------------------------------------- +# End-to-end: raw TypeMeta-less list items produce healthy diagnosis +# --------------------------------------------------------------------------- + + +class _RawManifestKube: + """Fake kube that holds raw manifests and dispatches summary_for with group.""" + + def __init__( + self, + service: dict[str, Any], + slice_manifests: list[dict[str, Any]], + ) -> None: + self._service = service + self._slice_manifests = slice_manifests + + async def get_object(self, meta: Any, namespace: str | None, name: str) -> dict[str, Any]: + return self._service + + async def list_objects(self, meta: Any, namespace: str | None) -> list[Any]: + return [summary_for(meta.kind, m, group=meta.group) for m in self._slice_manifests] + + +@pytest.mark.asyncio +async def test_diagnose_service_typemeta_less_slices_return_healthy() -> None: + """Raw LIST items that omit apiVersion/TypeMeta must produce EndpointSliceSummary + (via the group kwarg path) so diagnose_service reports healthy, not no_endpoint_slices.""" + raw_slice: dict[str, Any] = { + # Intentionally omits apiVersion and kind, mirroring real Kubernetes LIST responses. + "metadata": { + "name": "api-abc", + "namespace": "shop", + "uid": "slice-1", + "labels": {"kubernetes.io/service-name": "api"}, + "ownerReferences": [{"uid": "svc-1"}], + }, + "addressType": "IPv4", + "endpoints": [{"conditions": {"ready": True}}], + } + kube = _RawManifestKube( + service=_service_manifest(uid="svc-1"), + slice_manifests=[raw_slice], + ) + document = load_structured_document( + await _svc_executor(kube).execute( + "diagnose_service", {"service": "api", "namespace": "shop"} + ) + ) + assert document["outcome"] == "healthy" + + +@pytest.mark.asyncio +async def test_diagnose_service_unrelated_custom_owner_only_is_healthy() -> None: + """An EndpointSlice owned only by an unrelated CRD controller (no core/v1 Service + ownerRef) must not be flagged stale. Only Service refs are used for stale checks.""" + raw_slice: dict[str, Any] = { + "apiVersion": "discovery.k8s.io/v1", + "kind": "EndpointSlice", + "metadata": { + "name": "api-abc", + "namespace": "shop", + "uid": "slice-crd", + "labels": {"kubernetes.io/service-name": "api"}, + # Only a custom CRD controller ownerRef — no Service ref + "ownerReferences": [ + {"kind": "MeshController", "apiVersion": "mesh.example.io/v1", "uid": "crd-uid-99"}, + ], + }, + "addressType": "IPv4", + "endpoints": [{"conditions": {"ready": True}}], + } + kube = _RawManifestKube( + service=_service_manifest(uid="svc-real"), + slice_manifests=[raw_slice], + ) + document = load_structured_document( + await _svc_executor(kube).execute( + "diagnose_service", {"service": "api", "namespace": "shop"} + ) + ) + assert document["outcome"] == "healthy" diff --git a/tests/tools/test_follow.py b/tests/tools/test_follow.py index aaa1ef53..92c3bd16 100644 --- a/tests/tools/test_follow.py +++ b/tests/tools/test_follow.py @@ -113,6 +113,7 @@ async def test_every_followable_tool_reaches_the_bridge(tool: str) -> None: }, "list_operators": {}, "helm_list_releases": {}, + "diagnose_service": {"service": "x", "namespace": "d"}, }[tool] ui = FakeBridge() result = await mirror_read(ui, tool, args) @@ -150,3 +151,14 @@ async def test_helm_list_releases_mirrors_as_the_helm_view() -> None: result = await mirror_read(ui, "helm_list_releases", {"namespace": "prod"}) assert result is not None assert ui.calls == [("navigate", {"view": "helm", "namespace": "prod"})] + + +async def test_diagnose_service_follow_opens_service_describe() -> None: + ui = FakeBridge() + result = await mirror_read( + ui, + "diagnose_service", + {"service": "api", "namespace": "shop"}, + ) + assert result is not None + assert ui.calls == [("open_describe", {"kind": "services", "name": "api", "namespace": "shop"})] diff --git a/tests/tools/test_list_resources.py b/tests/tools/test_list_resources.py index d7703d5a..b126230f 100644 --- a/tests/tools/test_list_resources.py +++ b/tests/tools/test_list_resources.py @@ -9,6 +9,7 @@ from korvid.k8s.discovery import PODS_META, ResourceMeta from korvid.k8s.models import ( CSVSummary, + EndpointSliceSummary, GenericSummary, OLMSubscriptionSummary, PackageManifestSummary, @@ -158,6 +159,23 @@ def test_generic_facts_show_desired_when_present() -> None: assert summary_facts(bare) == "" +def test_endpoint_slice_facts_line() -> None: + s = EndpointSliceSummary( + name="api-x1", + namespace="shop", + kind="EndpointSlice", + created="", + service_name="api", + address_type="IPv4", + endpoints=3, + ready_endpoints=2, + ) + line = summary_facts(s) + assert "service=api" in line + assert "ready=2/3" in line + assert "address_type=IPv4" in line + + def test_every_typed_summary_has_a_facts_renderer() -> None: """The contract (issue #158): a future typed summary must not silently degrade back to name+age - it either registers a renderer or this diff --git a/tests/tools/test_registry.py b/tests/tools/test_registry.py index a5f50c97..de65d97a 100644 --- a/tests/tools/test_registry.py +++ b/tests/tools/test_registry.py @@ -82,10 +82,11 @@ def test_every_tool_declares_an_outbound_result_format() -> None: "untrusted_text", } assert registry_mod.tool_result_format("get_resource") == "structured_yaml" + assert registry_mod.tool_result_format("diagnose_service") == "structured_yaml" assert all( registry_mod.tool_result_format(d.name) == "untrusted_text" for d in TOOL_DEFS - if d.name != "get_resource" + if d.name not in ("get_resource", "diagnose_service") ) @@ -255,6 +256,7 @@ def test_validate_dispatch_targets_rejects_write_tool_naming_executor_method() - "helm_list_releases", "diagnose_pod", "diagnose_workload", + "diagnose_service", ] _UI_ORDER = ["navigate", "set_filter", "open_logs", "open_describe", "drill_down"] _WRITE_ORDER = ["delete_resource", "scale_resource", "rollout_restart"] @@ -583,3 +585,9 @@ def test_a_non_function_tool_schema_is_rejected() -> None: def test_an_unknown_tool_has_no_result_format() -> None: assert registry_mod.tool_result_format("fetch_manifest") is None + + +def test_registry_dispatches_diagnose_service() -> None: + validate_dispatch_targets(TOOL_DEFS, executor_cls=ToolExecutor, bridge_cls=UIBridge) + names = {schema["function"]["name"] for schema in mcp_tool_schemas()} + assert "diagnose_service" in names