1  """
  2  Complementary Altitude Engine
  3  
  4  Implements the key principle: the system should operate at the
  5  OPPOSITE altitude from the operator.
  6  
  7  When they're abstract (philosophical), ground them in concrete.
  8  When they're concrete (operational), lift them to principles.
  9  
 10  This creates productive tension that:
 11  - Prevents getting lost in abstraction
 12  - Prevents getting lost in weeds
 13  - Surfaces relevant connections across altitudes
 14  - Helps operator see full picture
 15  """
 16  
 17  from dataclasses import dataclass, field
 18  from datetime import datetime, timedelta
 19  from typing import Optional, List, Dict, Any, Tuple
 20  from enum import Enum
 21  
 22  from .detector import Altitude, AltitudeDetector, AltitudeDetection, OperatorAltitudeState
 23  from ..database import GodDatabase, Bullet
 24  
 25  
 26  class ComplementStrategy(Enum):
 27      """Strategy for complementary altitude selection."""
 28      DIRECT_OPPOSITE = "direct_opposite"    # Always opposite altitude
 29      ADJACENT_PULL = "adjacent_pull"        # One step toward opposite
 30      BRACKETING = "bracketing"              # Show both higher and lower
 31      GROUNDING = "grounding"                # Always pull toward operational
 32      LIFTING = "lifting"                    # Always pull toward philosophical
 33  
 34  
 35  @dataclass
 36  class ComplementaryRecommendation:
 37      """
 38      A recommendation for complementary content at a different altitude.
 39      """
 40      bullet_uuid: str
 41      bullet_altitude: Altitude
 42      relevance_score: float          # How relevant to current context
 43      altitude_gap: int               # Distance from operator altitude (1-3)
 44      recommendation_reason: str
 45      generated_at: datetime = field(default_factory=datetime.now)
 46  
 47  
 48  class ComplementaryAltitudeEngine:
 49      """
 50      Generates complementary altitude recommendations.
 51  
 52      When operator is at one altitude, surfaces relevant content
 53      from complementary altitudes to provide balance.
 54      """
 55  
 56      def __init__(
 57          self,
 58          god_db: GodDatabase,
 59          detector: Optional[AltitudeDetector] = None,
 60          strategy: ComplementStrategy = ComplementStrategy.DIRECT_OPPOSITE
 61      ):
 62          """
 63          Initialize the complementary altitude engine.
 64  
 65          Args:
 66              god_db: The God Database instance
 67              detector: Optional altitude detector (creates default if not provided)
 68              strategy: Strategy for selecting complementary altitude
 69          """
 70          self.god_db = god_db
 71          self.detector = detector or AltitudeDetector()
 72          self.strategy = strategy
 73  
 74      def get_complementary_altitude(
 75          self,
 76          operator_altitude: Altitude,
 77          strategy: Optional[ComplementStrategy] = None
 78      ) -> List[Altitude]:
 79          """
 80          Get complementary altitude(s) for operator's current altitude.
 81  
 82          Args:
 83              operator_altitude: Operator's current altitude
 84              strategy: Optional override strategy
 85  
 86          Returns:
 87              List of complementary altitudes to surface content from
 88          """
 89          strategy = strategy or self.strategy
 90  
 91          if strategy == ComplementStrategy.DIRECT_OPPOSITE:
 92              # Direct opposite on the altitude ladder
 93              opposites = {
 94                  Altitude.OPERATIONAL: [Altitude.PHILOSOPHICAL],
 95                  Altitude.TACTICAL: [Altitude.STRATEGIC],
 96                  Altitude.STRATEGIC: [Altitude.TACTICAL],
 97                  Altitude.PHILOSOPHICAL: [Altitude.OPERATIONAL],
 98              }
 99              return opposites[operator_altitude]
