Coverage for integrations / social / engagement_guardrails.py: 0.0%

1""" 

2HevolveSocial - Engagement Guardrails 

3Health-first limits for games and compute. No addiction by design. 

4Humans always decide — these are suggestions, not blocks. 

5""" 

6import logging 

7from datetime import datetime, timedelta 

8from typing import Tuple, Optional 

9 

10from sqlalchemy import func 

11from sqlalchemy.orm import Session 

12 

13from .models import GameSession, GameParticipant 

14 

15logger = logging.getLogger('hevolve_social') 

16 

17# ─── Configurable Limits ─── 

18MAX_GAMES_PER_DAY = 20 # Soft cap — friendly message, not a block 

19MAX_COMPUTE_CONTINUOUS_HOURS = 8 # Auto-pause suggestion after 8 hours 

20MIN_GAME_INTERVAL_MINUTES = 2 # Prevent rapid-fire game creation 

21COOLDOWN_AFTER_LOSS_STREAK = 3 # After 3 losses, suggest different activity 

22MAX_NOTIFICATIONS_PER_SESSION = 3 # Batch achievement notifications 

23 

24 

25class EngagementGuardrails: 

26 

27 @staticmethod 

28 def check_game_limit(db: Session, user_id: str) -> Tuple[bool, Optional[str]]: 

29 """Check if user is within healthy game limits. 

30 Returns (allowed, message). Always allowed — message is a suggestion.""" 

31 today_start = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0) 

32 

33 # Count today's games 

34 games_today = db.query(func.count(GameParticipant.id)).filter( 

35 GameParticipant.user_id == user_id, 

36 GameParticipant.joined_at >= today_start, 

37 ).scalar() or 0 

38 

39 if games_today >= MAX_GAMES_PER_DAY: 

40 return True, ( 

41 f"You've played {games_today} games today — great session! " 

42 "Maybe try a thought experiment or browse the feed for a change of pace." 

43 ) 

44 

45 # Check loss streak 

46 recent_results = db.query(GameParticipant.result).filter( 

47 GameParticipant.user_id == user_id, 

48 GameParticipant.result.isnot(None), 

49 ).order_by(GameParticipant.finished_at.desc()).limit(COOLDOWN_AFTER_LOSS_STREAK).all() 

50 

51 consecutive_losses = 0 

52 for (result,) in recent_results: 

53 if result == 'loss': 

54 consecutive_losses += 1 

55 else: 

56 break 

57 

58 if consecutive_losses >= COOLDOWN_AFTER_LOSS_STREAK: 

59 return True, ( 

60 f"{consecutive_losses} tough rounds — happens to everyone! " 

61 "How about trying a collaborative puzzle? Everyone wins together." 

62 ) 

63 

64 # Check rapid-fire 

65 last_game = db.query(GameParticipant.joined_at).filter( 

66 GameParticipant.user_id == user_id, 

67 ).order_by(GameParticipant.joined_at.desc()).first() 

68 

69 if last_game and last_game[0]: 

70 minutes_since = (datetime.utcnow() - last_game[0]).total_seconds() / 60 

71 if minutes_since < MIN_GAME_INTERVAL_MINUTES: 

72 return True, "Take a breath between games — no rush." 

73 

74 return True, None 

75 

76 @staticmethod 

77 def check_compute_health(db: Session, user_id: str, 

78 continuous_hours: float = 0) -> Tuple[bool, Optional[str]]: 

79 """Check compute sharing health. Returns (healthy, suggestion).""" 

80 if continuous_hours >= MAX_COMPUTE_CONTINUOUS_HOURS: 

81 return True, ( 

82 f"Your compute has been sharing for {continuous_hours:.0f} hours. " 

83 "Everything's running smoothly. Taking a break is healthy too." 

84 ) 

85 return True, None 

86 

87 @staticmethod 

88 def should_suggest_break(db: Session, user_id: str) -> Tuple[bool, Optional[str]]: 

89 """Check if user has been active too long. Gentle suggestion only.""" 

90 two_hours_ago = datetime.utcnow() - timedelta(hours=2) 

91 

92 recent_activity = db.query(func.count(GameParticipant.id)).filter( 

93 GameParticipant.user_id == user_id, 

94 GameParticipant.joined_at >= two_hours_ago, 

95 ).scalar() or 0 

96 

97 if recent_activity >= 8: 

98 return True, ( 

99 "You've been playing for a while — great session! " 

100 "Maybe take a stretch break?" 

101 ) 

102 return False, None