Fix CT hosts disappearing after degraded CNAME enrichment

.gitignore

@@ -73,6 +73,7 @@ docs/accuracy.md
 !/validation/2026-08-21-statuspage-fingerprint-review.md
 !/validation/2026-08-21-zendesk-fingerprint-review.md
 !/validation/2026-08-21-tencent-edgeone-fingerprint-review.md
+!/validation/2026-08-31-ct-related-domain-degradation.md
 !/validation/2026-08-30-cloudflare-fingerprint-review.md
 # Generic catalog-round frame preparer; no target data is embedded.
 !/validation/prepare_catalog_round.py

CHANGELOG.md

@@ -26,6 +26,34 @@ operator, corporate group, ownership, or control.
 
 ## [Unreleased]
 
+## [2.18.1] - 2026-08-31
+
+v2.18.1 restores CT-discovered related hosts when later CNAME enrichment
+degrades, without exposing CNAME-dependent classification or changing the
+stable lookup schema.
+
+### Tool Surface Changes
+
+Tool surface changes: no CLI command or flag changes.
+
+### Fixed
+
+- **CT related hosts survive degraded CNAME enrichment.** A successful live,
+  partial, or cached CT result now retains its exact related-domain names when
+  the later `dns:cname` classification channel is unavailable. CNAME-owned
+  attributions and motifs still fail closed, lexical observations are rebuilt
+  from the retained CT names, and lookup, Markdown, panel, MCP, cache, and
+  NamespaceReviewBundle surfaces now agree. Existing cache entries without the
+  additive CT name-provenance field remain conservatively empty under this
+  degraded condition.
+
+### Tests
+
+- Added synthetic regression coverage for live, partial, and cached CT
+  provenance; source and merged projection; legacy caches; stable JSON;
+  briefing and full renderers; connection-map classification; MCP JSON; and
+  NamespaceReviewBundle embedding.
+
 ## [2.18.0] - 2026-08-30
 
 NamespaceReviewBundle v1 adds a role-neutral, caller-owned handoff for one

ROADMAP.md

@@ -108,6 +108,12 @@ After v2.18.0 publication, the next steps are deliberately bounded:
 4. Keep the 29-family claim audit, both MCP SDK pins, full test gate, and release
    provenance checks blocking on every change.
 
+v2.18.1 addresses the first stability-soak defect: a degraded CNAME
+classification pass no longer erases names contributed by a separately
+successful CT path. The fix records exact CT ownership internally, preserves
+the briefing versus full-map split, and leaves CNAME-dependent classification
+fail closed.
+
 No v2.19 feature is scheduled. New product work must be justified by a concrete
 operator handoff that cannot be solved through the stable surfaces above.
 

docs/roadmap.md

@@ -179,6 +179,12 @@ MCP SDK pins, the full test gate, and release provenance blocking. No v2.19
 feature is scheduled without a named operator handoff that the stable surfaces
 cannot solve.
 
+v2.18.1 addresses the first stability-soak defect: a degraded CNAME
+classification pass no longer erases names contributed by a separately
+successful CT path. Exact CT name provenance is internal and cache-persistent;
+the stable lookup schema, briefing versus full-map split, and fail-closed
+CNAME-dependent classifications remain unchanged.
+
 Agent portability is a separate maintainer track, not a version-path
 milestone: its representative task, client, measure, privacy, and stop-rule
 contract is frozen, its protected-main prerequisite and the candidate's

docs/schema.md

@@ -226,7 +226,7 @@ table above. Field order in emitted JSON is not guaranteed; use the key name.
 |---|---|---|---|---|---|
 | `domain_count` | int | no | `0+` | stable | Number of domain strings retained from bounded public tenant-discovery responses. It is not organization size or guaranteed tenant cardinality. |
 | `tenant_domains` | `list[string]` | no | n/a | stable | Domain strings retained from bounded public tenant-discovery responses. The list may be incomplete and does not establish ownership or an exhaustive tenant namespace. |