100  
101          elif strategy == ComplementStrategy.ADJACENT_PULL:
102              # One step toward opposite (gentler transition)
103              adjacent = {
104                  Altitude.OPERATIONAL: [Altitude.TACTICAL],
105                  Altitude.TACTICAL: [Altitude.OPERATIONAL, Altitude.STRATEGIC],
106                  Altitude.STRATEGIC: [Altitude.TACTICAL, Altitude.PHILOSOPHICAL],
107                  Altitude.PHILOSOPHICAL: [Altitude.STRATEGIC],
108              }
109              return adjacent[operator_altitude]
110  
111          elif strategy == ComplementStrategy.BRACKETING:
112              # Both higher and lower (full context)
113              if operator_altitude == Altitude.OPERATIONAL:
114                  return [Altitude.TACTICAL, Altitude.STRATEGIC]
115              elif operator_altitude == Altitude.PHILOSOPHICAL:
116                  return [Altitude.STRATEGIC, Altitude.TACTICAL]
117              else:
118                  # Middle altitudes get both directions
119                  lower = Altitude(operator_altitude.value - 1)
120                  higher = Altitude(operator_altitude.value + 1)
121                  return [lower, higher]
122  
123          elif strategy == ComplementStrategy.GROUNDING:
124              # Always pull toward operational
125              if operator_altitude == Altitude.OPERATIONAL:
126                  return [Altitude.OPERATIONAL]  # Already grounded
127              return [Altitude(operator_altitude.value - 1)]
128  
129          elif strategy == ComplementStrategy.LIFTING:
130              # Always pull toward philosophical
131              if operator_altitude == Altitude.PHILOSOPHICAL:
132                  return [Altitude.PHILOSOPHICAL]  # Already lifted
133              return [Altitude(operator_altitude.value + 1)]
134  
135          return [Altitude.TACTICAL]  # Default fallback
136  
137      def get_complementary_bullets(
138          self,
139          operator_altitude: Altitude,
140          blanket_id: str,
141          current_topic: Optional[str] = None,
142          limit: int = 5
143      ) -> List[ComplementaryRecommendation]:
144          """
145          Get bullets from complementary altitudes.
146  
147          Args:
148              operator_altitude: Operator's current altitude
149              blanket_id: Which Markov blanket to search
150              current_topic: Optional topic for relevance filtering
151              limit: Maximum recommendations to return
152  
153          Returns:
154              List of ComplementaryRecommendation
155          """
156          target_altitudes = self.get_complementary_altitude(operator_altitude)
157  
158          # Get bullets from blanket
159          bullets = self.god_db.get_bullets_for_blanket(blanket_id)
160  
161          recommendations: List[ComplementaryRecommendation] = []
162  
163          for bullet in bullets:
164              # Detect bullet's altitude
165              detection = self.detector.detect(
166                  bullet.content,
167                  content_id=bullet.uuid,
168                  tags=bullet.visible_tags
169              )
170  
171              # Check if bullet is at a target altitude
172              if detection.primary_altitude not in target_altitudes:
173                  continue
174  
175              # Compute altitude gap
176              gap = abs(detection.primary_altitude.value - operator_altitude.value)
177  
178              # Compute topic relevance
179              relevance = 0.5  # Default
180              if current_topic:
181                  if current_topic.lower() in bullet.content.lower():
182                      relevance = 1.0
183                  elif any(current_topic.lower() in tag.lower() for tag in bullet.visible_tags):
184                      relevance = 0.8
185  
186              # Boost by resonance score if available
187              relevance = relevance * 0.7 + bullet.resonance_score * 0.3
188  
189              # Generate reason
190              reason = self._generate_reason(
191                  operator_altitude,
192                  detection.primary_altitude,
193                  gap
194              )
195  
196              recommendations.append(ComplementaryRecommendation(
197                  bullet_uuid=bullet.uuid,
198                  bullet_altitude=detection.primary_altitude,
199                  relevance_score=relevance,
200                  altitude_gap=gap,
201                  recommendation_reason=reason
202              ))
203  
204          # Sort by relevance and return top
205          recommendations.sort(key=lambda r: r.relevance_score, reverse=True)
206          return recommendations[:limit]
207  
208      def _generate_reason(
209          self,
210          operator_altitude: Altitude,
211          bullet_altitude: Altitude,
212          gap: int
213      ) -> str:
214          """Generate human-readable reason for recommendation."""
215          if operator_altitude.value > bullet_altitude.value:
216              # Operator is abstract, grounding them
217              return f"Grounding: You're thinking at {operator_altitude.name.lower()} level. Here's a {bullet_altitude.name.lower()} perspective."
218          else:
219              # Operator is concrete, lifting them
220              return f"Lifting: You're working at {operator_altitude.name.lower()} level. Here's a {bullet_altitude.name.lower()} perspective."
221  
222      def analyze_altitude_balance(
223          self,
224          operator_state: OperatorAltitudeState,
225          threshold_minutes: int = 30
226      ) -> Dict[str, Any]:
227          """
228          Analyze if operator needs altitude balancing.
229  
230          Returns:
231              Analysis dict with recommendations
232          """
233          analysis = {
234              'operator_id': operator_state.operator_id,
235              'current_altitude': operator_state.current_altitude.name,
236              'needs_balancing': False,
237              'recommended_direction': None,
238              'reason': None,
239          }
240  
241          # Check time at current altitude
242          current_time = datetime.now()
243          if operator_state.last_transition:
244              time_at_current = current_time - operator_state.last_transition
245  
246              if time_at_current > timedelta(minutes=threshold_minutes):
247                  analysis['needs_balancing'] = True
248  
249                  if operator_state.current_altitude.value >= 3:
250                      # Too long at high altitude
251                      analysis['recommended_direction'] = 'down'
252                      analysis['reason'] = f"You've been thinking abstractly for {int(time_at_current.total_seconds() / 60)} minutes. Consider grounding in specifics."
253                  else:
254                      # Too long at low altitude
255                      analysis['recommended_direction'] = 'up'
256                      analysis['reason'] = f"You've been in the weeds for {int(time_at_current.total_seconds() / 60)} minutes. Consider stepping back to see patterns."
257  
258          # Check dominant altitude over history
259          dominant = operator_state.get_dominant_altitude()
260          if dominant != operator_state.current_altitude:
261              analysis['dominant_altitude'] = dominant.name
262  
263          return analysis
264  
265      def generate_altitude_prompt(
266          self,
267          operator_altitude: Altitude,
268          complementary_bullets: List[ComplementaryRecommendation]
269      ) -> str:
270          """
271          Generate a prompt/nudge for altitude awareness.
272  
273          Args:
274              operator_altitude: Current operator altitude
275              complementary_bullets: Recommendations from other altitudes
276  
277          Returns:
278              Human-readable prompt
279          """
280          if not complementary_bullets:
281              return ""
282  
283          # Group by altitude
284          by_altitude: Dict[Altitude, List[ComplementaryRecommendation]] = {}
285          for rec in complementary_bullets:
286              if rec.bullet_altitude not in by_altitude:
287                  by_altitude[rec.bullet_altitude] = []
288              by_altitude[rec.bullet_altitude].append(rec)
289  
290          lines = []
291  
292          if operator_altitude.value >= 3:
293              # High altitude - offer grounding
294              lines.append("While you're thinking big picture...")
295              for alt in [Altitude.OPERATIONAL, Altitude.TACTICAL]:
296                  if alt in by_altitude:
297                      count = len(by_altitude[alt])
298                      lines.append(f"  • {count} {alt.name.lower()} items might ground this")
299          else:
300              # Low altitude - offer lifting
301              lines.append("While you're in the details...")
302              for alt in [Altitude.STRATEGIC, Altitude.PHILOSOPHICAL]:
303                  if alt in by_altitude:
304                      count = len(by_altitude[alt])
305                      lines.append(f"  • {count} {alt.name.lower()} items might illuminate this")
306  
307          return "\n".join(lines)
308  
309  
310  # Quick test
311  if __name__ == "__main__":
312      import tempfile
313      import os
314  
315      # Create temp database
316      db_path = os.path.join(tempfile.gettempdir(), "test_complementary.db")
317      db = GodDatabase(db_path)
318  
319      print("=== Complementary Altitude Test ===\n")
320  
321      # Create test bullets at different altitudes
322      bullets_data = [
323          ("Truth requires both honesty and courage", ["#principle"]),
324          ("Our architecture follows domain-driven design", ["#architecture"]),
325          ("Should we use Redis or Memcached for caching?", ["#decision"]),
326          ("Implement the user authentication endpoint", ["#task"]),
327          ("The meaning of security is maintaining trust", ["#philosophy"]),
328          ("Our caching strategy needs three layers", ["#strategy"]),
329          ("Deploy to staging by end of day", ["#action"]),
330      ]
331  
332      for content, tags in bullets_data:
333          db.create_bullet(
334              content=content,
335              blanket_id="test",
336              visible_tags=tags
337          )
338  
339      # Create engine
340      engine = ComplementaryAltitudeEngine(db)
341      detector = AltitudeDetector()
342  
343      # Test with operator at different altitudes
344      print("Testing complementary recommendations:\n")
345  
346      for operator_alt in Altitude:
347          print(f"Operator at {operator_alt.name}:")
348          complementary = engine.get_complementary_altitude(operator_alt)
349          print(f"  Complementary altitudes: {[a.name for a in complementary]}")
350  
351          recs = engine.get_complementary_bullets(
352              operator_altitude=operator_alt,
353              blanket_id="test",
354              limit=3
355          )
356  
357          for rec in recs:
358              bullet = db.get_bullet(rec.bullet_uuid)
359              print(f"  • [{rec.bullet_altitude.name}] {bullet.content[:40]}...")
360              print(f"    Reason: {rec.recommendation_reason}")
361          print()
362  
363      # Test altitude balance analysis
364      print("=== Altitude Balance Analysis ===\n")
365  
366      state = OperatorAltitudeState(
367          operator_id="rick",
368          current_altitude=Altitude.PHILOSOPHICAL
369      )
370      state.last_transition = datetime.now() - timedelta(minutes=45)
371  
372      analysis = engine.analyze_altitude_balance(state)
373      print(f"Current: {analysis['current_altitude']}")
374      print(f"Needs balancing: {analysis['needs_balancing']}")
375      if analysis['reason']:
376          print(f"Reason: {analysis['reason']}")
377  
378      # Cleanup
379      os.remove(db_path)
380      print("\n=== Test Complete ===")