""" ObservationEngine — watches the swarm motion environment for skill candidates ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Listens to motion snapshots from the swarm motion hive and the VM control dispatcher's action history, detects recurring patterns, and surfaces them as skill candidates for the writer. A "pattern" is a recurring sequence of VM actions observed within a temporal window. When the same sequence appears >= min_occurrences times, it becomes an ObservedPattern eligible for code generation. The engine can also be fed observations directly (for tests and for end-user skill teaching where we record the user's own actions). """ from __future__ import annotations import hashlib import logging import threading import time import uuid from collections import Counter, deque from dataclasses import dataclass, field from typing import Any, Deque, Dict, List, Optional, Tuple logger = logging.getLogger("aureon.code_architect.observer") # ───────────────────────────────────────────────────────────────────────────── # ObservedPattern # ───────────────────────────────────────────────────────────────────────────── @dataclass class ObservedPattern: """A pattern detected in the action stream — a skill candidate.""" pattern_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8]) signature: str = "" # hash of the action sequence action_sequence: List[Dict[str, Any]] = field(default_factory=list) occurrence_count: int = 0 first_seen: float = field(default_factory=time.time) last_seen: float = field(default_factory=time.time) coherence: float = 0.0 # avg snapshot coherence suggested_name: str = "" suggested_description: str = "" contributing_sessions: List[str] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: return { "pattern_id": self.pattern_id, "signature": self.signature, "action_sequence": self.action_sequence, "action_count": len(self.action_sequence), "occurrence_count": self.occurrence_count, "first_seen": self.first_seen, "last_seen": self.last_seen, "coherence": round(self.coherence, 4), "suggested_name": self.suggested_name, "suggested_description": self.suggested_description, "contributing_sessions": self.contributing_sessions, } # ───────────────────────────────────────────────────────────────────────────── # ObservationEngine # ───────────────────────────────────────────────────────────────────────────── class ObservationEngine: """ Tracks observed action sequences and surfaces recurring patterns. Usage: obs = ObservationEngine(window_size=5, min_occurrences=3) obs.record_action("vm_mouse_move", {"x": 500, "y": 300}) obs.record_action("vm_left_click", {"x": 500, "y": 300}) obs.record_action("vm_type_text", {"text": "hello"}) ... (repeat) patterns = obs.get_pending_patterns() """ def __init__( self, window_size: int = 3, # length of action sequences we track min_occurrences: int = 2, # threshold before surfacing as pattern max_history: int = 2000, snap_coherence_default: float = 0.5, ): self.window_size = window_size self.min_occurrences = min_occurrences self.max_history = max_history self.snap_coherence_default = snap_coherence_default self._actions: Deque[Dict[str, Any]] = deque(maxlen=max_history) self._pattern_counts: Counter = Counter() self._pattern_sequences: Dict[str, List[Dict[str, Any]]] = {} self._pattern_metadata: Dict[str, Dict[str, Any]] = {} self._surfaced: set = set() self._pending_patterns: List[ObservedPattern] = [] self._lock = threading.RLock() # Metrics self._total_actions_recorded = 0 self._total_patterns_detected = 0 self._recent_coherence: Deque[float] = deque(maxlen=50) # ───────────────────────────────────────────────────────────────────── # Recording # ───────────────────────────────────────────────────────────────────── def record_action( self, action: str, params: Optional[Dict[str, Any]] = None, session_id: str = "", coherence: Optional[float] = None, ) -> None: """Record a single VM action into the observation stream.""" entry = { "action": str(action), "params": dict(params or {}), "session_id": session_id, "timestamp": time.time(), "coherence": float(coherence) if coherence is not None else self.snap_coherence_default, } with self._lock: self._actions.append(entry) self._total_actions_recorded += 1 self._recent_coherence.append(entry["coherence"]) self._update_patterns() def record_sequence( self, actions: List[Tuple[str, Dict[str, Any]]], session_id: str = "", coherence: float = 0.8, ) -> None: """Record a pre-formed sequence of actions atomically.""" for action, params in actions: self.record_action(action, params, session_id=session_id, coherence=coherence) def record_from_dispatcher_history(self, dispatcher: Any, session_id: Optional[str] = None) -> int: """ Pull recent action history from a VMControlDispatcher's sessions. Returns the number of new actions recorded. """ count = 0 try: sessions = dispatcher.list_sessions() except Exception: return 0 for s in sessions: sid = s.get("session_id") if session_id and sid != session_id: continue controller = dispatcher.get_session(sid) if not controller: continue history = getattr(controller.session, "action_history", []) for result in history: try: self.record_action( action=result.action, params=result.data, session_id=sid, coherence=0.85 if result.ok else 0.2, ) count += 1 except Exception: pass return count def record_from_snapshot(self, snapshot: Any) -> None: """Record a MotionSnapshot as an 'observe' action.""" if hasattr(snapshot, "to_dict"): data = snapshot.to_dict() else: data = dict(snapshot) if isinstance(snapshot, dict) else {} self.record_action( action="observe_snapshot", params={ "image_hash": data.get("image_hash", "")[:16], "motion_delta": data.get("motion_delta", 0.0), "cursor_x": data.get("cursor_x", 0), "cursor_y": data.get("cursor_y", 0), }, session_id=data.get("session_id", ""), coherence=data.get("coherence", 0.5), ) # ───────────────────────────────────────────────────────────────────── # Pattern detection # ───────────────────────────────────────────────────────────────────── def _pattern_signature(self, actions: List[Dict[str, Any]]) -> str: """ Compute a signature that groups equivalent action sequences. Only the action NAMES are hashed (not the specific params), so click(500,300)+type("hello") and click(100,200)+type("world") are treated as the same pattern. """ names = "|".join(a["action"] for a in actions) return hashlib.sha1(names.encode()).hexdigest()[:12] def _update_patterns(self) -> None: """Scan the recent window for new pattern matches.""" if len(self._actions) < self.window_size: return # Take the last `window_size` actions as a candidate sequence recent = list(self._actions)[-self.window_size:] signature = self._pattern_signature(recent) self._pattern_counts[signature] += 1 if signature not in self._pattern_sequences: self._pattern_sequences[signature] = [dict(a) for a in recent] self._pattern_metadata[signature] = { "first_seen": time.time(), "sessions": set(), } self._pattern_metadata[signature]["last_seen"] = time.time() for a in recent: sid = a.get("session_id") or "" if sid: self._pattern_metadata[signature]["sessions"].add(sid) # Surface the pattern once it crosses the threshold count = self._pattern_counts[signature] if count >= self.min_occurrences and signature not in self._surfaced: self._surfaced.add(signature) self._create_pattern_record(signature) def _create_pattern_record(self, signature: str) -> None: sequence = self._pattern_sequences.get(signature, []) metadata = self._pattern_metadata.get(signature, {}) count = self._pattern_counts[signature] # Compute mean coherence from the actions in the sequence if sequence: coherence = sum(a.get("coherence", 0) for a in sequence) / len(sequence) else: coherence = 0.0 # Suggest a name from the action names action_names = [a["action"].replace("vm_", "").replace("_", " ") for a in sequence] suggested_name = "_".join(a.replace("vm_", "") for a in [x["action"] for x in sequence]) if len(suggested_name) > 48: suggested_name = suggested_name[:48] + "_etc" suggested_description = f"Observed sequence: {' → '.join(action_names)}" pattern = ObservedPattern( signature=signature, action_sequence=sequence, occurrence_count=count, first_seen=metadata.get("first_seen", time.time()), last_seen=metadata.get("last_seen", time.time()), coherence=coherence, suggested_name=suggested_name, suggested_description=suggested_description, contributing_sessions=sorted(metadata.get("sessions", set())), ) self._pending_patterns.append(pattern) self._total_patterns_detected += 1 logger.info( "ObservedPattern: %s (count=%d, actions=%d)", suggested_name, count, len(sequence), ) # ───────────────────────────────────────────────────────────────────── # Pattern retrieval # ───────────────────────────────────────────────────────────────────── def get_pending_patterns(self, clear: bool = True) -> List[ObservedPattern]: """Return all patterns that are ready for skill proposal.""" with self._lock: patterns = list(self._pending_patterns) if clear: self._pending_patterns.clear() return patterns def get_all_patterns(self) -> List[ObservedPattern]: """Return all known patterns (surfaced or not), sorted by count.""" with self._lock: patterns = [] for sig, count in self._pattern_counts.items(): sequence = self._pattern_sequences.get(sig, []) metadata = self._pattern_metadata.get(sig, {}) coherence = ( sum(a.get("coherence", 0) for a in sequence) / len(sequence) if sequence else 0.0 ) patterns.append(ObservedPattern( signature=sig, action_sequence=sequence, occurrence_count=count, first_seen=metadata.get("first_seen", 0), last_seen=metadata.get("last_seen", 0), coherence=coherence, suggested_name="_".join(a["action"].replace("vm_", "") for a in sequence)[:48], contributing_sessions=sorted(metadata.get("sessions", set())), )) patterns.sort(key=lambda p: p.occurrence_count, reverse=True) return patterns # ───────────────────────────────────────────────────────────────────── # Status # ───────────────────────────────────────────────────────────────────── def get_status(self) -> Dict[str, Any]: with self._lock: avg_coherence = ( sum(self._recent_coherence) / len(self._recent_coherence) if self._recent_coherence else 0.0 ) return { "window_size": self.window_size, "min_occurrences": self.min_occurrences, "actions_recorded": self._total_actions_recorded, "action_buffer": len(self._actions), "unique_patterns": len(self._pattern_counts), "surfaced_patterns": len(self._surfaced), "pending_patterns": len(self._pending_patterns), "total_patterns_detected": self._total_patterns_detected, "avg_recent_coherence": round(avg_coherence, 4), } def reset(self) -> None: with self._lock: self._actions.clear() self._pattern_counts.clear() self._pattern_sequences.clear() self._pattern_metadata.clear() self._surfaced.clear() self._pending_patterns.clear() self._total_actions_recorded = 0 self._total_patterns_detected = 0 self._recent_coherence.clear()