Coverage for integrations / social / peer_discovery.py: 59.5%

1""" 

2HevolveSocial - Decentralized Gossip Peer Discovery 

3Fully decentralized protocol for HevolveBot instances to discover each other. 

4No central registry. Peers exchange peer lists via gossip, new nodes propagate automatically. 

5""" 

6import os 

7import uuid 

8import time 

9import random 

10import logging 

11import threading 

12import requests 

13from datetime import datetime, timedelta 

14 

15from core.http_pool import pooled_get, pooled_post 

16 

17logger = logging.getLogger('hevolve_social') 

18 

19 

20def _load_or_create_node_id() -> str: 

21 """Persist node_id under platform_paths.get_data_dir() / 'node_id.json'. 

22 

23 Returns existing id on subsequent boots so the central side can 

24 dedupe joins by node_id. Falls back to a fresh in-memory uuid if 

25 the data dir is unwritable (degraded environments such as 

26 cx_Freeze read-only mode). 

27 """ 

28 import json 

29 try: 

30 from core.platform_paths import get_data_dir 

31 data_dir = get_data_dir() 

32 os.makedirs(data_dir, exist_ok=True) 

33 path = os.path.join(data_dir, 'node_id.json') 

34 if os.path.exists(path): 

35 try: 

36 with open(path, 'r', encoding='utf-8') as fh: 

37 payload = json.load(fh) 

38 nid = payload.get('node_id', '') 

39 if nid: 

40 return nid 

41 except Exception: 

42 pass # Corrupt file → regenerate 

43 nid = str(uuid.uuid4()) 

44 try: 

45 with open(path, 'w', encoding='utf-8') as fh: 

46 json.dump({'node_id': nid, 'created_at': datetime.utcnow().isoformat()}, fh) 

47 except Exception: 

48 pass # Best-effort persist; in-memory id is still valid for this boot 

49 return nid 

50 except Exception: 

51 return str(uuid.uuid4()) 

52 

53 

54# ═══════════════════════════════════════════════════════════════════════ 

55# Bandwidth Profiles - auto-selected by tier, override via env 

56# ═══════════════════════════════════════════════════════════════════════ 

57 

58BANDWIDTH_PROFILES = { 

59 'full': { 

60 'gossip_interval': 60, 

61 'health_interval': 120, 

62 'gossip_fanout': 3, 

63 'payload_mode': 'json', # Full JSON with all fields 

64 'stale_threshold': 300, 

65 'dead_threshold': 900, 

66 }, 

67 'constrained': { 

68 'gossip_interval': 300, 

69 'health_interval': 600, 

70 'gossip_fanout': 2, 

71 'payload_mode': 'json_compact', # Stripped optional fields 

72 'stale_threshold': 900, 

73 'dead_threshold': 2700, 

74 }, 

75 'minimal': { 

76 'gossip_interval': 900, 

77 'health_interval': 1800, 

78 'gossip_fanout': 1, 

79 'payload_mode': 'msgpack', # ~60% smaller than JSON 

80 'stale_threshold': 2700, 

81 'dead_threshold': 7200, 

82 }, 

83} 

84 

85# Bandwidth profile lookup — handles BOTH classification dimensions: 

86# 1. Capability tier (NodeTierLevel): embedded, observer, lite, standard, full, compute_host 

87# → describes what this node CAN do based on hardware 

88# 2. Topology mode (HEVOLVE_NODE_TIER): flat, regional, central 

89# → describes WHERE this node sits in the network hierarchy 

90# Lookup order: capability_tier first, then topology mode as fallback 

91# (see GossipProtocol.__init__: cap_tier or self.tier) 

92_TIER_BANDWIDTH_MAP = { 

93 # Capability tiers (from security/system_requirements.py NodeTierLevel) 

94 'embedded': 'minimal', 

95 'observer': 'constrained', 

96 'lite': 'constrained', 

97 'standard': 'full', 

98 'full': 'full', 

99 'compute_host': 'full', 

100 # Topology modes (fallback when capability_tier is not yet resolved) 

101 'flat': 'full', 

102 'regional': 'full', 

103 'central': 'full', 

104} 

105 

