File indexing completed on 2026-07-21 08:46:27
0001 """Publish bounded SWF System Status projections to Snapper."""
0002
0003 import json
0004 from dataclasses import dataclass
0005 from datetime import datetime
0006 from typing import Iterable, Optional
0007
0008 from django.db import transaction
0009 from django.utils import timezone
0010
0011 from snapper_ai.services import ComponentUpdate, publish_component, register_component
0012
0013 from .models import SystemStatus
0014 from .system_status import STATUS_STALE_AFTER
0015
0016
0017 PUBLISHER_IDENTITY = "swf-monitor:system-status"
0018 ASSESSMENT_POLICY_VERSION = "swf-system-status-v1"
0019 EVENT_RESOLVER = "swf-system-status-history"
0020 MAX_CHECKS = 128
0021 MAX_SUMMARY_LENGTH = 500
0022 MAX_SERIALIZED_BYTES = 64 * 1024
0023
0024 HEALTH_SCOPE_CHECKS = {
0025 "testbed": (
0026 "swf-monitor-mcp-asgi",
0027 "httpd",
0028 "github-actions",
0029 ),
0030 "epicprod": (
0031 "swf-monitor-mcp-asgi",
0032 "httpd",
0033 "github-actions",
0034 "epicprod-ops-agent",
0035 "swf-panda-bot",
0036 "campaign-assessments",
0037 "epic-devcloud-prod",
0038 "epic-devcloud-doc",
0039 ),
0040 }
0041
0042 HEALTH_REGISTRATION = {
0043 "title": "SWF assessed system health",
0044 "description": (
0045 "The bounded health assessment maintained by SWF System Status for one "
0046 "operational scope."
0047 ),
0048 "visibility": "public",
0049 "owning_subsystem": "swf-monitor System Status",
0050 "assessment_policy": ASSESSMENT_POLICY_VERSION,
0051 "max_serialized_bytes": MAX_SERIALIZED_BYTES,
0052 "quantities": {
0053 "overall_status": {
0054 "path": "overall.status",
0055 "type": "string",
0056 "required": True,
0057 "kind": "assessment",
0058 "enum": ["ok", "warning", "error", "unknown"],
0059 "description": "Deterministic aggregate status of included checks.",
0060 },
0061 "overall_reason": {
0062 "path": "overall.reason",
0063 "type": "string",
0064 "required": True,
0065 "kind": "assessment",
0066 "max_length": MAX_SUMMARY_LENGTH,
0067 "description": "Bounded deterministic explanation of aggregate status.",
0068 },
0069 "counts": {
0070 "path": "overall.counts",
0071 "type": "object",
0072 "required": True,
0073 "kind": "bounded_map",
0074 "max_items": 4,
0075 "description": "Included check counts by assessed status.",
0076 },
0077 "checks": {
0078 "path": "checks",
0079 "type": "object",
0080 "required": True,
0081 "kind": "bounded_map",
0082 "max_items": MAX_CHECKS,
0083 "description": "Bounded map of the checks determining overall health.",
0084 },
0085 },
0086 "event_sources": [
0087 {
0088 "name": "assessed-health-transitions",
0089 "resolver": EVENT_RESOLVER,
0090 "owner": "swf-monitor System Status",
0091 "event_kind": "health-transition",
0092 "event_time_field": "checked_at",
0093 "visibility": "public",
0094 }
0095 ],
0096 }
0097
0098 _STATUSES = ("ok", "warning", "error", "unknown")
0099
0100
0101 @dataclass(frozen=True)
0102 class HealthPublication:
0103 scope: str
0104 registration_update: ComponentUpdate
0105 update: ComponentUpdate
0106 projection: dict
0107 source_as_of: Optional[datetime]
0108
0109
0110 def _bounded_summary(value) -> str:
0111 summary = str(value or "No summary was provided.")
0112 return summary[:MAX_SUMMARY_LENGTH]
0113
0114
0115 def _project_check(row: Optional[SystemStatus], assessed_at: datetime) -> dict:
0116 if row is None:
0117 return {
0118 "category": "unknown",
0119 "status": "unknown",
0120 "summary": "No System Status row is available.",
0121 }
0122 status = row.status if row.status in _STATUSES else "unknown"
0123 summary = _bounded_summary(row.summary)
0124 if row.checked_at is None:
0125 status = "unknown"
0126 summary = "The System Status row has no assessment time."
0127 elif row.checked_at > assessed_at:
0128 status = "unknown"
0129 summary = "The System Status assessment time is in the future."
0130 elif assessed_at - row.checked_at > STATUS_STALE_AFTER:
0131 status = "error"
0132 summary = "The System Status assessment is older than 15 minutes."
0133 return {
0134 "category": str(row.category or "unknown")[:80],
0135 "status": status,
0136 "summary": _bounded_summary(summary),
0137 }
0138
0139
0140 def _overall(counts: dict) -> tuple[str, str]:
0141 if counts["error"]:
0142 return "error", f"{counts['error']} included check(s) are red."
0143 if counts["warning"]:
0144 suffix = (
0145 f"; {counts['unknown']} unknown"
0146 if counts["unknown"]
0147 else ""
0148 )
0149 return "warning", f"{counts['warning']} included check(s) are warning{suffix}."
0150 if counts["unknown"]:
0151 return "unknown", f"{counts['unknown']} included check(s) are unknown."
0152 return "ok", "All included checks are OK."
0153
0154
0155 def health_projection(scope: str, assessed_at: Optional[datetime] = None):
0156 """Build one scoped projection and its oldest authoritative source time."""
0157 names = HEALTH_SCOPE_CHECKS.get(scope)
0158 if names is None:
0159 raise ValueError(f"unknown Snapper health scope {scope!r}")
0160 if len(names) > MAX_CHECKS:
0161 raise ValueError(f"health scope {scope!r} exceeds {MAX_CHECKS} checks")
0162 assessed_at = assessed_at or timezone.now()
0163 rows = {
0164 row.name: row
0165 for row in SystemStatus.objects.filter(name__in=names)
0166 }
0167 checks = {
0168 name: _project_check(rows.get(name), assessed_at)
0169 for name in names
0170 }
0171 counts = {status: 0 for status in _STATUSES}
0172 for check in checks.values():
0173 counts[check["status"]] += 1
0174 overall_status, overall_reason = _overall(counts)
0175 checked_times = [
0176 row.checked_at
0177 for row in rows.values()
0178 if row.checked_at is not None and row.checked_at <= assessed_at
0179 ]
0180 source_as_of = min(checked_times) if checked_times else None
0181 projection = {
0182 "overall": {
0183 "status": overall_status,
0184 "reason": overall_reason,
0185 "counts": counts,
0186 },
0187 "checks": checks,
0188 }
0189 return projection, source_as_of, assessed_at
0190
0191
0192 @transaction.atomic
0193 def publish_health_scope(
0194 scope: str,
0195 assessed_at: Optional[datetime] = None,
0196 ) -> HealthPublication:
0197 """Idempotently register and publish one SWF health component."""
0198 projection, source_as_of, assessed_at = health_projection(scope, assessed_at)
0199 registration_update = register_component(
0200 scope=scope,
0201 name="health",
0202 publisher_identity=PUBLISHER_IDENTITY,
0203 registration=HEALTH_REGISTRATION,
0204 component_schema_version=1,
0205 )
0206 update = publish_component(
0207 scope=scope,
0208 name="health",
0209 publisher_identity=PUBLISHER_IDENTITY,
0210 data=projection,
0211 assessed_at=assessed_at,
0212 source_as_of=source_as_of,
0213 assessment_policy_version=ASSESSMENT_POLICY_VERSION,
0214 )
0215 return HealthPublication(
0216 scope=scope,
0217 registration_update=registration_update,
0218 update=update,
0219 projection=projection,
0220 source_as_of=source_as_of,
0221 )
0222
0223
0224 def publish_health_components(
0225 scopes: Optional[Iterable[str]] = None,
0226 ) -> list[HealthPublication]:
0227 """Publish the explicit health component set after a completed refresh."""
0228 selected = tuple(scopes or HEALTH_SCOPE_CHECKS)
0229 return [publish_health_scope(scope) for scope in selected]
0230
0231
0232 def compact_publication_report(publications: Iterable[HealthPublication]) -> str:
0233 """Render a bounded operator-facing publication report."""
0234 return json.dumps(
0235 [
0236 {
0237 "scope": item.scope,
0238 "revision": item.update.revision,
0239 "content_changed": item.update.content_changed,
0240 "registration_changed": (
0241 item.registration_update.registration_changed
0242 ),
0243 "overall_status": item.projection["overall"]["status"],
0244 "source_as_of": (
0245 item.source_as_of.isoformat() if item.source_as_of else None
0246 ),
0247 }
0248 for item in publications
0249 ],
0250 indent=2,
0251 sort_keys=True,
0252 )
