diff --git a/platform/inneros_core_runtime/ag60_causal_diagnostics.py b/platform/inneros_core_runtime/ag60_causal_diagnostics.py new file mode 100644 index 00000000..bd9aefb5 --- /dev/null +++ b/platform/inneros_core_runtime/ag60_causal_diagnostics.py @@ -0,0 +1,119 @@ +"""AG-60 read-only, evidence-grounded incident diagnostics. + +This module never initiates network probes or device mutations. It only interprets +previously captured samples, inventory records and separately supplied topology. +""" +from __future__ import annotations + +from collections import defaultdict +from datetime import datetime, timezone +from typing import Any + + +def _as_utc(value: Any) -> datetime | None: + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None: + return None + return parsed.astimezone(timezone.utc) + except ValueError: + return None + + +def report(guardian: dict, inventory: dict, topology: dict | None = None, + previous: dict | None = None, now: datetime | None = None) -> dict: + """Build conservative, human-readable incident hypotheses from existing evidence. + + topology maps asset IDs to optional upstream, location, port, power, VLAN + and relationship_verified fields. An unverified configured relationship + is never treated as a confirmed physical dependency. + """ + topology = topology or {} + inventory = inventory or {} + assets = {a["asset_id"]: a for a in inventory.get("inventory", []) + if isinstance(a, dict) and a.get("asset_id")} + sampled = _as_utc(guardian.get("timestamp")) + now = now or datetime.now(timezone.utc) + age = (now - sampled).total_seconds() if sampled else None + fresh = age is not None and -300 <= age <= 3600 + incidents = [] + for obs in guardian.get("targets", []): + if not isinstance(obs, dict) or obs.get("online") is not False: + continue + asset_id = obs.get("asset_id") or "unidentified" + asset = assets.get(asset_id, {}) + topo = topology.get(asset_id, {}) + upstream = topo.get("upstream_asset_id") or "no identificado" + trustworthy_topology = topo.get("relationship_verified") is True + identity = asset.get("device_type") or obs.get("device_type") or "tipo no identificado" + vendor = asset.get("vendor") or "no identificado" + model = asset.get("model") or "no identificado" + observed_ports = obs.get("tcp") or {} + no_reply_ports = [str(k) for k, v in observed_ports.items() + if isinstance(v, dict) and v.get("open") is False] + hypotheses = [ + {"cause": "routing/firewall or unsuitable probe", "confidence": "low", + "evidence": "ICMP and tested TCP ports cannot establish device power status", + "read_only_check": "Compare type-specific ports and a fresh cloud/device heartbeat"}, + {"cause": "device power, PoE, or upstream link", "confidence": "low", + "evidence": "Device did not respond from the monitoring vantage point", + "read_only_check": "Check upstream switch, PoE, uplink and UPS telemetry"}, + ] + incidents.append({ + "asset_id": asset_id, + "name": asset.get("device_name") or obs.get("name") or "Equipo no identificado", + "type": identity, "device_type": identity, "vendor": vendor, + "model": model, "ip": obs.get("ip") or asset.get("ip"), + "site": guardian.get("site_id") or asset.get("site_id"), + "floor": topo.get("floor") or "ubicación no identificada", + "area": topo.get("area") or "ubicación no identificada", + "rack": topo.get("rack") or "no identificado", + "switch_port": topo.get("switch_port") or "no identificado", + "upstream": upstream, "power": topo.get("power_source") or "no identificado", + "vlan": topo.get("vlan") or "no identificada", + "relationship_source": topo.get("relationship_source") or "no documentada", + "relationship_verified": trustworthy_topology, + "sample_time": guardian.get("timestamp"), + "inventory_time": asset.get("updated_at"), + "observations": {"icmp": obs.get("icmp") or {}, + "tcp": observed_ports, "no_reply_tcp_ports": no_reply_ports}, + "hypotheses": hypotheses, + "recommended_action": "Read type-specific heartbeat and upstream telemetry before proposing a physical intervention", + }) + groups: dict[str, list] = defaultdict(list) + for incident in incidents: + if incident["upstream"] != "no identificado": + groups[incident["upstream"]].append(incident) + shared_upstreams = {} + for upstream, members in groups.items(): + if len(members) < 2: + continue + shared_upstreams[upstream] = [m["asset_id"] for m in members] + all_verified = all(m["relationship_verified"] for m in members) + for member in members: + member["hypotheses"].insert(0, { + "cause": "shared switch/uplink/power", + "confidence": "medium" if all_verified and fresh else "low", + "evidence": f"{len(members)} nonresponsive devices share " + + ("verified" if all_verified else "configured, unverified") + + f" upstream {upstream}", + "read_only_check": "Read shared switch interface errors, PoE and UPS state", + }) + old = {o.get("asset_id") for o in (previous or {}).get("targets", []) + if isinstance(o, dict) and o.get("asset_id") and o.get("online") is False} + current = {o["asset_id"] for o in incidents} + return { + "status": "CURRENT" if fresh else "STALE", + "freshness": "CURRENT" if fresh else "STALE", + "timestamp": guardian.get("timestamp"), + "sample_age_seconds": age, "source": guardian.get("source"), + "site_id": guardian.get("site_id"), + "incidents": incidents, "offline_count": len(incidents), + "shared_upstreams": shared_upstreams, + "new_offline_ids": sorted(current - old) if previous is not None else [], + "recovered_ids": sorted(old - current) if previous is not None else [], + "root_cause_verified": False, "mutations_performed": False, + "notice": "Suspected causes only. ICMP/TCP failures are not proof of power failure.", + } diff --git a/platform/inneros_core_runtime/ag60_source_reconciliation.py b/platform/inneros_core_runtime/ag60_source_reconciliation.py new file mode 100644 index 00000000..220ed97e --- /dev/null +++ b/platform/inneros_core_runtime/ag60_source_reconciliation.py @@ -0,0 +1,144 @@ +"""Read-only identity reconciliation: Bellini Guardian probes versus GWN Cloud. + +Never assumes the same IP means the same physical device. No network access, +credentials, mutable operations or repair commands in this module. +""" +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + + +def _utc_timestamp(value: Any): + try: + if isinstance(value, (int, float)): + stamp = float(value) / (1000 if value > 10**11 else 1) + return datetime.fromtimestamp(stamp, tz=timezone.utc) + if isinstance(value, str): + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + return parsed.astimezone(timezone.utc) if parsed.tzinfo else None + except (ValueError, TypeError, OverflowError, OSError): + pass + return None + + +def _mac(value: Any) -> str: + if not isinstance(value, str): + return "" + stripped = "".join(c.lower() for c in value if c.isalnum()) + return stripped if len(stripped) == 12 and all(c in "0123456789abcdef" for c in stripped) else "" + + +def reconcile(guardian: dict, inventory: dict, cloud: dict | None = None, + now: datetime | None = None, max_age_seconds: int = 1800) -> dict: + """Compare source identities and freshness; make NO physical-outage claims. + + A MAC link established by inventory is stronger than a same-IP coincidence. + Cloud status is provider availability, not proof of physical connectivity. + """ + now = now or datetime.now(timezone.utc) + cloud = cloud or inventory.get("cloud_live") or {} + devices = cloud.get("devices") or {} + all_devices = devices.get("all") or {} + rows = all_devices.get("items") + # A successful outer call does not establish that the nested GWN API + # succeeded. Never silently convert provider failure into an empty, + # apparently healthy inventory. + if (not cloud.get("ok") or not isinstance(all_devices, dict) + or all_devices.get("ok") is False + or not isinstance(rows, list) or not rows): + return {"status": "UNAVAILABLE", "reason": "gwn_cloud_snapshot_not_available", + "source": "gwn_cloud", "mutations_performed": False, + "identity_verified_by_ip_only": False} + cloud_ap_macs = { + _mac(row.get("mac")) + for row in (devices.get("access_points") or {}).get("items", []) + if isinstance(row, dict) and _mac(row.get("mac")) + } + samples = [r for r in guardian.get("targets", []) if isinstance(r, dict)] + by_asset = {s.get("asset_id"): s for s in samples if s.get("asset_id")} + by_ip = {} + for s in samples: + if s.get("ip"): + by_ip.setdefault(s["ip"], []).append(s) + assets = [a for a in inventory.get("inventory", []) if isinstance(a, dict)] + assets_by_mac = {_mac(a.get("mac")): a for a in assets if _mac(a.get("mac"))} + gtime = _utc_timestamp(guardian.get("timestamp")) + g_fresh = bool(gtime and -300 <= (now - gtime).total_seconds() <= max_age_seconds) + results = [] + for row in rows: + if not isinstance(row, dict): + continue + mac = _mac(row.get("mac")) + ip = row.get("ipv4") + asset = assets_by_mac.get(mac) if mac else None + matched = by_asset.get(asset.get("asset_id")) if asset else None + match_method = "INVENTORY_MAC" if matched else None + if not matched and ip and len(by_ip.get(ip, [])) == 1: + matched = by_ip[ip][0] + match_method = "IP_ONLY_UNVERIFIED" + seen = _utc_timestamp(row.get("lastSeen")) + cloud_fresh = bool(seen and -300 <= (now - seen).total_seconds() <= max_age_seconds) + cloud_online = row.get("status") == 1 + # Prefer the provider's explicit AP family; GWN78xx and GWN77xx + # are switches, while GWN7052F is a router, not an AP. + is_ap = (mac in cloud_ap_macs if cloud_ap_macs + else str(row.get("apType") or "").startswith("GWN76")) + if not cloud_fresh: + classification = "CLOUD_SAMPLE_STALE" + elif not matched: + classification = "NOT_IN_GUARDIAN" if is_ap else "NO_VERIFIED_GUARDIAN_IDENTITY" + elif not g_fresh: + classification = "GUARDIAN_SAMPLE_STALE" + elif match_method == "IP_ONLY_UNVERIFIED": + classification = "POTENTIAL_IDENTITY_MATCH_VERIFY_MAC" + elif ip != matched.get("ip"): + classification = "IP_DIFFERENCE_VERIFY_INTERFACES_OR_DHCP" + elif cloud_online and matched.get("online") is False: + classification = "DIFFERENT_SOURCE_CONNECTIVITY_EVIDENCE" + else: + classification = "SOURCES_CONSISTENT_WITH_LIMITATIONS" + results.append({ + "cloud_name": row.get("name") or "sin identificar", + "cloud_type": row.get("apType") or "no identificado", + "cloud_ip": ip, "cloud_mac": row.get("mac"), + "cloud_status": "ONLINE" if cloud_online else "OFFLINE" if row.get("status") == 0 else "UNKNOWN", + "cloud_seen_at": seen.isoformat() if seen else None, + "cloud_fresh": cloud_fresh, + "guardian_asset_id": matched.get("asset_id") if matched else None, + "guardian_name": matched.get("name") if matched else None, + "guardian_ip": matched.get("ip") if matched else None, + "guardian_probe_responds": matched.get("online") if matched else None, + "identity_match": match_method or "NOT_MATCHED", + "classification": classification, + "root_cause_verified": False, + }) + missing = [r for r in results if r["classification"] == "NOT_IN_GUARDIAN"] + stale_cloud_count = sum(not r["cloud_fresh"] for r in results) + snapshot_status = ( + "GUARDIAN_STALE" if not g_fresh + else "CLOUD_STALE" if stale_cloud_count == len(results) + else "PARTIAL_CLOUD_STALE" if stale_cloud_count else "CURRENT" + ) + return { + "status": snapshot_status, + "cloud_stale_devices": stale_cloud_count, + "source": "gwn_cloud_and_guardian", + "guardian_sample_at": guardian.get("timestamp"), + "cloud_devices": len(results), + "unmonitored_cloud_aps": len(missing), + "cloud_accesspoints_not_monitored": [ + {"name": r["cloud_name"], "model": r["cloud_type"], + "ip": r["cloud_ip"], "mac": r["cloud_mac"]} + for r in missing + ], + "comparisons": results, + "identity_verified_by_ip_only": False, + "root_cause_verified": False, + "mutations_performed": False, + "recommended_read_only_checks": [ + "Compare inventory MAC with controller MAC and current DHCP leases", + "Read switch FDB / LLDP / PoE status before inferring physical dependencies", + "Check the router VLAN/interface assignments when the same MAC shows multiple IPs", + ], + } diff --git a/platform/inneros_core_runtime/bellini_network_guardian.py b/platform/inneros_core_runtime/bellini_network_guardian.py index 7af56fc7..7f5fea67 100644 --- a/platform/inneros_core_runtime/bellini_network_guardian.py +++ b/platform/inneros_core_runtime/bellini_network_guardian.py @@ -475,30 +475,26 @@ def _analyze_incidents(self, sample: dict[str, Any]) -> None: ap_targets = [r for r in targets if r["device_type"] == "ap"] offline_aps = [r for r in ap_targets if not r["online"]] + # Network probes are symptoms, never calibrated physical failure proof. incident_trigger = None - root_cause = "NORMAL_OPERATION" - confidence = 1.0 + root_cause = "NO_CONFIRMED_ROOT_CAUSE" + confidence = None if not gw_online: incident_trigger = "GATEWAY_UNREACHABLE" - root_cause = "CORE_ROUTER_OR_POWER_OUTAGE" - confidence = 0.98 + root_cause = "GATEWAY_NO_RESPONSE_CAUSE_UNKNOWN" elif gw_rtt > 100.0 and ts.get("derp"): incident_trigger = "TAILSCALE_DERP_HIGH_LATENCY" - root_cause = "REMOTE_ACCESS_TAILSCALE_DERP_DEGRADATION" - confidence = 0.95 - elif not sw185.get("online") and not ap188.get("online") and gw_online: + root_cause = "REMOTE_ACCESS_RELAY_LATENCY_OBSERVED" + elif sw185.get("online") is False and ap188.get("online") is False and gw_online: incident_trigger = "ADMIN_SEGMENT_DROP" - root_cause = "ADMIN_SEGMENT_PHYSICAL_FAILURE" - confidence = 0.92 + root_cause = "ADMIN_SEGMENT_MULTIPLE_NO_REPLY_CAUSE_UNKNOWN" elif len(offline_aps) >= 2 and gw_online: incident_trigger = "MULTIPLE_AP_OUTAGE" - root_cause = "WIFI_DISTRIBUTION_SWITCH_OR_POE_DROP" - confidence = 0.88 + root_cause = "MULTIPLE_AP_NO_REPLY_CAUSE_UNKNOWN" elif len(offline_aps) == 1: incident_trigger = f"SINGLE_AP_OUTAGE_{offline_aps[0]['asset_id']}" - root_cause = "SINGLE_AP_OR_POE_DROP" - confidence = 0.90 + root_cause = "SINGLE_AP_NO_REPLY_CAUSE_UNKNOWN" if incident_trigger: if not self.active_incident: @@ -512,8 +508,11 @@ def _analyze_incidents(self, sample: dict[str, Any]) -> None: "trigger": incident_trigger, "root_cause_candidate": root_cause, "confidence": confidence, + "confidence_qualifier": "unverified_not_calibrated", + "root_cause_verified": False, + "observation_basis": "ICMP_TCP_TAILSCALE_ONLY", "affected_devices": [r["asset_id"] for r in targets if not r["online"]], - "first_failed_component": targets[0]["asset_id"] if not gw_online else (offline_aps[0]["asset_id"] if offline_aps else "unknown"), + "first_failed_component": "unknown", # No timestamp-ordered component evidence "timeline": [{"timestamp": sample["timestamp"], "event": f"Triggered: {incident_trigger}"}], "raw_snapshot": { "tailscale": ts, @@ -680,20 +679,147 @@ def get_guardian() -> BelliniNetworkGuardian: return _guardian_instance + +def _read_mongo_sample_evidence( + db: Any, current_timestamp: str | None +) -> tuple[dict[str, Any], dict[str, Any] | None]: + """Read two recent, independently stored Bellini samples, fail closed. + + The latest state in bellini_guardian_state is one upserted document and + cannot itself demonstrate two independent measurements. We therefore + require distinct _id AND timestamp from bellini_telemetry. + """ + try: + cursor = ( + db.bellini_telemetry.find( + {"site_id": SITE_ID}, + {"_id": 1, "timestamp": 1, "targets": 1}, + ) + .sort("created_at", -1) + .limit(10) + ) + now = datetime.datetime.now(datetime.timezone.utc) + distinct: list[dict[str, Any]] = [] + seen_times: set[str] = set() + seen_ids: set[str] = set() + previous: dict[str, Any] | None = None + try: + current_time = datetime.datetime.fromisoformat( + (current_timestamp or "").replace("Z", "+00:00") + ) + except ValueError: + current_time = None + for item in cursor: + ts = item.get("timestamp") + sample_id = str(item.get("_id") or "") + if not isinstance(ts, str) or not sample_id or ts in seen_times or sample_id in seen_ids: + continue + try: + parsed = datetime.datetime.fromisoformat(ts.replace("Z", "+00:00")) + age = (now - parsed).total_seconds() + except (ValueError, TypeError): + continue + if parsed.tzinfo is None: + continue + seen_times.add(ts) + seen_ids.add(sample_id) + if current_time is not None and current_time.tzinfo is not None and parsed < current_time: + if previous is None: + previous = {"timestamp": ts, "targets": item.get("targets", [])} + distinct.append({ + "id": sample_id, "timestamp": ts, + "recent": -300 <= age <= 3600, + }) + if len(distinct) >= 2: + break + two_fresh = len(distinct) >= 2 and all(x["recent"] for x in distinct[:2]) + return { + "source": "bellini_telemetry", + "status": "VERIFIED_TWO_RECENT" if two_fresh else "INSUFFICIENT", + "two_recent_distinct_samples": two_fresh, + "samples": distinct, + "read_only": True, + }, previous + except Exception as exc: + return { + "source": "bellini_telemetry", + "status": "UNAVAILABLE", + "two_recent_distinct_samples": False, + "samples": [], + "error": type(exc).__name__, + "read_only": True, + }, None + + +def _enrich_guardian_status(status: dict[str, Any], previous: dict[str, Any] | None = None) -> dict[str, Any]: + """Add read-only incident diagnostics without claiming physical root causes. + + The TARGETS topology is configured but unverified in the physical network. + """ + try: + from inneros_core_runtime.ag60_causal_diagnostics import report + from inneros_core_runtime.device_fabric import device_fabric_inventory + + inventory = device_fabric_inventory(client_id=CLIENT_ID, site_id=SITE_ID, live=False) + topology = { + target["asset_id"]: { + "upstream_asset_id": target.get("upstream_device"), + "switch_port": target.get("upstream_port"), + "relationship_source": "guardian_target_config_unverified", + "relationship_verified": False, + } + for target in TARGETS if target.get("asset_id") + } + status["diagnostics"] = report(status, inventory, topology=topology, previous=previous) + except Exception as exc: + status["diagnostics"] = {"status": "UNAVAILABLE", + "root_cause_verified": False, + "error": type(exc).__name__} + return status + + def bellini_guardian_status() -> dict[str, Any]: """Retorna el estado operativo en vivo y telemetría de Bellini Network Guardian.""" db = get_mongo_db() if db is not None: state = db.bellini_guardian_state.find_one({"site_id": SITE_ID}, {"_id": 0}) if state: - return {"ok": True, "source": "mongodb_guardian_state", **state} + evidence, previous = _read_mongo_sample_evidence(db, state.get("timestamp")) + enriched = _enrich_guardian_status( + {"ok": True, "source": "mongodb_guardian_state", **state}, + previous=previous, + ) + enriched["mongo_sample_evidence"] = evidence + return enriched g = get_guardian() - return {"ok": True, "source": "live_sweep", **g.capture_single_sweep()} + enriched = _enrich_guardian_status({"ok": True, "source": "live_sweep", **g.capture_single_sweep()}) + enriched["mongo_sample_evidence"] = { + "status": "UNAVAILABLE", "source": "bellini_telemetry", + "two_recent_distinct_samples": False, "samples": [], "read_only": True, + } + return enriched def bellini_guardian_dashboard() -> dict[str, Any]: """Retorna el panel operativo de Bellini I-II con métricas, incidentes y telemetría.""" status = bellini_guardian_status() + # Cloud is queried only on explicit dashboard access, never during + # periodic Guardian sampling or ordinary MCP status reads. + source_reconciliation: dict[str, Any] + try: + from inneros_core_runtime.device_fabric import device_fabric_inventory + from inneros_core_runtime.ag60_source_reconciliation import reconcile + inventory = device_fabric_inventory(client_id=CLIENT_ID, site_id=SITE_ID, live=True) + source_reconciliation = reconcile( + status, inventory, cloud=inventory.get("cloud_live") + ) + except Exception as exc: + source_reconciliation = { + "status": "UNAVAILABLE", + "reason": type(exc).__name__, + "root_cause_verified": False, + "mutations_performed": False, + } db = get_mongo_db() recent_incidents = [] if db is not None: @@ -718,6 +844,9 @@ def bellini_guardian_dashboard() -> dict[str, Any]: "tailscale": status.get("tailscale", {}), "summary": status.get("summary", {}), "device_matrix": status.get("targets", []), + "diagnostics": status.get("diagnostics", {}), + "source_reconciliation": source_reconciliation, + "mongo_sample_evidence": status.get("mongo_sample_evidence", {}), "recent_incidents": recent_incidents, "mutation_policy": MUTATION_POLICY, "generated_at": _now(), diff --git a/platform/tests/probe_ag60_live_mongo_readonly.py b/platform/tests/probe_ag60_live_mongo_readonly.py new file mode 100644 index 00000000..dea459cf --- /dev/null +++ b/platform/tests/probe_ag60_live_mongo_readonly.py @@ -0,0 +1,37 @@ +"""Opt-in, read-only AG-60 Mongo witness probe. + +Not named test_*, so this is NEVER automatically collected by CI or routine +unit tests. Invoke explicitly on authorized local AMD worktree: + pytest -q -s platform/tests/probe_ag60_live_mongo_readonly.py +Displays only sample IDs/timestamps/status; no credentials, device data or writes. +""" +import importlib.util +import json +from pathlib import Path + + +def test_live_mongo_two_distinct_recent_samples(): + path = Path(__file__).resolve().parents[1] / "inneros_core_runtime" / "bellini_network_guardian.py" + spec = importlib.util.spec_from_file_location("ag60_local_readonly_mongo_witness", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + db = module.get_mongo_db() + assert db is not None, "Mongo connector unavailable on test runner" + state = db.bellini_guardian_state.find_one({"site_id": module.SITE_ID}, + {"_id": 0, "timestamp": 1, "site_id": 1}) + assert isinstance(state, dict) and state.get("timestamp"), "No stored current Bellini Guardian timestamp" + proof, _ = module._read_mongo_sample_evidence(db, state["timestamp"]) + witness = { + "site_id": module.SITE_ID, + "current_state_timestamp": state["timestamp"], + "proof_status": proof.get("status"), + "two_recent_distinct_samples": proof.get("two_recent_distinct_samples"), + "samples": [ + {"id": x.get("id"), "timestamp": x.get("timestamp"), "recent": x.get("recent")} + for x in proof.get("samples", []) + ], + } + print("AG60_MONGO_READONLY_WITNESS=" + json.dumps(witness, ensure_ascii=True, default=str)) + assert proof.get("two_recent_distinct_samples") is True, ( + "No two independently stored recent Mongo samples verified; inspect printed witness" + ) diff --git a/platform/tests/probe_ag60_shadow_readonly.py b/platform/tests/probe_ag60_shadow_readonly.py new file mode 100644 index 00000000..9c77ae0e --- /dev/null +++ b/platform/tests/probe_ag60_shadow_readonly.py @@ -0,0 +1,99 @@ +"""Opt-in AG-60 shadow canary: exercise BRANCH dashboard with live READS. + +Explicit-only; not selected by routine pytest or GitHub Actions. +No active services are restarted or mutated. Any probe/sweep attempt fails. +Mongo is wrapped to expose only allowed find / find_one calls; the GWN +inventory path is the existing live read-only provider. +""" +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + + +def _branch_module(fullname, filename): + path = Path(__file__).resolve().parents[1] / "inneros_core_runtime" / filename + spec = importlib.util.spec_from_file_location(fullname, path) + module = importlib.util.module_from_spec(spec) + sys.modules[fullname] = module + spec.loader.exec_module(module) + return module + + +class _ReadOnlyCollection: + def __init__(self, original): + self.original = original + + def find(self, *args, **kwargs): + return self.original.find(*args, **kwargs) + + def find_one(self, *args, **kwargs): + return self.original.find_one(*args, **kwargs) + + +class _ReadOnlyDb: + def __init__(self, original): + self.bellini_guardian_state = _ReadOnlyCollection(original.bellini_guardian_state) + self.bellini_telemetry = _ReadOnlyCollection(original.bellini_telemetry) + self.bellini_incidents = _ReadOnlyCollection(original.bellini_incidents) + + +def _forbidden(*args, **kwargs): + raise AssertionError("Shadow canary forbids active network probes or mutations") + + +def test_shadow_dashboard_uses_real_mongo_and_cloud_read_only(monkeypatch): + _branch_module("inneros_core_runtime.ag60_causal_diagnostics", + "ag60_causal_diagnostics.py") + _branch_module("inneros_core_runtime.ag60_source_reconciliation", + "ag60_source_reconciliation.py") + guardian = _branch_module("ag60_branch_guardian_shadow", "bellini_network_guardian.py") + + real_db = guardian.get_mongo_db() + assert real_db is not None, "MongoDB not available on local runner" + existing = real_db.bellini_guardian_state.find_one( + {"site_id": guardian.SITE_ID}, {"_id": 0, "timestamp": 1} + ) + assert existing and existing.get("timestamp"), "No current existing sample: refusing fallback" + monkeypatch.setattr(guardian, "get_mongo_db", lambda: _ReadOnlyDb(real_db)) + monkeypatch.setattr(guardian, "get_guardian", _forbidden) + monkeypatch.setattr(guardian, "probe_icmp", _forbidden) + monkeypatch.setattr(guardian, "probe_tcp", _forbidden) + monkeypatch.setattr(guardian, "probe_tailscale", _forbidden) + + dashboard = guardian.bellini_guardian_dashboard() + diagnosis = dashboard.get("diagnostics") or {} + proof = dashboard.get("mongo_sample_evidence") or {} + cloud = dashboard.get("source_reconciliation") or {} + assert dashboard.get("ok") is True + assert diagnosis.get("status") == "CURRENT", diagnosis.get("status") + assert diagnosis.get("root_cause_verified") is False + assert proof.get("two_recent_distinct_samples") is True, proof.get("status") + assert len(proof.get("samples") or []) >= 2 + assert cloud.get("status") == "CURRENT", cloud.get("status") + assert cloud.get("root_cause_verified") is False + assert cloud.get("mutations_performed") is False + assert isinstance(cloud.get("comparisons"), list) + out = { + "guardian_sample": diagnosis.get("timestamp"), + "diagnosis_status": diagnosis.get("status"), + "offline_count": diagnosis.get("offline_count"), + "mongo_proof_status": proof.get("status"), + "mongo_sample_ids_and_times": [ + {"id": x.get("id"), "timestamp": x.get("timestamp")} + for x in proof.get("samples", [])[:2] + ], + "cloud_snapshot_status": cloud.get("status"), + "cloud_devices_compared": cloud.get("cloud_devices"), + "cloud_accesspoints_not_monitored": cloud.get("unmonitored_cloud_aps"), + "ip_only_matches": sum(1 for x in cloud["comparisons"] + if x.get("identity_match") == "IP_ONLY_UNVERIFIED"), + "physical_root_cause_verified": False, + "network_mutations": False, + "production_deployed": False, + } + print("AG60_BRANCH_SHADOW_CANARY=" + json.dumps(out, sort_keys=True)) diff --git a/platform/tests/test_ag60_causal_diagnostics.py b/platform/tests/test_ag60_causal_diagnostics.py new file mode 100644 index 00000000..ee3b9151 --- /dev/null +++ b/platform/tests/test_ag60_causal_diagnostics.py @@ -0,0 +1,38 @@ +import sys +from pathlib import Path +from datetime import datetime, timezone + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "inneros_core_runtime")) +from ag60_causal_diagnostics import report + + +def sample(ids): + return {"timestamp": datetime.now(timezone.utc).isoformat(), "site_id": "bellini-i-ii", + "targets": [{"asset_id": x, "online": False, "ip": "192.168.3.185"} for x in ids]} + + +def test_unknown_stays_unknown(): + r = report(sample(["a"]), {"inventory": []}) + assert r["incidents"][0]["type"] == "tipo no identificado" + assert r["incidents"][0]["upstream"] == "no identificado" + assert r["root_cause_verified"] is False + + +def test_known_switch_and_shared_upstream(): + inv = {"inventory": [{"asset_id": "a", "device_type": "poe_switch", "vendor": "Hikvision"}]} + topo = {"a": {"upstream_asset_id": "s"}, "b": {"upstream_asset_id": "s"}} + r = report(sample(["a", "b"]), inv, topo) + assert r["incidents"][0]["type"] == "poe_switch" + assert r["shared_upstreams"] == {"s": ["a", "b"]} + assert r["incidents"][0]["hypotheses"][0]["confidence"] == "low" + assert "unverified" in r["incidents"][0]["hypotheses"][0]["evidence"] + verified = {"a": {"upstream_asset_id": "s", "relationship_verified": True}, + "b": {"upstream_asset_id": "s", "relationship_verified": True}} + assert report(sample(["a", "b"]), inv, verified)["incidents"][0]["hypotheses"][0]["confidence"] == "medium" + + +def test_delta_and_stale(): + r = report(sample(["b"]), {"inventory": []}, previous=sample(["a"])) + assert r["new_offline_ids"] == ["b"] + assert r["recovered_ids"] == ["a"] + assert report({"targets": []}, {"inventory": []})["freshness"] == "STALE" diff --git a/platform/tests/test_ag60_guardian_integration.py b/platform/tests/test_ag60_guardian_integration.py new file mode 100644 index 00000000..54dc4b55 --- /dev/null +++ b/platform/tests/test_ag60_guardian_integration.py @@ -0,0 +1,237 @@ +"""Offline contract tests loading the branch files, not an installed InnerOS copy.""" +from datetime import datetime, timezone +import importlib.util +from pathlib import Path +import sys + +import pytest +from inneros_core_runtime import device_fabric + + +def _local_module(name, filename): + path = Path(__file__).resolve().parents[1] / "inneros_core_runtime" / filename + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture +def bng(monkeypatch): + diagnostics = _local_module("inneros_core_runtime.ag60_causal_diagnostics", + "ag60_causal_diagnostics.py") + monkeypatch.setitem(sys.modules, "inneros_core_runtime.ag60_causal_diagnostics", diagnostics) + # CI intentionally installs only minimal offline test requirements. + # Stub the unused Mongo import, never connect to a database. + if "pymongo" not in sys.modules: + from types import ModuleType + stub = ModuleType("pymongo") + stub.database = ModuleType("pymongo.database") + monkeypatch.setitem(sys.modules, "pymongo", stub) + return _local_module("bellini_network_guardian_under_test", "bellini_network_guardian.py") + + +class _State: + def find_one(self, *_args, **_kwargs): + return { + "timestamp": datetime.now(timezone.utc).isoformat(), + "site_id": "bellini-i-ii", + "targets": [ + {"asset_id": "bellini_switch_185", "ip": "192.168.3.185", + "online": False, "icmp": {"loss_pct": 100}}, + {"asset_id": "bellini_ap_213", "ip": "192.168.3.213", + "online": False, "icmp": {"loss_pct": 100}}, + ], + "summary": {"offline_count": 2}, + } + + +class _DB: + bellini_guardian_state = _State() + + +def test_existing_mcp_status_is_enriched_without_extra_probes(monkeypatch, bng): + monkeypatch.setattr(bng, "get_mongo_db", lambda: _DB()) + monkeypatch.setattr(device_fabric, "device_fabric_inventory", lambda **kwargs: { + "inventory": [{"asset_id": "bellini_switch_185", "device_name": "Switch Administración", + "device_type": "poe_switch", "vendor": "Hikvision", + "model": "DS-3E1510P-EI/M"}]}) + status = bng.bellini_guardian_status() + assert status["source"] == "mongodb_guardian_state" + diag = status["diagnostics"] + assert diag["status"] == "CURRENT" + assert diag["root_cause_verified"] is False + assert diag["offline_count"] == 2 + switch, ap = diag["incidents"] + assert switch["vendor"] == "Hikvision" + assert switch["model"] == "DS-3E1510P-EI/M" + assert ap["upstream"] == "SW_T1_P0" + assert not ap["relationship_verified"] + assert ap["floor"] == "ubicación no identificada" + + +def test_dashboard_exposes_diagnostics_without_fake_root_cause(monkeypatch, bng): + monkeypatch.setattr(bng, "get_mongo_db", lambda: None) + monkeypatch.setattr(bng, "bellini_guardian_status", lambda: { + "summary": {"offline_count": 1}, "targets": [], + "diagnostics": {"status": "STALE", "root_cause_verified": False}}) + dash = bng.bellini_guardian_dashboard() + assert dash["diagnostics"]["root_cause_verified"] is False + assert dash["diagnostics"]["status"] == "STALE" + + +def test_inventory_failure_fails_closed(monkeypatch, bng): + def broken(**kwargs): + raise RuntimeError("down") + monkeypatch.setattr(device_fabric, "device_fabric_inventory", broken) + status = bng._enrich_guardian_status({"timestamp": "2026-10-19T00:00:00Z", "targets": []}) + assert status["diagnostics"]["status"] == "UNAVAILABLE" + assert status["diagnostics"]["root_cause_verified"] is False + + +def test_root_cause_is_never_declared_by_icmp(monkeypatch, bng): + monkeypatch.setattr(bng, "get_mongo_db", lambda: None) + sample = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "gateway_online": False, "gateway_rtt_ms": 0.0, "tailscale": {}, + "summary": {"offline_count": 1}, + "targets": [{"ip": "192.168.3.1", "asset_id": "gw", + "device_type": "gateway", "online": False}], + } + guardian = bng.BelliniNetworkGuardian() + guardian._analyze_incidents(sample) + incident = guardian.active_incident + assert incident["trigger"] == "GATEWAY_UNREACHABLE" + assert incident["root_cause_candidate"] == "GATEWAY_NO_RESPONSE_CAUSE_UNKNOWN" + assert incident["confidence"] is None + assert incident["root_cause_verified"] is False + assert incident["first_failed_component"] == "unknown" + + +def test_multiple_aps_do_not_imply_poe_outage(monkeypatch, bng): + monkeypatch.setattr(bng, "get_mongo_db", lambda: None) + sample = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "gateway_online": True, "gateway_rtt_ms": 10.0, + "tailscale": {"derp": False}, "summary": {"offline_count": 2}, + "targets": [ + {"ip": "192.168.3.1", "asset_id": "gw", "device_type": "gateway", "online": True}, + {"ip": "192.168.3.188", "asset_id": "ap188", "device_type": "ap", "online": False}, + {"ip": "192.168.3.213", "asset_id": "ap213", "device_type": "ap", "online": False}, + ], + } + guardian = bng.BelliniNetworkGuardian() + guardian._analyze_incidents(sample) + incident = guardian.active_incident + assert incident["trigger"] == "MULTIPLE_AP_OUTAGE" + assert incident["root_cause_candidate"] == "MULTIPLE_AP_NO_REPLY_CAUSE_UNKNOWN" + assert incident["confidence_qualifier"] == "unverified_not_calibrated" + assert incident["root_cause_verified"] is False + + +def test_dashboard_reconciles_cloud_without_touching_devices(monkeypatch, bng): + from inneros_core_runtime import device_fabric + cloud_module = _local_module("inneros_core_runtime.ag60_source_reconciliation", + "ag60_source_reconciliation.py") + monkeypatch.setitem(sys.modules, "inneros_core_runtime.ag60_source_reconciliation", + cloud_module) + monkeypatch.setattr(bng, "get_mongo_db", lambda: None) + monkeypatch.setattr(bng, "bellini_guardian_status", lambda: { + "timestamp": datetime.now(timezone.utc).isoformat(), + "site_id": "bellini-i-ii", + "targets": [{"asset_id": "bellini_ap_212", "ip": "192.168.3.212", + "name": "GWN Wi-Fi AP 212", "online": False}], + "summary": {"offline_count": 1}, + "diagnostics": {"root_cause_verified": False}, + }) + called = [] + + def fake_inventory(**kwargs): + called.append(kwargs) + now_ms = int(datetime.now(timezone.utc).timestamp() * 1000) + return {"inventory": [], "cloud_live": {"ok": True, "devices": {"all": {"items": [ + {"apType": "GWN7660", "name": "Salón Inglés", + "mac": "EC:74:D7:3D:27:2C", "ipv4": "192.168.3.212", + "status": 1, "lastSeen": now_ms}, + {"apType": "GWN7605", "name": "Lobby Principal", + "mac": "EC:74:D7:1A:01:D0", "ipv4": "192.168.3.197", + "status": 1, "lastSeen": now_ms}, + ]}}}} + + monkeypatch.setattr(device_fabric, "device_fabric_inventory", fake_inventory) + dashboard = bng.bellini_guardian_dashboard() + assert called == [{"client_id": "bellini", "site_id": "bellini-i-ii", "live": True}] + source = dashboard["source_reconciliation"] + assert source["cloud_devices"] == 2 + assert source["unmonitored_cloud_aps"] == 1 + assert source["root_cause_verified"] is False + assert source["comparisons"][0]["identity_match"] == "IP_ONLY_UNVERIFIED" + assert "credentials" not in source + + +class _SampleCursor(list): + def sort(self, *args, **kwargs): + return self + + def limit(self, n): + return _SampleCursor(self[:n]) + + +class _Telemetry: + def __init__(self, records): + self.records = records + + def find(self, *args, **kwargs): + return _SampleCursor(self.records) + + +def test_two_recent_mongo_samples_require_distinct_ids_and_timestamps(bng): + from datetime import timedelta + now = datetime.now(timezone.utc) + recent1 = (now - timedelta(seconds=10)).isoformat() + recent2 = (now - timedelta(seconds=30)).isoformat() + class DB: + bellini_telemetry = _Telemetry([ + {"_id": "a", "timestamp": recent1, "targets": [{"asset_id": "x", "online": False}]}, + {"_id": "b", "timestamp": recent2, "targets": [{"asset_id": "x", "online": True}]}, + ]) + proof, previous = bng._read_mongo_sample_evidence(DB(), recent1) + assert proof["status"] == "VERIFIED_TWO_RECENT" + assert proof["two_recent_distinct_samples"] is True + assert len(proof["samples"]) == 2 + assert proof["samples"][0]["id"] != proof["samples"][1]["id"] + assert previous["timestamp"] == recent2 + + +def test_duplicate_timestamp_is_not_two_samples(bng): + ts = datetime.now(timezone.utc).isoformat() + class DB: + bellini_telemetry = _Telemetry([ + {"_id": "a", "timestamp": ts}, + {"_id": "b", "timestamp": ts}, + ]) + proof, previous = bng._read_mongo_sample_evidence(DB(), ts) + assert proof["status"] == "INSUFFICIENT" + assert proof["two_recent_distinct_samples"] is False + assert len(proof["samples"]) == 1 + assert previous is None + + +def test_old_mongo_samples_are_not_certified_as_recent(bng): + class DB: + bellini_telemetry = _Telemetry([ + {"_id": "a", "timestamp": "2026-10-01T12:00:00+00:00"}, + {"_id": "b", "timestamp": "2026-10-01T11:00:00+00:00"}, + ]) + proof, previous = bng._read_mongo_sample_evidence(DB(), "2026-10-01T12:00:00+00:00") + assert proof["status"] == "INSUFFICIENT" + assert not proof["two_recent_distinct_samples"] + + +def test_mongo_evidence_query_errors_fail_closed(bng): + class DB: + pass + proof, previous = bng._read_mongo_sample_evidence(DB(), "2026-10-19T10:00:00Z") + assert proof["status"] == "UNAVAILABLE" + assert proof["two_recent_distinct_samples"] is False + assert previous is None diff --git a/platform/tests/test_ag60_source_reconciliation.py b/platform/tests/test_ag60_source_reconciliation.py new file mode 100644 index 00000000..ad5ef09b --- /dev/null +++ b/platform/tests/test_ag60_source_reconciliation.py @@ -0,0 +1,147 @@ +"""AG-60 cloud-versus-Guardian reconciliation with offline vendor-like fixtures.""" +from datetime import datetime, timezone +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "inneros_core_runtime")) +from ag60_source_reconciliation import reconcile + + +NOW = datetime(2026, 10, 19, 22, 30, tzinfo=timezone.utc) + + +def guardian(*targets): + return {"timestamp": NOW.isoformat(), "targets": list(targets)} + + +def cloud(*rows): + return {"ok": True, "devices": {"all": {"items": [ + {"status": 1, "lastSeen": int(NOW.timestamp() * 1000), **row} for row in rows]}}} + + +def test_ip_only_cannot_establish_same_device(): + g = guardian({"asset_id": "old", "name": "AP 212", "ip": "192.168.3.212", "online": False}) + snap = cloud({"apType": "GWN7660", "mac": "EC:74:D7:3D:27:2C", + "ipv4": "192.168.3.212", "name": "Salón Inglés"}) + r = reconcile(g, {"inventory": []}, snap, now=NOW) + assert r["comparisons"][0]["identity_match"] == "IP_ONLY_UNVERIFIED" + assert r["comparisons"][0]["classification"] == "POTENTIAL_IDENTITY_MATCH_VERIFY_MAC" + assert not r["root_cause_verified"] + + +def test_mac_match_with_different_ip_is_not_a_confirmed_outage(): + g = guardian({"asset_id": "gateway", "name": "Gateway", "ip": "192.168.3.1", "online": True}) + inv = {"inventory": [{"asset_id": "gateway", "mac": "ec:74:d7:23:bc:91"}]} + snap = cloud({"apType": "GCC6010", "mac": "EC:74:D7:23:BC:91", + "ipv4": "192.168.1.1", "name": "Router Principal"}) + result = reconcile(g, inv, snap, now=NOW)["comparisons"][0] + assert result["identity_match"] == "INVENTORY_MAC" + assert result["classification"] == "IP_DIFFERENCE_VERIFY_INTERFACES_OR_DHCP" + assert result["root_cause_verified"] is False + + +def test_cloud_aps_missing_from_guardian_are_counted(): + g = guardian({"asset_id": "only", "ip": "192.168.3.212", "online": True}) + snap = cloud({"apType": "GWN7605", "mac": "EC:74:D7:1A:01:D0", + "ipv4": "192.168.3.197", "name": "Lobby Principal"}, + {"apType": "GWN7801P", "mac": "EC:74:D7:13:EE:B8", + "ipv4": "192.168.3.226", "name": "Switch Consola"}) + r = reconcile(g, {"inventory": []}, snap, now=NOW) + assert r["cloud_devices"] == 2 + assert r["unmonitored_cloud_aps"] == 1 + assert r["cloud_accesspoints_not_monitored"][0]["name"] == "Lobby Principal" + + +def test_stale_sources_dont_certify_current_connectivity(): + g = {"timestamp": "2026-10-18T20:00:00+00:00", + "targets": [{"asset_id": "ap", "ip": "192.168.3.212", "online": False}]} + snap = cloud({"apType": "GWN7660", "mac": "EC:74:D7:3D:27:2C", + "ipv4": "192.168.3.212", "name": "Salón Inglés"}) + r = reconcile(g, {"inventory": []}, snap, now=NOW) + assert r["status"] == "GUARDIAN_STALE" + assert r["comparisons"][0]["classification"] == "GUARDIAN_SAMPLE_STALE" + + +def test_no_cloud_read_is_unavailable_not_zero_devices(): + r = reconcile(guardian(), {"inventory": []}, {"ok": False}, now=NOW) + assert r["status"] == "UNAVAILABLE" + assert not r["identity_verified_by_ip_only"] + + +def test_fresh_mac_match_conflicting_online_is_not_physical_cause(): + snap = cloud({"apType": "GWN7660", "mac": "EC:74:D7:3D:27:2C", + "ipv4": "192.168.3.212", "name": "Salón Inglés"}) + g = guardian({"asset_id": "ap212", "ip": "192.168.3.212", "online": False}) + inv = {"inventory": [{"asset_id": "ap212", "mac": "ec74d73d272c"}]} + r = reconcile(g, inv, snap, now=NOW) + assert r["comparisons"][0]["classification"] == "DIFFERENT_SOURCE_CONNECTIVITY_EVIDENCE" + assert r["comparisons"][0]["root_cause_verified"] is False + + +def test_incomplete_cloud_snapshot_is_unavailable(): + current = guardian() + for response in [ + {"ok": True, "devices": {"all": {"ok": False, "items": []}}}, + {"ok": True, "devices": {"all": {"ok": False, "items": [ + {"name": "stale", "apType": "GWN7660", "ipv4": "192.168.3.212"}]}}}, + {"ok": True, "devices": {"all": {}}}, + {"ok": True, "devices": {"all": {"items": []}}}, + ]: + result = reconcile(current, {"inventory": []}, response, now=NOW) + assert result["status"] == "UNAVAILABLE" + assert result["identity_verified_by_ip_only"] is False + + +def test_stale_cloud_unmatched_device_not_reported_as_new_unmonitored_ap(): + now_ms = int((NOW.timestamp() - 7200) * 1000) + snap = {"ok": True, "devices": {"all": {"ok": True, "items": [ + {"apType": "GWN7660", "mac": "EC:74:D7:3D:27:2C", + "ipv4": "192.168.3.212", "name": "Salón Inglés", + "status": 1, "lastSeen": now_ms}]}}} + result = reconcile(guardian(), {"inventory": []}, snap, now=NOW) + assert result["status"] == "CLOUD_STALE" + assert result["cloud_stale_devices"] == 1 + assert result["comparisons"][0]["classification"] == "CLOUD_SAMPLE_STALE" + assert result["unmonitored_cloud_aps"] == 0 + + +def test_switch_and_router_models_never_count_as_unmonitored_aps(): + snap = cloud( + {"apType": "GWN7802P", "mac": "EC:74:D7:13:EE:B9", + "ipv4": "192.168.3.228", "name": "Switch Torre"}, + {"apType": "GWN7052F", "mac": "EC:74:D7:13:EE:C0", + "ipv4": "192.168.3.207", "name": "Router WiFi"}, + {"apType": "GWN7660", "mac": "EC:74:D7:3D:27:2C", + "ipv4": "192.168.3.212", "name": "Salón Inglés"}, + ) + result = reconcile(guardian(), {"inventory": []}, snap, now=NOW) + assert result["unmonitored_cloud_aps"] == 1 + assert result["cloud_accesspoints_not_monitored"][0]["name"] == "Salón Inglés" + + +def test_provider_ap_family_overrides_untrusted_model_guess(): + snap = cloud( + {"apType": "OTHER", "mac": "EC:74:D7:3D:27:2C", + "ipv4": "192.168.3.212", "name": "AP con modelo pendiente"}, + {"apType": "GWN7802P", "mac": "EC:74:D7:13:EE:B9", + "ipv4": "192.168.3.228", "name": "Switch"}, + ) + snap["devices"]["access_points"] = {"items": [ + {"mac": "EC:74:D7:3D:27:2C", "apType": "OTHER"}]} + result = reconcile(guardian(), {"inventory": []}, snap, now=NOW) + assert result["unmonitored_cloud_aps"] == 1 + assert result["cloud_accesspoints_not_monitored"][0]["name"] == "AP con modelo pendiente" + + +def test_partial_stale_cloud_snapshot_not_certified_current(): + snap = cloud( + {"apType": "GWN7660", "mac": "EC:74:D7:3D:27:2C", + "ipv4": "192.168.3.212", "name": "Salón Inglés"}, + {"apType": "GWN7605", "mac": "EC:74:D7:1A:01:D0", + "ipv4": "192.168.3.197", "name": "Lobby Principal"}, + ) + snap["devices"]["all"]["items"][1]["lastSeen"] = int((NOW.timestamp() - 7200) * 1000) + result = reconcile(guardian(), {"inventory": []}, snap, now=NOW) + assert result["status"] == "PARTIAL_CLOUD_STALE" + assert result["cloud_stale_devices"] == 1 + assert result["unmonitored_cloud_aps"] == 1