106# Compact payload: only essential fields for gossip on constrained links 

107_COMPACT_FIELDS = frozenset({ 

108 'node_id', 'url', 'public_key', 'guardrail_hash', 'code_hash', 

109 'signature', 'tier', 'capability_tier', 'timestamp', 'hart_tag', 

110}) 

111 

112 

113class GossipProtocol: 

114 """Gossip-based peer discovery for HevolveBot network.""" 

115 

116 def __init__(self): 

117 # Identity — persisted across restarts so the central side can 

118 # dedupe joins by node_id. Without persistence, every watchdog 

119 # restart of the gossip thread fabricates a fresh uuid and the 

120 # central dashboard sees infinite "new node" rows for a single 

121 # install (witnessed 2026-04-30: same install logged 

122 # node=a7ba8cc9 then node=d016058b in one server.log). 

123 self.node_id = _load_or_create_node_id() 

124 self.node_name = os.environ.get( 

125 'HEVOLVE_NODE_NAME', f'hevolve-{self.node_id[:8]}') 

126 from core.port_registry import get_port 

127 self.base_url = os.environ.get( 

128 'HEVOLVE_BASE_URL', f'http://localhost:{get_port("backend")}').rstrip('/') 

129 self.version = '1.0.0' 

130 self.started_at = datetime.utcnow() 

131 

132 # Hierarchy configuration (needed before bandwidth selection) 

133 try: 

134 from security.key_delegation import get_node_tier 

135 self.tier = get_node_tier() 

136 except ImportError: 

137 self.tier = 'flat' 

138 

139 # Bandwidth profile: auto-select from tier, allow env override 

140 self.bandwidth_profile = os.environ.get('HEVOLVE_GOSSIP_BANDWIDTH', '') 

141 if not self.bandwidth_profile: 

142 # Auto-select from capability tier (from system_requirements) 

143 cap_tier = '' 

144 try: 

145 from security.system_requirements import get_capabilities 

146 caps = get_capabilities() 

147 if caps: 

148 cap_tier = caps.tier.value 

149 except Exception: 

150 pass 

151 self.bandwidth_profile = _TIER_BANDWIDTH_MAP.get( 

152 cap_tier or self.tier, 'full') 

153 profile = BANDWIDTH_PROFILES.get(self.bandwidth_profile, BANDWIDTH_PROFILES['full']) 

154 

155 # Configuration - profile defaults, overridable by env 

156 self.gossip_interval = int(os.environ.get( 

157 'HEVOLVE_GOSSIP_INTERVAL', str(profile['gossip_interval']))) 

158 self.health_interval = int(os.environ.get( 

159 'HEVOLVE_HEALTH_INTERVAL', str(profile['health_interval']))) 

160 self.stale_threshold = int(os.environ.get( 

161 'HEVOLVE_STALE_THRESHOLD', str(profile['stale_threshold']))) 

162 self.dead_threshold = int(os.environ.get( 

163 'HEVOLVE_DEAD_THRESHOLD', str(profile['dead_threshold']))) 

164 self.gossip_fanout = int(os.environ.get( 

165 'HEVOLVE_GOSSIP_FANOUT', str(profile['gossip_fanout']))) 

166 self.payload_mode = profile['payload_mode'] 

167 

168 self.central_url = os.environ.get('HEVOLVE_CENTRAL_URL', '').rstrip('/') 

169 self.regional_url = os.environ.get('HEVOLVE_REGIONAL_URL', '').rstrip('/') 

170 

171 # Parse seed peers — env override + canonical genesis peers. 

172 # Genesis peers prevent bootstrap poisoning: even if env var is 

173 # compromised, the node always knows at least the real network. 

174 # Sourced from `core.superadmins.ALL_CENTRAL_URLS` (primary + 

175 # fallback) so gossip can find EITHER the .hevolve.ai central 

176 # OR the azurekong.hertzai.com fallback — single source of 

177 # truth, no parallel literals that could drift (Gate 4 / DRY). 

178 try: 

179 from core.superadmins import ALL_CENTRAL_URLS 

180 _GENESIS_PEERS = list(ALL_CENTRAL_URLS) 

181 except Exception: 

182 # Degraded environment fallback (cx_Freeze import chain race 

183 # at very-early boot). Mirror the canonical default. 

184 _GENESIS_PEERS = [ 

185 'https://central.hevolve.ai', 

186 'https://azurekong.hertzai.com', 

187 ] 

188 seed_str = os.environ.get('HEVOLVE_SEED_PEERS', '') 

189 env_peers = [ 

190 u.strip().rstrip('/') for u in seed_str.split(',') 

191 if u.strip() 

192 ] 

193 # Merge: env peers first (user-specified), then genesis (always present) 

194 seen = set() 

195 self.seed_peers = [] 

196 for url in env_peers + _GENESIS_PEERS: 

197 if url not in seen: 

198 seen.add(url) 

199 self.seed_peers.append(url) 

200 

201 # HART node identity (loaded on start) 

202 self._hart_tag = '' 

203 

204 # State 

205 self._running = False 

206 self._thread = None 

207 self._lock = threading.Lock() 

208 

209 # Exponential backoff for unreachable peers 

210 from core.circuit_breaker import PeerBackoff 

211 self._peer_backoff = PeerBackoff(initial=10, maximum=300) 

212 

213 logger.info( 

214 f"Gossip bandwidth: {self.bandwidth_profile} " 

215 f"(gossip={self.gossip_interval}s, health={self.health_interval}s, " 

216 f"fanout={self.gossip_fanout}, payload={self.payload_mode})" 

217 ) 

218 

219 # ─── Peer Backoff (delegates to core.circuit_breaker.PeerBackoff) ─── 

220 

221 def _is_peer_backed_off(self, peer_url: str) -> bool: 

222 return self._peer_backoff.is_backed_off(peer_url) 

223 

224 def _record_peer_failure(self, peer_url: str): 

225 self._peer_backoff.record_failure(peer_url) 

226 

227 def _record_peer_success(self, peer_url: str): 

228 self._peer_backoff.record_success(peer_url) 

229 

230 # ─── Payload Serialization ─── 

231 

232 def _gossip_self_info(self): 

233 """Return self info appropriate for current bandwidth profile.""" 

234 info = self._self_info() 

235 if self.payload_mode == 'json_compact': 

236 return {k: v for k, v in info.items() if k in _COMPACT_FIELDS} 

237 return info 

238 

239 def _gossip_peer_list(self): 

240 """Return peer list appropriate for current bandwidth profile.""" 

241 peers = self.get_peer_list() 

242 if self.payload_mode == 'json_compact': 

243 return [{k: v for k, v in p.items() if k in _COMPACT_FIELDS} 

244 for p in peers] 

245 return peers 

246 

247 @staticmethod 

248 def _serialize_payload(data) -> bytes: 

249 """Serialize payload using msgpack if available, else JSON.""" 

250 try: 

251 import msgpack 

252 return msgpack.packb(data, use_bin_type=True) 

253 except ImportError: 

254 import json 

255 return json.dumps(data).encode('utf-8') 

256 

257 @staticmethod 

258 def _deserialize_payload(raw: bytes): 

259 """Deserialize payload from msgpack or JSON.""" 

260 try: 

261 import msgpack 

262 return msgpack.unpackb(raw, raw=False) 

263 except ImportError: 

264 import json 

265 return json.loads(raw.decode('utf-8')) 

266 except Exception: 

267 import json 

268 return json.loads(raw.decode('utf-8')) 

269 

270 def start(self): 

271 """Load peers from DB, announce to seeds/known peers, start background thread.""" 

272 with self._lock: 

273 if self._running: 

274 return 

275 self._running = True 

276 

277 # Generate HART node identity if not yet established 

278 self._ensure_hart_identity() 

279 

280 # Seed peers into DB 

281 self._seed_initial_peers() 

282 

283 # Announce self to all known peers (non-blocking) 

284 threading.Thread(target=self._announce_to_all, daemon=True).start() 

285 

286 # Start gossip background loop 

287 self._thread = threading.Thread(target=self._background_loop, daemon=True) 

288 self._thread.start() 

289 logger.info(f"Gossip started: node={self.node_id[:8]}, " 

290 f"name={self.node_name}, hart_tag={self._hart_tag}, " 

291 f"seeds={len(self.seed_peers)}, " 

292 f"bandwidth={self.bandwidth_profile}") 

293 

294 # Start the central report-in loop. Every install — bundled, 

295 # Docker, ISO — reports identity to the canonical superadmin 

296 # allowlist (core.superadmins.SUPERADMIN_CENTRAL_URLS) so the 

297 # superadmin dashboard sees every node that has ever joined. 

298 # Offline-tolerant via outbox; cheap when centrals are 

299 # unreachable (PeerBackoff handles DNS + connect timeouts). 

300 try: 

301 from core.superadmin_report import start_background_loop 

302 start_background_loop(self._self_info) 

303 except Exception as e: 

304 logger.debug(f"Superadmin report-in loop not started: {e}") 

305 

306 def _ensure_hart_identity(self): 

307 """Generate HART node identity on first startup. Like getting an IP address. 

308 

309 - Central: picks a unique element → @element 

310 - Regional: picks a unique spirit → @central.spirit 

311 - Flat: no node tag needed (users get individual tags) 

312 """ 

313 try: 

314 from hart_onboarding import generate_node_identity, get_node_identity 

315 

316 # Check if already generated 

317 existing = get_node_identity() 

318 if existing and existing.get('node_tag'): 

319 self._hart_tag = existing['node_tag'] 

320 return 

321 

322 # Gather known tags from peer network for collision avoidance 

323 known_tags = set() 

324 try: 

325 peers = self._load_peers_from_db(exclude_dead=True) 

326 for p in peers: 

327 tag = p.get('hart_tag', '') 

328 if tag: 

329 known_tags.add(tag) 

330 except Exception: 

331 pass 

332 

333 # Central element comes from env or the central we connect to 

334 central_element = os.environ.get('HART_CENTRAL_ELEMENT', '') 

335 

336 identity = generate_node_identity( 

337 tier=self.tier, 

338 central_element=central_element or None, 

339 known_tags=known_tags, 

340 ) 

341 

342 self._hart_tag = identity.get('node_tag', '') 

343 except Exception as e: 

344 logger.debug(f"HART identity generation skipped: {e}") 

345 

346 def stop(self): 

347 """Stop the gossip background thread.""" 

348 with self._lock: 

349 self._running = False 

350 if self._thread: 

351 self._thread.join(timeout=10) 

352 

353 # ─── Background Loop ─── 

354 

355 def _background_loop(self): 

356 last_gossip = 0 

357 last_health = 0 

358 last_integrity = 0 

359 integrity_interval = int(os.environ.get('HEVOLVE_INTEGRITY_INTERVAL', '300')) 

360 while self._running: 

361 now = time.time() 

362 # Heartbeat to watchdog 

363 try: 

364 from security.node_watchdog import get_watchdog 

365 wd = get_watchdog() 

366 if wd: 

367 wd.heartbeat('gossip') 

368 except Exception: 

369 pass 

370 try: 

371 if now - last_gossip >= self.gossip_interval: 

372 self._gossip_round() 

373 last_gossip = now 

374 if now - last_health >= self.health_interval: 

375 self._health_check_round() 

376 last_health = now 

377 if now - last_integrity >= integrity_interval: 

378 self._integrity_round() 

379 last_integrity = now 

380 except Exception as e: 

381 logger.debug(f"Gossip loop error: {e}") 

382 time.sleep(5) 

383 

384 # ─── Gossip Round ─── 

385 

386 def _heartbeat(self): 

387 """Send heartbeat to watchdog between potentially blocking operations.""" 

388 try: 

389 from security.node_watchdog import get_watchdog 

390 wd = get_watchdog() 

391 if wd: 

392 wd.heartbeat('gossip') 

393 except Exception: 

394 pass 

395 

396 def _gossip_round(self): 

397 # Tier-aware gossip: scope targets by tier 

398 if self.tier == 'flat': 

399 peers = self._load_peers_from_db(exclude_dead=True) 

400 else: 

401 peers = self._load_peers_by_tier() 

402 

403 if not peers: 

404 # Retry seeds if we have no peers — limit to 2 seeds max 

405 # to avoid blocking for N × 5s when all seeds are unreachable 

406 for url in list(self.seed_peers)[:2]: 

407 if not self._running: 

408 return 

409 if self._is_peer_backed_off(url): 

410 continue 

411 self._announce_to_peer(url) 

412 self._heartbeat() 

413 return 

414 

415 targets = random.sample(peers, min(self.gossip_fanout, len(peers))) 

416 for peer in targets: 

417 if not self._running: 

418 return 

419 peer_url = peer['url'] 

420 if self._is_peer_backed_off(peer_url): 

421 continue 

422 try: 

423 their_peers = self._exchange_with_peer(peer_url) 

424 if their_peers: 

425 self._merge_peer_list(their_peers) 

426 # Backoff is handled inside _exchange_with_peer() 

427 except Exception as e: 

428 logger.debug(f"Gossip exchange failed with {peer_url}: {e}") 

429 self._record_peer_failure(peer_url) 

430 self._heartbeat() 

431 

432 def _health_check_round(self): 

433 self._peer_backoff.prune_expired() 

434 from .models import get_db, PeerNode 

435 db = get_db() 

436 try: 

437 peers = db.query(PeerNode).filter(PeerNode.status != 'dead').all() 

438 now = datetime.utcnow() 

439 for peer in peers: 

440 if not self._running: 

441 break 

442 if peer.node_id == self.node_id: 

443 continue 

444 reachable = self._ping_peer(peer.url) 

445 self._heartbeat() 

446 if reachable: 

447 peer.last_seen = now 

448 peer.status = 'active' 

449 else: 

450 age = (now - (peer.last_seen or peer.first_seen)).total_seconds() 

451 if age > self.dead_threshold: 

452 peer.status = 'dead' 

453 elif age > self.stale_threshold: 

454 peer.status = 'stale' 

455 db.commit() 

456 # Update contribution scores for active/stale peers 

457 try: 

458 from .hosting_reward_service import HostingRewardService 

459 for peer in peers: 

460 if peer.status in ('active', 'stale') and peer.node_id != self.node_id: 

461 HostingRewardService.compute_contribution_score(db, peer.node_id) 

462 db.commit() 

463 except Exception: 

464 pass 

465 except Exception as e: 

466 db.rollback() 

467 logger.debug(f"Health check error: {e}") 

468 finally: 

469 db.close() 

470 

471 # ─── Announce ─── 

472 

473 def _announce_to_all(self): 

474 peers = self._load_peers_from_db(exclude_dead=False) 

475 urls = set(p['url'] for p in peers) 

476 # Always include seed peers — flat-mode installs MUST be allowed 

477 # to talk to central.hevolve.ai for the universal peer-join + 

478 # central report-in spec (memory: 

479 # project_universal_peer_join_central_report.md). The previous 

480 # 2026-04 fix dropped seeds entirely in flat mode to suppress 

481 # NameResolutionError noise, which silently broke central 

482 # registration for every desktop install. PeerBackoff (line ~429) 

483 # already absorbs DNS failures with exponential backoff, so we 

484 # get the announce attempt without the log spam. 

485 urls.update(self.seed_peers) 

486 for url in urls: 

487 if not self._running: 

488 return 

489 if url != self.base_url: 

490 # Skip peers that are currently backed off (exponential 

491 # backoff from PeerBackoff). This prevents unnecessary 

492 # DNS lookups for hosts that have been unreachable. 

493 if self._is_peer_backed_off(url): 

494 continue 

495 self._announce_to_peer(url) 

496 self._heartbeat() 

497 

498 def _announce_to_peer(self, peer_url): 

499 try: 

500 resp = pooled_post( 

501 f"{peer_url}/api/social/peers/announce", 

502 json=self._self_info(), 

503 timeout=5, 

504 ) 

505 if resp.status_code == 200: 

506 self._record_peer_success(peer_url) 

507 return True 

508 self._record_peer_failure(peer_url) 

509 return False 

510 except requests.RequestException: 

511 self._record_peer_failure(peer_url) 

512 return False 

513 

514 # ─── Exchange ─── 

515 

516 def _exchange_with_peer(self, peer_url): 

517 try: 

518 payload = { 

519 'peers': self._gossip_peer_list(), 

520 'sender': self._gossip_self_info(), 

521 } 

522 

523 # Try PeerLink first (direct WebSocket, no HTTP overhead) 

524 try: 

525 peer_id = self._url_to_node_id(peer_url) 

526 if peer_id: 

527 from core.peer_link.link_manager import get_link_manager 

528 link = get_link_manager().get_link(peer_id) 

529 if link: 

530 result = link.send('gossip', payload, 

531 wait_response=True, timeout=10) 

532 if result is not None: 

533 get_link_manager().record_http_exchange(peer_id) 

534 return result.get('peers', []) 

535 except Exception: 

536 pass # Fall through to HTTP 

537 

538 # HTTP path (original) 

539 if self.payload_mode == 'msgpack': 

540 try: 

541 import msgpack 

542 resp = pooled_post( 

543 f"{peer_url}/api/social/peers/exchange", 

544 data=self._serialize_payload(payload), 

545 headers={'Content-Type': 'application/msgpack'}, 

546 timeout=10, 

547 ) 

548 except ImportError: 

549 resp = pooled_post( 

550 f"{peer_url}/api/social/peers/exchange", 

551 json=payload, timeout=10, 

552 ) 

553 else: 

554 resp = pooled_post( 

555 f"{peer_url}/api/social/peers/exchange", 

556 json=payload, timeout=10, 

557 ) 

558 if resp.status_code == 200: 

559 data = resp.json() 

560 self._record_peer_success(peer_url) 

561 # Record exchange for auto-upgrade to PeerLink 

562 try: 

563 peer_id = self._url_to_node_id(peer_url) 

564 if peer_id: 

565 from core.peer_link.link_manager import get_link_manager 

566 get_link_manager().record_http_exchange(peer_id) 

567 except Exception: 

568 pass 

569 return data.get('peers', []) 

570 self._record_peer_failure(peer_url) 

571 except requests.RequestException: 

572 self._record_peer_failure(peer_url) 

573 return None 

574 

575 def _url_to_node_id(self, peer_url: str) -> str: 

576 """Look up node_id for a peer URL from the DB.""" 

577 try: 

578 from .models import PeerNode, db_session 

579 with db_session() as db: 

580 peer = db.query(PeerNode).filter( 

581 PeerNode.url == peer_url).first() 

582 if peer: 

583 return peer.node_id 

584 except Exception: 

585 pass 

586 return '' 

587 

588 def _ping_peer(self, peer_url): 

589 if self._is_peer_backed_off(peer_url): 

590 return False 

591 try: 

592 resp = pooled_get(f"{peer_url}/api/social/peers/health", timeout=3) 

593 if resp.status_code == 200: 

594 self._record_peer_success(peer_url) 

595 return True 

596 self._record_peer_failure(peer_url) 

597 return False 

598 except requests.RequestException: 

599 self._record_peer_failure(peer_url) 

600 return False 

601 

602 # ─── Handlers (called by Flask endpoints) ─── 

603 

604 def handle_announce(self, peer_data): 

605 """Process an incoming peer announcement. Returns True if peer was new.""" 

606 from .models import get_db 

607 db = get_db() 

608 try: 

609 is_new = self._merge_peer(db, peer_data) 

610 db.commit() 

611 if is_new: 

612 logger.info(f"New peer discovered: {peer_data.get('name', '')} " 

613 f"at {peer_data.get('url', '')}") 

614 return is_new 

615 except Exception as e: 

616 db.rollback() 

617 logger.debug(f"Announce handler error: {e}") 

618 return False 

619 finally: 

620 db.close() 

621 

622 def broadcast(self, message: dict, targets: list = None) -> int: 

623 """Broadcast a message to active peers via gossip. 

624 

625 Used by RALT skill distribution and skill queries. 

626 Posts to /api/social/peers/broadcast on each target node. 

627 

628 Returns number of peers that responded with HTTP 2xx. 

629 

630 Audit fix #8 (April 2026): previously this counted every HTTP 

631 response as "sent" — even 404s when the peer didn't implement 

632 the broadcast endpoint. Callers thought they succeeded when 

633 they hadn't. Now only 2xx counts, and non-2xx is recorded as 

634 a peer failure so the back-off path kicks in. 

635 """ 

636 peers = self._load_peers_from_db(exclude_dead=True) 

637 if targets: 

638 target_set = set(targets) 

639 peers = [p for p in peers if p.get('node_id') in target_set] 

640 

641 sent = 0 

642 for peer in peers: 

643 url = peer.get('url', '') 

644 if not url or peer.get('node_id') == self.node_id: 

645 continue 

646 if self._is_peer_backed_off(url): 

647 continue 

648 try: 

649 resp = pooled_post( 

650 f"{url}/api/social/peers/broadcast", 

651 json=message, 

652 timeout=5, 

653 ) 

654 # Only 2xx counts as a successful delivery. 4xx means the 

655 # peer rejected (rate-limited, missing endpoint, bad 

656 # payload); 5xx means the peer errored. Either way the 

657 # payload did NOT land, so don't tell callers it did. 

658 if 200 <= resp.status_code < 300: 

659 self._record_peer_success(url) 

660 sent += 1 

661 else: 

662 self._record_peer_failure(url) 

663 logger.debug( 

664 f"gossip.broadcast to {url}: HTTP " 

665 f"{resp.status_code} (type={message.get('type')})") 

666 except requests.RequestException: 

667 self._record_peer_failure(url) 

668 return sent 

669 

670 def handle_exchange(self, their_peers): 

671 """Process incoming peer list, return our peer list.""" 

672 if their_peers: 

673 self._merge_peer_list(their_peers) 

674 return self.get_peer_list() 

675 

676 # ─── Peer List ─── 

677 

678 def get_peer_list(self): 

679 """Return all non-dead peers as dicts, including self.""" 

680 peers = self._load_peers_from_db(exclude_dead=True) 

681 # Include self 

682 self_info = self._self_info() 

683 if not any(p.get('node_id') == self.node_id for p in peers): 

684 peers.append(self_info) 

685 return peers 

686 

687 def get_health(self): 

688 """Return this node's health info for the /health endpoint.""" 

689 uptime = (datetime.utcnow() - self.started_at).total_seconds() 

690 peers = self._load_peers_from_db(exclude_dead=True) 

691 return { 

692 'node_id': self.node_id, 

693 'name': self.node_name, 

694 'version': self.version, 

695 'uptime_seconds': int(uptime), 

696 'peer_count': len(peers), 

697 'agent_count': self._get_count('agent'), 

698 'post_count': self._get_count('post'), 

699 'status': 'healthy', 

700 } 

701 

702 # ─── Internal Helpers ─── 

703 

704 def _self_info(self): 

705 info = { 

706 'node_id': self.node_id, 

707 'url': self.base_url, 

708 'name': self.node_name, 

709 'version': self.version, 

710 'agent_count': self._get_count('agent'), 

711 'post_count': self._get_count('post'), 

712 'timestamp': int(time.time()), 

713 'tier': self.tier, 

714 'hart_tag': self._hart_tag, 

715 } 

716 # Add cryptographic identity if available 

717 try: 

718 from security.node_integrity import get_public_key_hex, compute_code_hash, sign_json_payload 

719 info['public_key'] = get_public_key_hex() 

720 info['code_hash'] = compute_code_hash() 

721 # Include release manifest info if available 

722 try: 

723 from security.master_key import load_release_manifest 

724 manifest = load_release_manifest() 

725 if manifest: 

726 info['release_version'] = manifest.get('version', '') 

727 info['release_manifest_signature'] = manifest.get('master_signature', '') 

728 except Exception: 

729 pass 

730 # Include certificate for regional/central nodes 

731 try: 

732 from security.key_delegation import load_node_certificate 

733 cert = load_node_certificate() 

734 if cert: 

735 info['certificate'] = cert 

736 except Exception: 

737 pass 

738 info['signature'] = sign_json_payload(info) 

739 except Exception: 

740 pass 

741 # Include X25519 public key for E2E encryption 

742 try: 

743 from security.channel_encryption import get_x25519_public_hex 

744 info['x25519_public'] = get_x25519_public_hex() 

745 except Exception: 

746 pass 

747 # Include guardrail hash for peer verification 

748 try: 

749 from security.hive_guardrails import get_guardrail_hash 

750 info['guardrail_hash'] = get_guardrail_hash() 

751 except Exception: 

752 pass 

753 # Include HART OS capabilities (contribution tier + enabled features) 

754 try: 

755 from security.system_requirements import get_capabilities 

756 caps = get_capabilities() 

757 if caps: 

758 info['capability_tier'] = caps.tier.value 

759 info['enabled_features'] = caps.enabled_features 

760 info['hardware_summary'] = { 

761 'cpu_cores': caps.hardware.cpu_cores, 

762 'ram_gb': caps.hardware.ram_gb, 

763 'gpu_vram_gb': caps.hardware.gpu_vram_gb, 

764 'disk_free_gb': caps.hardware.disk_free_gb, 

765 } 

766 except Exception: 

767 pass 

768 # Advertise idle compute availability for distributed task execution 

769 try: 

770 from integrations.coding_agent.idle_detection import IdleDetectionService 

771 from integrations.social.models import get_db 

772 db = get_db() 

773 try: 

774 idle_stats = IdleDetectionService.get_idle_stats(db) 

775 info['idle_compute'] = { 

776 'available': idle_stats.get('currently_idle', 0) > 0, 

777 'idle_agents': idle_stats.get('currently_idle', 0), 

778 'opted_in': idle_stats.get('total_opted_in', 0), 

779 } 

780 finally: 

781 db.close() 

782 except Exception: 

783 pass 

784 # Advertise version info for autonomous upgrade discovery 

785 try: 

786 from integrations.agent_engine.upgrade_orchestrator import get_upgrade_orchestrator 

787 orch = get_upgrade_orchestrator() 

788 info['current_version'] = self.version 

789 status = orch.get_status() 

790 if status.get('version') and status.get('stage') == 'completed': 

791 info['available_version'] = status['version'] 

792 except Exception: 

793 pass 

794 return info 

795 

796 def _seed_initial_peers(self): 

797 """Insert seed peers into DB if not already present.""" 

798 from .models import get_db, PeerNode 

799 db = get_db() 

800 try: 

801 for url in self.seed_peers: 

802 existing = db.query(PeerNode).filter(PeerNode.url == url).first() 

803 if not existing: 

804 seed = PeerNode( 

805 node_id=f'seed_{uuid.uuid4().hex[:12]}', 

806 url=url, name='seed', version='', 

807 status='active', 

808 ) 

809 db.add(seed) 

810 # Also ensure self is in DB 

811 self_peer = db.query(PeerNode).filter( 

812 PeerNode.node_id == self.node_id).first() 

813 if not self_peer: 

814 self_peer = PeerNode( 

815 node_id=self.node_id, url=self.base_url, 

816 name=self.node_name, version=self.version, 

817 status='active', 

818 ) 

819 db.add(self_peer) 

820 db.commit() 

821 except Exception as e: 

822 db.rollback() 

823 logger.debug(f"Seed peers init error: {e}") 

824 finally: 

825 db.close() 

826 

827 def _merge_peer_list(self, peer_list): 

828 """Merge a list of peer dicts into the DB.""" 

829 from .models import get_db 

830 db = get_db() 

831 try: 

832 new_count = 0 

833 for p in peer_list: 

834 if p.get('node_id') and p.get('node_id') != self.node_id: 

835 if self._merge_peer(db, p): 

836 new_count += 1 

837 if new_count > 0: 

838 logger.info(f"Gossip: merged {new_count} new peers") 

839 db.commit() 

840 except Exception as e: 

841 db.rollback() 

842 logger.debug(f"Merge peer list error: {e}") 

843 finally: 

844 db.close() 

845 

846 def _merge_peer(self, db, peer_data): 

847 """Upsert a single peer into PeerNode table. Returns True if new. 

848 Verifies Ed25519 signature if present. Rejects banned nodes.""" 

849 from .models import PeerNode