-| `related_domains` | `list[string]` | no | n/a | stable | Domain names linked by bounded CT, CNAME, Exchange/identity endpoint, autodiscover, or DKIM tenant-domain breadcrumbs. The stable field name does not imply ownership or an organizational relationship. |
+| `related_domains` | `list[string]` | no | n/a | stable | Domain names linked by bounded CT, CNAME, Exchange/identity endpoint, autodiscover, or DKIM tenant-domain breadcrumbs. A degraded CNAME-classification pass does not erase names retained from a separately successful CT path; `surface_attributions` can therefore be empty while this list is populated. The stable field name does not imply ownership or an organizational relationship. |
 | `surface_attributions` | `list[SurfaceAttribution]` | no | n/a | stable | Per-subdomain attribution of each related domain to a SaaS or infrastructure provider, from CNAME-chain classification. Empty when nothing classified. See the [`SurfaceAttribution`](#surfaceattribution) nested object. |
 
 ### Email security

scripts/check_schema_sources.py

@@ -33,6 +33,7 @@
 
 INTENTIONAL_TENANTINFO_OMISSIONS: Mapping[str, str] = {
     "cached_at": "cache metadata, not emitted by the lookup JSON formatter",
+    "ct_related_domains": "internal CT name provenance, not emitted by the lookup JSON formatter",
     "dns_catalog_summaries": "opt-in private catalog diagnostic, excluded from the stable schema",
     "dmarc_testing": "internal RFC 9989 effective-policy input, not a stable output field",
     "insight_claims": "internal generation-time lineage, exposed only through explain-mode projections",

src/recon_tool/cache.py

@@ -334,6 +334,7 @@ def tenant_info_to_dict(info: TenantInfo) -> dict[str, Any]:
         "likely_primary_email_provider": info.likely_primary_email_provider,
         "ct_provider_used": info.ct_provider_used,
         "ct_subdomain_count": info.ct_subdomain_count,
+        "ct_related_domains": list(info.ct_related_domains),
         "ct_cache_age_days": info.ct_cache_age_days,
         "ct_attempt_outcome": info.ct_attempt_outcome,
         "merge_conflicts": (
@@ -951,6 +952,7 @@ def tenant_info_from_dict(data: dict[str, Any]) -> TenantInfo:
         ),
         ct_provider_used=_optional_cache_string(data.get("ct_provider_used"), "ct_provider_used"),
         ct_subdomain_count=_cache_count(data.get("ct_subdomain_count"), "ct_subdomain_count"),
+        ct_related_domains=cache_string_tuple(data.get("ct_related_domains", []), "ct_related_domains"),
         ct_cache_age_days=_optional_cache_count(data.get("ct_cache_age_days"), "ct_cache_age_days"),
         ct_attempt_outcome=_optional_cache_string(data.get("ct_attempt_outcome"), "ct_attempt_outcome"),
         slug_confidences=_read_slug_confidences(data.get("slug_confidences")),

src/recon_tool/collection_view.py

@@ -173,6 +173,11 @@
 _IDENTITY_SOURCE_MARKERS = (
     _USER_REALM_MARKERS | _AUTODISCOVER_MARKERS | _OIDC_MARKERS | _GOOGLE_IDENTITY_MARKERS | _GOOGLE_WORKSPACE_MARKERS
 )
+_LEXICAL_INSIGHT_CATEGORIES = (
+    "Environment-like Labels",
+    "Region-like Labels",
+    "Tenant-like Labels",
+)
 _IDENTITY_EVIDENCE_RULES = (
     (_USER_REALM_MARKERS, frozenset({"GetUserRealm"})),
     (_AUTODISCOVER_MARKERS, frozenset({"Autodiscover"})),
@@ -193,6 +198,44 @@ def auth_type_channel_unavailable(degraded_sources: Iterable[str]) -> bool:
     return not frozenset(degraded_sources).isdisjoint(_USER_REALM_MARKERS)
 
 
+def _observable_related_domains(
+    info: TenantInfo | SourceResult,
+    status: SourceStatus,
+) -> tuple[str, ...]:
+    """Retain only related names with an available discovery path.
+
+    ``related_domains`` mixes CT discovery with later CNAME enrichment. A
+    failed CNAME channel therefore cannot own the whole field. Current
+    collector output carries the exact CT subset separately; older cache
+    entries without that provenance retain the conservative empty projection.
+    """
+    if status.whole_dns_unavailable:
+        return ()
+    if status.channel_available("cname"):
+        return info.related_domains
+    return tuple(sorted(set(info.related_domains).intersection(info.ct_related_domains)))
+
+
+def _project_lexical_related_insights(
+    info: TenantInfo,
+    related_domains: tuple[str, ...],
+    visible_insights: tuple[str, ...],
+) -> tuple[tuple[str, ...], tuple[str, ...]]:
+    """Rebuild lexical diagnostics from the related names still observable."""
+    from recon_tool.lexical import lexical_observations
+
+    raw_lexical_insights = {
+        f"{category}: {statement}"
+        for category in _LEXICAL_INSIGHT_CATEGORIES
+        for statement in info.lexical_observations
+    }
+    retained_insights = tuple(insight for insight in visible_insights if insight not in raw_lexical_insights)
+    observations = lexical_observations(related_domains, base_domain=info.queried_domain)
+    statements = tuple(observation.statement for observation in observations)
+    projected_insights = tuple(f"{observation.category}: {observation.statement}" for observation in observations)
+    return statements, retained_insights + projected_insights
+
+
 def _is_retired_numeric_domain_insight(normalized: str) -> bool:
     """Match only the three numeric domain-count formats emitted historically."""
     count, separator, claim = normalized.partition(" ")
@@ -483,11 +526,12 @@ def collection_observable_result(result: SourceResult) -> SourceResult:
         mta_sts_mode=result.mta_sts_mode if status.channel_available("mta_sts") else None,
         site_verification_tokens=(result.site_verification_tokens if status.channel_available("apex_txt") else ()),
         bimi_identity=None,
-        related_domains=(() if status.channel_unavailable("cname") else result.related_domains),
+        related_domains=_observable_related_domains(result, status),
         cert_summary=(None if status.whole_dns_unavailable else result.cert_summary),
         raw_dns_records=(() if status.whole_dns_unavailable else result.raw_dns_records),
         ct_provider_used=(None if status.whole_dns_unavailable else result.ct_provider_used),
         ct_subdomain_count=(0 if status.whole_dns_unavailable else result.ct_subdomain_count),
+        ct_related_domains=(() if status.whole_dns_unavailable else result.ct_related_domains),
         ct_cache_age_days=(None if status.whole_dns_unavailable else result.ct_cache_age_days),
         ct_attempt_outcome=(None if status.whole_dns_unavailable else result.ct_attempt_outcome),
         surface_attributions=(() if status.channel_unavailable("cname") else result.surface_attributions),
@@ -710,6 +754,15 @@ def collection_observable_info(info: TenantInfo) -> TenantInfo:
             slug_confidences=(info.slug_confidences if posterior_contract_current else ()),
         )
 
+    related_domains = _observable_related_domains(info, status)
+    lexical_observations = info.lexical_observations
+    if status.channel_unavailable("cname"):
+        lexical_observations, visible_insights = _project_lexical_related_insights(
+            info,
+            related_domains,
+            visible_insights,
+        )
+
     services = set(info.services)
     slugs = set(info.slugs)
     if status.whole_dns_unavailable:
@@ -783,13 +836,14 @@ def collection_observable_info(info: TenantInfo) -> TenantInfo:
         likely_primary_email_provider=likely_primary_email_provider,
         site_verification_tokens=info.site_verification_tokens if status.channel_available("apex_txt") else (),
         bimi_identity=None,
-        related_domains=(() if status.channel_unavailable("cname") else info.related_domains),
+        related_domains=related_domains,
         cert_summary=(None if status.whole_dns_unavailable else info.cert_summary),
         ct_provider_used=(None if status.whole_dns_unavailable else info.ct_provider_used),
         ct_subdomain_count=(0 if status.whole_dns_unavailable else info.ct_subdomain_count),
+        ct_related_domains=(() if status.whole_dns_unavailable else info.ct_related_domains),
         ct_cache_age_days=(None if status.whole_dns_unavailable else info.ct_cache_age_days),
         ct_attempt_outcome=(None if status.whole_dns_unavailable else info.ct_attempt_outcome),
-        lexical_observations=(() if status.channel_unavailable("cname") else info.lexical_observations),
+        lexical_observations=lexical_observations,
         surface_attributions=(() if status.channel_unavailable("cname") else info.surface_attributions),
         unclassified_cname_chains=(() if status.channel_unavailable("cname") else info.unclassified_cname_chains),
         chain_motifs=(() if status.channel_unavailable("cname") else info.chain_motifs),

src/recon_tool/merger.py

@@ -4,6 +4,7 @@
 
 from typing import Any, NamedTuple
 
+from recon_tool import merger_catalog
 from recon_tool.confidence import (
     compute_confidence,
     compute_evidence_confidence,
@@ -22,12 +23,6 @@
 from recon_tool.email_security import claim_safe_email_services, observed_email_control_services
 from recon_tool.insight_pipeline import build_insights_with_signals
 from recon_tool.lexical import lexical_observations
-from recon_tool.merger_catalog import (
-    dedupe_motifs,
-    dedupe_surface,
-    dedupe_unclassified,
-    merge_dns_catalog_diagnostics,
-)
 from recon_tool.models import (
     BIMIIdentity,
     CandidateValue,
@@ -598,10 +593,10 @@ def merge_results(
     detection_scores = compute_detection_scores(evidence_tuple)
     lexical_observation_statements = _append_lexical_observations(insights, all_related, queried_domain)
 
-    surface_tuple = dedupe_surface(usable_results)
-    unclassified_tuple = dedupe_unclassified(usable_results)
-    dns_catalog_summaries, unclassified_dns_observations = merge_dns_catalog_diagnostics(usable_results)
-    chain_motifs_tuple = dedupe_motifs(usable_results)
+    surface_tuple = merger_catalog.dedupe_surface(usable_results)
+    unclassified_tuple = merger_catalog.dedupe_unclassified(usable_results)
+    dns_catalog_summaries, unclassified_dns_observations = merger_catalog.merge_dns_catalog_diagnostics(usable_results)
+    chain_motifs_tuple = merger_catalog.dedupe_motifs(usable_results)
     infrastructure_clusters: InfrastructureClusterReport | None = _first_non_none(
         usable_results, "infrastructure_clusters"
     )
@@ -643,6 +638,7 @@ def merge_results(
         likely_primary_email_provider=likely_primary_email_provider,
         ct_provider_used=ct_provider_used,
         ct_subdomain_count=ct_subdomain_count,
+        ct_related_domains=merger_catalog.merge_ct_related_domains(usable_results, all_domains, queried_domain),
         ct_cache_age_days=ct_cache_age_days,
         ct_attempt_outcome=ct_attempt_outcome,
         cloud_instance=cloud_instance,

src/recon_tool/merger_catalog.py

@@ -12,6 +12,18 @@
 )
 
 
+def merge_ct_related_domains(
+    results: list[SourceResult],
+    tenant_domains: set[str],
+    queried_domain: str,
+) -> tuple[str, ...]:
+    """Merge exact CT name provenance without duplicating known tenant names."""
+    related = {domain for result in results for domain in result.ct_related_domains}
+    related.difference_update(tenant_domains)
+    related.discard(queried_domain.lower())
+    return tuple(sorted(related))
+
+
 def dedupe_surface(results: list[SourceResult]) -> tuple[SurfaceAttribution, ...]:
     """Deduplicate surface attributions by subdomain using first observation."""
     merged = {item.subdomain: item for result in reversed(results) for item in reversed(result.surface_attributions)}

src/recon_tool/models.py

@@ -626,6 +626,11 @@ class SourceResult:
     # after filtering. None when no CT provider was queried or all failed.
     ct_provider_used: str | None = None
     ct_subdomain_count: int = 0
+    # Exact related-domain names contributed by the successful CT path. This
+    # is internal collection provenance, not a stable public JSON field. It
+    # lets the reporting projection retain independently observed CT names
+    # when later CNAME enrichment is unavailable.
+    ct_related_domains: tuple[str, ...] = ()
 
     # --- CT cache fallback ---
     # When all live CT providers fail, the per-domain CT cache serves as
@@ -766,6 +771,10 @@ class TenantInfo:
     # visible. None when no CT provider succeeded.
     ct_provider_used: str | None = None
     ct_subdomain_count: int = 0
+    # Internal CT ownership for the mixed-source ``related_domains`` field.
+    # Persisted in the result cache so degraded-channel reprojection remains
+    # exact across cache reads, but deliberately omitted from lookup JSON.
+    ct_related_domains: tuple[str, ...] = ()
 
     # --- CT cache fallback ---
     # Age of the CT cache entry in days when cached data was used as a

src/recon_tool/resolver.py

@@ -75,6 +75,7 @@ def _inventory_only_related_result(result: SourceResult) -> SourceResult:
         spf_include_count=0,
         ct_provider_used=None,
         ct_subdomain_count=0,
+        ct_related_domains=(),
         ct_cache_age_days=None,
         ct_attempt_outcome=None,
         cloud_instance=None,

src/recon_tool/sources/dns.py

@@ -754,6 +754,7 @@ async def lookup(self, domain: str, **kwargs: Any) -> SourceResult:
                 dmarc_pct=ctx.dmarc_pct,
                 ct_provider_used=ctx.ct_provider_used,
                 ct_subdomain_count=ctx.ct_subdomain_count,
+                ct_related_domains=tuple(sorted(ctx.ct_related_domains)),
                 ct_cache_age_days=ctx.ct_cache_age_days,
                 ct_attempt_outcome=ctx.ct_attempt_outcome,
                 raw_dns_records=tuple(
@@ -784,6 +785,7 @@ async def lookup(self, domain: str, **kwargs: Any) -> SourceResult:
             dmarc_pct=ctx.dmarc_pct,
             ct_provider_used=ctx.ct_provider_used,
             ct_subdomain_count=ctx.ct_subdomain_count,
+            ct_related_domains=tuple(sorted(ctx.ct_related_domains)),
             ct_cache_age_days=ctx.ct_cache_age_days,
             ct_attempt_outcome=ctx.ct_attempt_outcome,
             raw_dns_records=tuple((rtype, val) for rtype, vals in sorted(ctx.raw_dns_records.items()) for val in vals),
@@ -867,6 +869,7 @@ async def _isolate(name: str, coro: Any) -> None:
 
         # Remove the queried domain itself from related_domains
         ctx.related_domains.discard(domain.lower())
+        ctx.ct_related_domains.discard(domain.lower())
 
         # Surface-attribution pass. Runs after the main gather because it
         # depends on related_domains being fully populated by CT and the

src/recon_tool/sources/dns_base.py

@@ -176,6 +176,7 @@ class DetectionCtx:
         "ct_attempt_outcome",
         "ct_cache_age_days",
         "ct_provider_used",
+        "ct_related_domains",
         "ct_subdomain_count",
         "degraded_sources",
         "dmarc_np",
@@ -234,6 +235,9 @@ def __init__(self) -> None:
         # pagination returned 87 entries". None until a provider succeeds.
         self.ct_provider_used: str | None = None
         self.ct_subdomain_count: int = 0
+        # Preserve exact CT ownership separately from the mixed-source related
+        # inventory. A later CNAME failure must not erase successful CT names.
+        self.ct_related_domains: set[str] = set()
         # CT cache age in days when cached data used as fallback
         self.ct_cache_age_days: int | None = None
         # Per-record CT attempt outcome. See ``TenantInfo.ct_attempt_outcome``

src/recon_tool/sources/dns_infra.py

@@ -654,6 +654,7 @@ async def detect_srv(ctx: dns_base.DetectionCtx, domain: str) -> None:
 def _apply_cached_cert_intel(ctx: dns_base.DetectionCtx, cached: CTCacheEntry, attribution: str) -> None:
     """Apply a CT cache entry to the context (shared by cache-first and fallback)."""
     ctx.related_domains.update(cached.subdomains)
+    ctx.ct_related_domains.update(cached.subdomains)
     if cached.cert_summary is not None:
         ctx.cert_summary = cached.cert_summary
     if cached.infrastructure_clusters is not None:
@@ -686,6 +687,7 @@ async def _query_cert_providers(ctx: dns_base.DetectionCtx, domain: str) -> tupl
             failures["rate_limit"] += 1
             ctx.degraded_sources.add(provider.name)
             ctx.related_domains.update(partial.subdomains)
+            ctx.ct_related_domains.update(partial.subdomains)
             if partial.cert_summary is not None:
                 ctx.cert_summary = partial.cert_summary
             if partial.infrastructure_clusters is not None:
@@ -727,6 +729,7 @@ async def _query_cert_providers(ctx: dns_base.DetectionCtx, domain: str) -> tupl
             continue
 
         ctx.related_domains.update(subdomains)
+        ctx.ct_related_domains.update(subdomains)
         if cert_summary is not None:
             ctx.cert_summary = cert_summary
         if infrastructure_clusters is not None: