From 2be87b4e5880368df964c099764466c3617a2d3b Mon Sep 17 00:00:00 2001
From: hellices <hellices@naver.com>
Date: Fri, 7 Aug 2026 01:39:51 +0900
Subject: [PATCH 01/12] feat: add deterministic service endpoint analyzer

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 src/korvid/core/service_analysis.py | 303 ++++++++++++++++++++++++++++
 tests/core/test_service_analysis.py | 117 +++++++++++
 2 files changed, 420 insertions(+)
 create mode 100644 src/korvid/core/service_analysis.py
 create mode 100644 tests/core/test_service_analysis.py

diff --git a/src/korvid/core/service_analysis.py b/src/korvid/core/service_analysis.py
new file mode 100644
index 00000000..542766e8
--- /dev/null
+++ b/src/korvid/core/service_analysis.py
@@ -0,0 +1,303 @@
+"""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
+    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),
+            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 service_uid and item.owner_uids and service_uid not in item.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.owner_uids:
+            items.append(Evidence(item.identity, "endpoints.owner_uids", ",".join(item.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.owner_uids for item in slices if item.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)
+    noun = "slice" if count == 1 else "slices"
+    return f"{count} EndpointSlice {noun} are owned by a different Service UID than {service.identity.uid!r}."
+
+
+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/tests/core/test_service_analysis.py b/tests/core/test_service_analysis.py
new file mode 100644
index 00000000..5caca9da
--- /dev/null
+++ b/tests/core/test_service_analysis.py
@@ -0,0 +1,117 @@
+"""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",
+    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,
+        owner_uids=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(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(owner_uids=("old-uid",), endpoints=1, ready_endpoints=1),),
+    )
+    assert report.outcome == "incomplete"
+    assert report.findings == ()
+    assert report.gaps[0].source == "endpointslices/stale-owner"
+
+
+def test_current_slices_without_ready_endpoints_warn() -> None:
+    report = analyze_service_endpoints(
+        _service(uid="svc-1"),
+        (_slice(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(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",
+    )

From 2a270a4c9a6d9362eb4baf30933adbacb7efcf73 Mon Sep 17 00:00:00 2001
From: hellices <hellices@naver.com>
Date: Fri, 7 Aug 2026 01:49:12 +0900
Subject: [PATCH 02/12] fix: tighten service analysis ownership checks

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 src/korvid/core/service_analysis.py | 18 +++++++++---
 tests/core/test_service_analysis.py | 43 +++++++++++++++++++++++++++++
 2 files changed, 57 insertions(+), 4 deletions(-)

diff --git a/src/korvid/core/service_analysis.py b/src/korvid/core/service_analysis.py
index 542766e8..659c64dd 100644
--- a/src/korvid/core/service_analysis.py
+++ b/src/korvid/core/service_analysis.py
@@ -118,7 +118,12 @@ def analyze_service_endpoints(
 
     matching = tuple(
         sorted(
-            (item for item in slices if item.service_name == service.identity.name),
+            (
+                item
+                for item in slices
+                if item.service_name == service.identity.name
+                and item.identity.namespace == service.identity.namespace
+            ),
             key=_slice_sort_key,
         )
     )
@@ -150,7 +155,7 @@ def _partition_current(
     current: list[EndpointSliceSnapshot] = []
     stale: list[EndpointSliceSnapshot] = []
     for item in slices:
-        if service_uid and item.owner_uids and service_uid not in item.owner_uids:
+        if item.owner_uids and (not service_uid or service_uid not in item.owner_uids):
             stale.append(item)
         else:
             current.append(item)
@@ -264,8 +269,13 @@ def _not_applicable_report(primary: ResourceIdentity) -> AnalysisReport:
 
 def _stale_reason(service: ServiceSnapshot, stale: Sequence[EndpointSliceSnapshot]) -> str:
     count = len(stale)
-    noun = "slice" if count == 1 else "slices"
-    return f"{count} EndpointSlice {noun} are owned by a different Service UID than {service.identity.uid!r}."
+    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]:
diff --git a/tests/core/test_service_analysis.py b/tests/core/test_service_analysis.py
index 5caca9da..e7e4659f 100644
--- a/tests/core/test_service_analysis.py
+++ b/tests/core/test_service_analysis.py
@@ -81,6 +81,49 @@ def test_replaced_service_slice_is_stale_not_healthy() -> None:
     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(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", owner_uids=("old-uid",)),
+            _slice(name="web-2", uid="slice-2", 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", 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:

From 6c7e98f2b3657be6dfba0a77bdd1262a4ae1160a Mon Sep 17 00:00:00 2001
From: hellices <hellices@naver.com>
Date: Fri, 7 Aug 2026 01:56:08 +0900
Subject: [PATCH 03/12] feat: project endpoint slice readiness

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 src/korvid/k8s/models.py           | 40 +++++++++++++++++++
 src/korvid/tools/executor.py       | 10 +++++
 tests/k8s/test_models.py           | 64 ++++++++++++++++++++++++++++++
 tests/tools/test_list_resources.py | 18 +++++++++
 4 files changed, 132 insertions(+)

diff --git a/src/korvid/k8s/models.py b/src/korvid/k8s/models.py
index 4b049162..f1aa76eb 100644
--- a/src/korvid/k8s/models.py
+++ b/src/korvid/k8s/models.py
@@ -327,6 +327,43 @@ 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
+
+    @classmethod
+    def from_manifest(cls, kind: str, manifest: dict[str, Any]) -> EndpointSliceSummary:
+        base = GenericSummary.from_manifest(kind, manifest)
+        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,
+        )
+
+
+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 +410,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/"
 
@@ -423,6 +461,8 @@ def summary_for(kind: str, manifest: dict[str, Any]) -> GenericSummary:
     if kind == "ReplicaSet":
         return ReplicaSetSummary.from_manifest(kind, manifest)
     api_version = str(manifest.get("apiVersion") or "")
+    if kind == "EndpointSlice" and api_version.startswith(_DISCOVERY_GROUP_PREFIX):
+        return EndpointSliceSummary.from_manifest(kind, manifest)
     if kind == "PackageManifest" and api_version.startswith(_PACKAGES_GROUP_PREFIX):
         return PackageManifestSummary.from_manifest(kind, manifest)
     if api_version.startswith(_OLM_GROUP_PREFIX):
diff --git a/src/korvid/tools/executor.py b/src/korvid/tools/executor.py
index 8cdef1e5..1af1ed3b 100644
--- a/src/korvid/tools/executor.py
+++ b/src/korvid/tools/executor.py
@@ -22,6 +22,7 @@
 from korvid.k8s.helm import HelmReleaseSummary, HelmRevisionSummary
 from korvid.k8s.models import (
     CSVSummary,
+    EndpointSliceSummary,
     GenericSummary,
     OLMSubscriptionSummary,
     PackageManifestSummary,
@@ -254,6 +255,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 +298,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.
diff --git a/tests/k8s/test_models.py b/tests/k8s/test_models.py
index 511b9500..bf406402 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": {
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

From 5da1755fdc5c277becd984fe6ff99ddbea79fcba Mon Sep 17 00:00:00 2001
From: hellices <hellices@naver.com>
Date: Fri, 7 Aug 2026 02:08:59 +0900
Subject: [PATCH 04/12] feat: expose deterministic service diagnosis

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 docs/agent.md                |  13 ++--
 docs/mcp.md                  |   7 ++
 src/korvid/tools/executor.py |  85 ++++++++++++++++++++
 src/korvid/tools/follow.py   |  13 ++++
 src/korvid/tools/registry.py |  29 +++++++
 tests/tools/test_executor.py | 147 ++++++++++++++++++++++++++++++++++-
 tests/tools/test_follow.py   |  12 +++
 tests/tools/test_registry.py |  10 ++-
 8 files changed, 309 insertions(+), 7 deletions(-)

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..f6061d89 100644
--- a/docs/mcp.md
+++ b/docs/mcp.md
@@ -39,6 +39,12 @@ 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.
 - **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 +105,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/tools/executor.py b/src/korvid/tools/executor.py
index 1af1ed3b..408f60ca 100644
--- a/src/korvid/tools/executor.py
+++ b/src/korvid/tools/executor.py
@@ -17,6 +17,13 @@
     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
@@ -1045,6 +1052,10 @@ 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:
@@ -1546,6 +1557,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,
+        owner_uids=item.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/tools/test_executor.py b/tests/tools/test_executor.py
index 8bc2128d..6e2159f0 100644
--- a/tests/tools/test_executor.py
+++ b/tests/tools/test_executor.py
@@ -14,7 +14,7 @@
 from korvid.k8s.discovery import PODS_META
 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",
     ]
 
 
@@ -3072,3 +3073,147 @@ 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, ...] = (),
+    ready_endpoints: int = 0,
+) -> EndpointSliceSummary:
+    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,
+    )
+
+
+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=("old-uid",),
+                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
+    assert load_structured_document(text)["outcome"] == "incomplete"
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_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

From 95daef3dc0c25becdd6325ec98411adf787a9d6b Mon Sep 17 00:00:00 2001
From: hellices <hellices@naver.com>
Date: Fri, 7 Aug 2026 02:30:34 +0900
Subject: [PATCH 05/12] fix: derive runtime ceiling from compact read tools
 schema

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 tests/agent/test_runtime.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/tests/agent/test_runtime.py b/tests/agent/test_runtime.py
index 07dca703..67e5cd18 100644
--- a/tests/agent/test_runtime.py
+++ b/tests/agent/test_runtime.py
@@ -1505,6 +1505,8 @@ async def test_over_ceiling_request_drops_the_oldest_turn_and_still_reaches_the_
     Trimming the oldest retained turn shrinks the same conversation until
     it fits, so a long session keeps working. The current prompt is never
     the thing that gets dropped (issue #189)."""
+    compact_read_tools_chars = len(json.dumps(READ_TOOLS, separators=(",", ":")))
+    non_tool_request_budget = 7_500
     provider = ScriptedProvider(
         [
             [{"type": "text_delta", "text": "first answer"}, {"type": "done"}],
@@ -1515,7 +1517,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=compact_read_tools_chars + non_tool_request_budget,
     )
 
     first = await collect(runtime, "a" * 6_000)

From 261448ebc8cc914739c52f1bee149999477cfb2a Mon Sep 17 00:00:00 2001
From: hellices <hellices@naver.com>
Date: Fri, 7 Aug 2026 02:41:47 +0900
Subject: [PATCH 06/12] fix: harden runtime request-ceiling tests

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 tests/agent/test_runtime.py | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 deletions(-)

diff --git a/tests/agent/test_runtime.py b/tests/agent/test_runtime.py
index 67e5cd18..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")
@@ -1505,8 +1509,6 @@ async def test_over_ceiling_request_drops_the_oldest_turn_and_still_reaches_the_
     Trimming the oldest retained turn shrinks the same conversation until
     it fits, so a long session keeps working. The current prompt is never
     the thing that gets dropped (issue #189)."""
-    compact_read_tools_chars = len(json.dumps(READ_TOOLS, separators=(",", ":")))
-    non_tool_request_budget = 7_500
     provider = ScriptedProvider(
         [
             [{"type": "text_delta", "text": "first answer"}, {"type": "done"}],
@@ -1517,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=compact_read_tools_chars + non_tool_request_budget,
+        max_request_chars=_read_tools_request_ceiling(10_000),
     )
 
     first = await collect(runtime, "a" * 6_000)
@@ -1649,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

From dc587e0a99d1f3b0a53642efb851ac101f32cd48 Mon Sep 17 00:00:00 2001
From: hellices <hellices@naver.com>
Date: Fri, 7 Aug 2026 02:42:14 +0900
Subject: [PATCH 07/12] docs: update task 4 report

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 .superpowers/sdd/task-4-report.md | 26 ++++++++++++++++++++++++++
 1 file changed, 26 insertions(+)
 create mode 100644 .superpowers/sdd/task-4-report.md

diff --git a/.superpowers/sdd/task-4-report.md b/.superpowers/sdd/task-4-report.md
new file mode 100644
index 00000000..469695dd
--- /dev/null
+++ b/.superpowers/sdd/task-4-report.md
@@ -0,0 +1,26 @@
+# Task 4 Report
+
+## RED evidence
+- `uv run pytest -p no:tach tests/agent/test_runtime.py -k 'test_over_ceiling_request_drops_the_oldest_turn_and_still_reaches_the_model or test_estimated_prompt_cost_reflects_the_history_actually_sent'`
+  - Result: `1 failed, 1 passed, 159 deselected`
+  - Failure: `IndexError: list index out of range` in `test_estimated_prompt_cost_reflects_the_history_actually_sent` because the first 6,000-char prompt was blocked before a second provider call existed.
+- Root cause: the test still used `max_request_chars=12_000`, which was now too small once the compact `READ_TOOLS` schema grew.
+
+## GREEN evidence
+- Added `_read_tools_request_ceiling(non_tool_request_budget)` so both 6,000+6,000 trimming tests derive the ceiling from compact `READ_TOOLS` JSON plus a non-tool budget.
+- Switched both tests to `_read_tools_request_ceiling(10_000)`, which leaves headroom for one prompt but still forces the second request to trim history.
+- In the estimated-cost test, kept both collected event lists and asserted neither turn emitted `AgentError` before indexing `provider.calls[1]`.
+
+## Exact commands and results
+- `uv run pytest -p no:tach tests/agent/test_runtime.py -k 'test_over_ceiling_request_drops_the_oldest_turn_and_still_reaches_the_model or test_estimated_prompt_cost_reflects_the_history_actually_sent'`
+  - Result: `2 passed, 159 deselected`
+- `uv run ruff check tests/agent/test_runtime.py && uv run ruff format --check tests/agent/test_runtime.py`
+  - Result: `All checks passed!` / `1 file already formatted`
+- `git commit -m "fix: harden runtime request-ceiling tests" -m "Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"`
+  - Result: commit `261448e`
+
+## Self-review
+- Change is limited to `tests/agent/test_runtime.py` and this report.
+- The ceiling is now derived in one helper instead of duplicated local calculations.
+- No production code or tool surface was changed.
+- The two targeted tests and Ruff checks passed.

From 89f80a6e4be4df7f03883487a56db4649c9405e7 Mon Sep 17 00:00:00 2001
From: hellices <hellices@naver.com>
Date: Fri, 7 Aug 2026 03:15:30 +0900
Subject: [PATCH 08/12] fix: pass authoritative group to summary_for; harden
 _meta_for_kind_name

- summary_for gains optional keyword-only group: str | None = None.
  For EndpointSlice, when group is provided it takes precedence over
  apiVersion, so native LIST items that omit TypeMeta dispatch correctly
  to EndpointSliceSummary when group == 'discovery.k8s.io'. A provided
  non-discovery group stays GenericSummary even if apiVersion claims
  discovery. When group is absent the existing apiVersion-prefix fallback
  is unchanged (direct callers / tests unaffected).

- KubeClient._object_summary passes meta.group so every list_objects /
  watch path uses the authoritative group.

- fake_kube.py and test_executor.py's inline fake list_objects both pass
  meta.group, mirroring production.

- _meta_for_kind_name: for kinds in _DIAGNOSE_BUILTIN_METAS, only
  considers discovered aliases whose group equals the builtin's group.
  A same-kind CRD from a different group can no longer shadow the stable
  builtin (fixes false no_endpoint_slices when a CRD is named
  EndpointSlice in a different API group).

- test_diagnose_service_result_remains_bounded_yaml: slices are now
  current (owner_uids='svc-1'); asserts outcome==healthy, size bound
  respected, and elision marker present.

- New RED->GREEN tests:
  * summary_for TypeMeta-less EndpointSlice + group='discovery.k8s.io'
    -> EndpointSliceSummary
  * summary_for group='example.io' -> GenericSummary even with
    apiVersion='discovery.k8s.io/v1'
  * KubeClient.list_objects with no-apiVersion raw item -> typed summary
  * _meta_for_kind_name ignores wrong-group CRD alias
  * End-to-end: raw TypeMeta-less list items -> healthy diagnosis

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 src/korvid/evals/fake_kube.py |  2 +-
 src/korvid/k8s/client.py      |  2 +-
 src/korvid/k8s/models.py      | 27 +++++++++++--
 src/korvid/tools/executor.py  | 20 +++++++--
 tests/k8s/test_client.py      | 39 ++++++++++++++++++
 tests/k8s/test_models.py      | 38 ++++++++++++++++++
 tests/tools/test_executor.py  | 76 +++++++++++++++++++++++++++++++++--
 7 files changed, 191 insertions(+), 13 deletions(-)

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 f1aa76eb..004a305e 100644
--- a/src/korvid/k8s/models.py
+++ b/src/korvid/k8s/models.py
@@ -454,15 +454,34 @@ 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 == "EndpointSlice" and api_version.startswith(_DISCOVERY_GROUP_PREFIX):
-        return EndpointSliceSummary.from_manifest(kind, manifest)
     if kind == "PackageManifest" and api_version.startswith(_PACKAGES_GROUP_PREFIX):
         return PackageManifestSummary.from_manifest(kind, manifest)
     if api_version.startswith(_OLM_GROUP_PREFIX):
diff --git a/src/korvid/tools/executor.py b/src/korvid/tools/executor.py
index 408f60ca..c576fd25 100644
--- a/src/korvid/tools/executor.py
+++ b/src/korvid/tools/executor.py
@@ -1060,12 +1060,26 @@ async def _get_events(self, args: dict[str, Any]) -> str:
 
     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."""
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 bf406402..060cd4f4 100644
--- a/tests/k8s/test_models.py
+++ b/tests/k8s/test_models.py
@@ -380,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)
diff --git a/tests/tools/test_executor.py b/tests/tools/test_executor.py
index 6e2159f0..2498e17c 100644
--- a/tests/tools/test_executor.py
+++ b/tests/tools/test_executor.py
@@ -11,7 +11,7 @@
 
 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 EndpointSliceSummary, GenericSummary, summary_for
@@ -1188,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(
@@ -3206,7 +3206,7 @@ async def test_diagnose_service_result_remains_bounded_yaml() -> None:
         slices=[
             _endpoint_slice_summary(
                 name=f"api-{index:05d}",
-                owner_uids=("old-uid",),
+                owner_uids=("svc-1",),
                 ready_endpoints=1,
             )
             for index in range(2_000)
@@ -3216,4 +3216,72 @@ async def test_diagnose_service_result_remains_bounded_yaml() -> None:
         "diagnose_service", {"service": "api", "namespace": "shop"}
     )
     assert len(text) <= MAX_RESULT_CHARS
-    assert load_structured_document(text)["outcome"] == "incomplete"
+    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"

From 6272705ecf2fa76986be35c46e4a63eda45e6041 Mon Sep 17 00:00:00 2001
From: hellices <hellices@naver.com>
Date: Fri, 7 Aug 2026 03:40:06 +0900
Subject: [PATCH 09/12] chore: exclude SDD scratch report

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 .superpowers/sdd/task-4-report.md | 26 --------------------------
 1 file changed, 26 deletions(-)
 delete mode 100644 .superpowers/sdd/task-4-report.md

diff --git a/.superpowers/sdd/task-4-report.md b/.superpowers/sdd/task-4-report.md
deleted file mode 100644
index 469695dd..00000000
--- a/.superpowers/sdd/task-4-report.md
+++ /dev/null
@@ -1,26 +0,0 @@
-# Task 4 Report
-
-## RED evidence
-- `uv run pytest -p no:tach tests/agent/test_runtime.py -k 'test_over_ceiling_request_drops_the_oldest_turn_and_still_reaches_the_model or test_estimated_prompt_cost_reflects_the_history_actually_sent'`
-  - Result: `1 failed, 1 passed, 159 deselected`
-  - Failure: `IndexError: list index out of range` in `test_estimated_prompt_cost_reflects_the_history_actually_sent` because the first 6,000-char prompt was blocked before a second provider call existed.
-- Root cause: the test still used `max_request_chars=12_000`, which was now too small once the compact `READ_TOOLS` schema grew.
-
-## GREEN evidence
-- Added `_read_tools_request_ceiling(non_tool_request_budget)` so both 6,000+6,000 trimming tests derive the ceiling from compact `READ_TOOLS` JSON plus a non-tool budget.
-- Switched both tests to `_read_tools_request_ceiling(10_000)`, which leaves headroom for one prompt but still forces the second request to trim history.
-- In the estimated-cost test, kept both collected event lists and asserted neither turn emitted `AgentError` before indexing `provider.calls[1]`.
-
-## Exact commands and results
-- `uv run pytest -p no:tach tests/agent/test_runtime.py -k 'test_over_ceiling_request_drops_the_oldest_turn_and_still_reaches_the_model or test_estimated_prompt_cost_reflects_the_history_actually_sent'`
-  - Result: `2 passed, 159 deselected`
-- `uv run ruff check tests/agent/test_runtime.py && uv run ruff format --check tests/agent/test_runtime.py`
-  - Result: `All checks passed!` / `1 file already formatted`
-- `git commit -m "fix: harden runtime request-ceiling tests" -m "Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>"`
-  - Result: commit `261448e`
-
-## Self-review
-- Change is limited to `tests/agent/test_runtime.py` and this report.
-- The ceiling is now derived in one helper instead of duplicated local calculations.
-- No production code or tool surface was changed.
-- The two targeted tests and Ruff checks passed.

From fac44a02e435057deb2cfa2195a9998ba5ffbf2d Mon Sep 17 00:00:00 2001
From: hellices <hellices@naver.com>
Date: Fri, 7 Aug 2026 04:09:22 +0900
Subject: [PATCH 10/12] fix: scope EndpointSlice stale-owner checks to core/v1
 Service refs only
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

An EndpointSlice owned only by a custom CRD controller was incorrectly
flagged stale because _partition_current checked owner_uids (all owner
refs) against the Service UID.

Changes:
- Add _service_owner_uids() to models.py: extracts UIDs only from
  ownerReferences where kind=='Service' and apiVersion=='v1'.
- Add EndpointSliceSummary.service_owner_uids: tuple[str, ...] populated
  by the new helper; generic owner_uids is preserved for relation logic.
- Rename EndpointSliceSnapshot.owner_uids -> service_owner_uids so the
  snapshot field clearly carries only Service UIDs.
- Update _partition_current, _confidence_for_healthy, _current_evidence
  and _endpoint_slice_snapshot to use service_owner_uids throughout.
- Update all test helpers and call sites consistently.

Tested:
- Unrelated custom-controller owner + ready endpoints → healthy
- Mismatching service_owner_uids → incomplete (stale-owner gap)
- Mixed custom+Service refs: generic owner_uids has all UIDs;
  service_owner_uids has only the Service UID
- Wrong apiVersion 'Service' ref excluded from service_owner_uids
- Full gate: 4113 passed, 21 skipped

Fixes PR #212 review finding.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 src/korvid/core/service_analysis.py | 22 +++++--
 src/korvid/k8s/models.py            | 14 +++++
 src/korvid/tools/executor.py        |  2 +-
 tests/core/test_service_analysis.py | 96 ++++++++++++++++++++++++++---
 tests/k8s/test_models.py            | 69 +++++++++++++++++++++
 tests/tools/test_executor.py        | 37 +++++++++++
 6 files changed, 224 insertions(+), 16 deletions(-)

diff --git a/src/korvid/core/service_analysis.py b/src/korvid/core/service_analysis.py
index 659c64dd..1223425c 100644
--- a/src/korvid/core/service_analysis.py
+++ b/src/korvid/core/service_analysis.py
@@ -72,7 +72,7 @@ class EndpointSliceSnapshot:
 
     identity: ResourceIdentity
     service_name: str
-    owner_uids: tuple[str, ...]
+    service_owner_uids: tuple[str, ...]
     address_type: str
     endpoints: int
     ready_endpoints: int
@@ -155,7 +155,9 @@ def _partition_current(
     current: list[EndpointSliceSnapshot] = []
     stale: list[EndpointSliceSnapshot] = []
     for item in slices:
-        if item.owner_uids and (not service_uid or service_uid not in item.owner_uids):
+        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)
@@ -177,8 +179,14 @@ def _current_evidence(slices: Sequence[EndpointSliceSnapshot]) -> tuple[Evidence
         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.owner_uids:
-            items.append(Evidence(item.identity, "endpoints.owner_uids", ",".join(item.owner_uids)))
+        if item.service_owner_uids:
+            items.append(
+                Evidence(
+                    item.identity,
+                    "endpoints.service_owner_uids",
+                    ",".join(item.service_owner_uids),
+                )
+            )
     return tuple(items)
 
 
@@ -187,7 +195,11 @@ def _confidence_for_healthy(
 ) -> Confidence:
     if not service.identity.uid:
         return "medium"
-    if any(service.identity.uid in item.owner_uids for item in slices if item.owner_uids):
+    if any(
+        service.identity.uid in item.service_owner_uids
+        for item in slices
+        if item.service_owner_uids
+    ):
         return "high"
     return "medium"
 
diff --git a/src/korvid/k8s/models.py b/src/korvid/k8s/models.py
index 004a305e..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")
@@ -335,10 +344,14 @@ class EndpointSliceSummary(GenericSummary):
     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))
@@ -348,6 +361,7 @@ def from_manifest(cls, kind: str, manifest: dict[str, Any]) -> EndpointSliceSumm
             address_type=str(manifest.get("addressType") or ""),
             endpoints=len(endpoints),
             ready_endpoints=ready_endpoints,
+            service_owner_uids=_service_owner_uids(meta),
         )
 
 
diff --git a/src/korvid/tools/executor.py b/src/korvid/tools/executor.py
index c576fd25..960b1c54 100644
--- a/src/korvid/tools/executor.py
+++ b/src/korvid/tools/executor.py
@@ -1634,7 +1634,7 @@ def _endpoint_slice_snapshot(item: EndpointSliceSummary) -> EndpointSliceSnapsho
     return EndpointSliceSnapshot(
         identity=ResourceIdentity("EndpointSlice", item.namespace, item.name, item.uid),
         service_name=item.service_name,
-        owner_uids=item.owner_uids,
+        service_owner_uids=item.service_owner_uids,
         address_type=item.address_type,
         endpoints=item.endpoints,
         ready_endpoints=item.ready_endpoints,
diff --git a/tests/core/test_service_analysis.py b/tests/core/test_service_analysis.py
index e7e4659f..8686f658 100644
--- a/tests/core/test_service_analysis.py
+++ b/tests/core/test_service_analysis.py
@@ -34,7 +34,7 @@ def _slice(
     name: str = "web-1",
     uid: str = "",
     service_name: str = "web",
-    owner_uids: tuple[str, ...] = (),
+    service_owner_uids: tuple[str, ...] = (),
     address_type: str = "IPv4",
     endpoints: int = 1,
     ready_endpoints: int = 1,
@@ -42,7 +42,7 @@ def _slice(
     return EndpointSliceSnapshot(
         identity=ResourceIdentity(kind=kind, namespace=namespace, name=name, uid=uid),
         service_name=service_name,
-        owner_uids=owner_uids,
+        service_owner_uids=service_owner_uids,
         address_type=address_type,
         endpoints=endpoints,
         ready_endpoints=ready_endpoints,
@@ -52,7 +52,7 @@ def _slice(
 def test_ready_current_slice_is_healthy() -> None:
     report = analyze_service_endpoints(
         _service(uid="svc-1"),
-        (_slice(owner_uids=("svc-1",), endpoints=2, ready_endpoints=1),),
+        (_slice(service_owner_uids=("svc-1",), endpoints=2, ready_endpoints=1),),
     )
     assert report.outcome == "healthy"
     assert report.findings == ()
@@ -76,7 +76,7 @@ def test_unavailable_slice_evidence_never_reports_healthy() -> None:
 def test_replaced_service_slice_is_stale_not_healthy() -> None:
     report = analyze_service_endpoints(
         _service(uid="new-uid"),
-        (_slice(owner_uids=("old-uid",), endpoints=1, ready_endpoints=1),),
+        (_slice(service_owner_uids=("old-uid",), endpoints=1, ready_endpoints=1),),
     )
     assert report.outcome == "incomplete"
     assert report.findings == ()
@@ -90,7 +90,7 @@ def test_replaced_service_slice_is_stale_not_healthy() -> None:
 def test_owned_slice_with_missing_service_uid_is_stale() -> None:
     report = analyze_service_endpoints(
         _service(),
-        (_slice(owner_uids=("other",), endpoints=1, ready_endpoints=1),),
+        (_slice(service_owner_uids=("other",), endpoints=1, ready_endpoints=1),),
     )
     assert report.outcome == "incomplete"
     assert report.findings == ()
@@ -105,8 +105,8 @@ def test_stale_owner_reason_uses_plural_grammar() -> None:
     report = analyze_service_endpoints(
         _service(uid="new-uid"),
         (
-            _slice(name="web-1", uid="slice-1", owner_uids=("old-uid",)),
-            _slice(name="web-2", uid="slice-2", owner_uids=("old-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"
@@ -120,7 +120,7 @@ def test_stale_owner_reason_uses_plural_grammar() -> None:
 def test_matching_requires_namespace_and_service_name() -> None:
     report = analyze_service_endpoints(
         _service(uid="svc-1", namespace="default"),
-        (_slice(namespace="other", owner_uids=("svc-1",), endpoints=1, ready_endpoints=1),),
+        (_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"
@@ -129,7 +129,7 @@ def test_matching_requires_namespace_and_service_name() -> None:
 def test_current_slices_without_ready_endpoints_warn() -> None:
     report = analyze_service_endpoints(
         _service(uid="svc-1"),
-        (_slice(owner_uids=("svc-1",), endpoints=2, ready_endpoints=0),),
+        (_slice(service_owner_uids=("svc-1",), endpoints=2, ready_endpoints=0),),
     )
     assert report.findings[0].rule_id == "service.no_ready_endpoints"
 
@@ -137,7 +137,7 @@ def test_current_slices_without_ready_endpoints_warn() -> None:
 def test_selectorless_unowned_slice_is_valid_manual_evidence() -> None:
     report = analyze_service_endpoints(
         _service(uid="svc-1", selector=()),
-        (_slice(owner_uids=(), endpoints=1, ready_endpoints=1),),
+        (_slice(service_owner_uids=(), endpoints=1, ready_endpoints=1),),
     )
     assert report.outcome == "healthy"
 
@@ -158,3 +158,79 @@ def test_document_uses_stable_public_keys() -> None:
         "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_models.py b/tests/k8s/test_models.py
index 060cd4f4..f796c8aa 100644
--- a/tests/k8s/test_models.py
+++ b/tests/k8s/test_models.py
@@ -1333,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 2498e17c..64a184cb 100644
--- a/tests/tools/test_executor.py
+++ b/tests/tools/test_executor.py
@@ -3115,8 +3115,12 @@ def _service_manifest(uid: str = "") -> dict[str, Any]:
 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",
@@ -3129,6 +3133,7 @@ def _endpoint_slice_summary(
         address_type="IPv4",
         endpoints=1,
         ready_endpoints=ready_endpoints,
+        service_owner_uids=resolved_service_owner_uids,
     )
 
 
@@ -3285,3 +3290,35 @@ async def test_diagnose_service_typemeta_less_slices_return_healthy() -> None:
         )
     )
     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"

From 00aa4dc43e5ba7b8764f2ee61264588febd3f9be Mon Sep 17 00:00:00 2001
From: hellices <hellices@naver.com>
Date: Fri, 7 Aug 2026 04:39:35 +0900
Subject: [PATCH 11/12] ci: validate review fixes

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

From d2eef513ff42e397b748c5c3bee2877d1444c956 Mon Sep 17 00:00:00 2001
From: hellices <hellices@naver.com>
Date: Fri, 7 Aug 2026 05:13:28 +0900
Subject: [PATCH 12/12] docs: clarify EndpointSlice owner handling

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 docs/mcp.md | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/docs/mcp.md b/docs/mcp.md
index f6061d89..889c74b9 100644
--- a/docs/mcp.md
+++ b/docs/mcp.md
@@ -45,6 +45,8 @@ format:
   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