import time from typing import Dict, Any, Optional, Callable, Tuple class InformationCurrencyValidator: """ A core component for Queen AIVA's Genesis Prime Mother evolution, designed to validate the currency (freshness) of information. This validator tracks data freshness, flags outdated information, and can trigger auto-refresh mechanisms to prevent errors caused by stale data. It embodies Patent P2: Proactive Data Currency Management. """ _data_store: Dict[str, Dict[str, Any]] def __init__(self): """ Initializes the InformationCurrencyValidator. The internal data store will hold registered data items along with their last update timestamps and Time To Live (TTL) values. """ self._data_store = {} def register_data(self, data_id: str, value: Any, ttl_seconds: int) -> None: """ Registers or updates a data item with its value and a Time To Live (TTL). This marks the data as fresh at the moment of registration. Args: data_id (str): A unique identifier for the data item. value (Any): The actual data content. ttl_seconds (int): The duration in seconds after which the data is considered stale. """ if not isinstance(ttl_seconds, int) or ttl_seconds <= 0: raise ValueError("TTL must be a positive integer in seconds.") self._data_store[data_id] = { "value": value, "timestamp": time.time(), # Record current time as last update "ttl": ttl_seconds } def _calculate_staleness(self, data_id: str) -> Optional[float]: """ Internal helper to calculate how much time has passed since data was fresh, or how much time is left until it becomes stale. Returns positive float if stale (seconds past TTL), negative float if fresh (seconds until TTL), or None if data_id is not found. """ if data_id not in self._data_store: return None data_info = self._data_store[data_id] elapsed_time = time.time() - data_info["timestamp"] time_until_stale = data_info["ttl"] - elapsed_time return -time_until_stale # Positive if stale, negative if fresh def is_stale(self, data_id: str) -> bool: """ Checks if a registered data item is currently considered stale. Args: data_id (str): The unique identifier for the data item. Returns: bool: True if the data is stale or not found, False otherwise. """ staleness_value = self._calculate_staleness(data_id) if staleness_value is None: # Data not found, can be considered stale or invalid for processing return True return staleness_value > 0 # Positive value means it's past its TTL def get_freshness_status(self, data_id: str) -> Tuple[str, Optional[float]]: """ Provides a detailed freshness status for a data item. Args: data_id (str): The unique identifier for the data item. Returns: Tuple[str, Optional[float]]: A tuple containing the status string ("FRESH", "STALE", "UNKNOWN") and the time remaining until stale (positive) or time past stale (negative) in seconds. Returns (UNKNOWN, None) if data_id is not found. """ staleness_value = self._calculate_staleness(data_id) if staleness_value is None: return "UNKNOWN", None elif staleness_value > 0: return "STALE", staleness_value # How many seconds past TTL else: return "FRESH", abs(staleness_value) # How many seconds remaining until TTL def flag_outdated_data(self, data_id: str) -> Optional[Dict[str, Any]]: """ Explicitly flags data as outdated and returns its current state and staleness details. This directly addresses the "Outdated info flagged" acceptance criteria. Args: data_id (str): The unique identifier for the data item. Returns: Optional[Dict[str, Any]]: A dictionary containing 'status', 'staleness_duration_seconds', and 'value' if data is found and stale. Returns None if data is not found or not stale. """ status, time_info = self.get_freshness_status(data_id) if status == "STALE": return { "status": "STALE", "staleness_duration_seconds": time_info, "value": self._data_store[data_id]["value"] } return None def refresh_data(self, data_id: str, new_value: Any = None) -> bool: """ Manually refreshes a data item, updating its value (if provided) and resetting its timestamp. This can be used by an "Auto-refresh trigger" mechanism. Args: data_id (str): The unique identifier for the data item. new_value (Any, optional): The new value for the data item. If None, the existing value is kept but timestamp is reset. Returns: bool: True if data was successfully refreshed, False if data_id not found. """ if data_id not in self._data_store: return False if new_value is not None: self._data_store[data_id]["value"] = new_value self._data_store[data_id]["timestamp"] = time.time() return True def get_data(self, data_id: str, auto_refresh_if_stale: bool = False, refresh_callback: Optional[Callable[[str], Any]] = None) -> Optional[Any]: """ Retrieves the value of a data item. Can optionally trigger an auto-refresh if the data is stale and a refresh_callback is provided. Args: data_id (str): The unique identifier for the data item. auto_refresh_if_stale (bool): If True, and data is stale, attempts to refresh it. refresh_callback (Optional[Callable[[str], Any]]): A function that takes the data_id and returns the new, fresh value for that data_id. Required if auto_refresh_if_stale is True. Returns: Optional[Any]: The current (potentially refreshed) value of the data item, or None if data_id is not found. """ if data_id not in self._data_store: return None if self.is_stale(data_id) and auto_refresh_if_stale: if refresh_callback: try: new_value = refresh_callback(data_id) self.refresh_data(data_id, new_value) except Exception as e: # In a production system, this would log the error. # For AIVA, this is a learning opportunity for resilience. pass else: # Warning: Data is stale, but no refresh_callback provided for auto-refresh. pass return self._data_store[data_id]["value"] # Example Usage (for testing and demonstration within the file) if __name__ == "__main__": print("--- Genesis Flash 2.0: Information Currency Validator Test ---") validator = InformationCurrencyValidator() # --- Acceptance Criteria 1: Data freshness tracking --- print("\n[TEST] Data Freshness Tracking:") validator.register_data("sensor_feed_alpha", {"temp": 25.5, "humidity": 60}, ttl_seconds=2) validator.register_data("config_setting_beta", {"threshold": 0.75, "mode": "adaptive"}, ttl_seconds=10) print(f"Initial status for 'sensor_feed_alpha': {validator.get_freshness_status('sensor_feed_alpha')}") print(f"Initial status for 'config_setting_beta': {validator.get_freshness_status('config_setting_beta')}") time.sleep(1.5) # Wait for some time, but less than sensor_feed_alpha's TTL print(f"Status after 1.5s for 'sensor_feed_alpha': {validator.get_freshness_status('sensor_feed_alpha')}") print(f"Is 'sensor_feed_alpha' stale? {validator.is_stale('sensor_feed_alpha')}") time.sleep(1.0) # Wait past sensor_feed_alpha's TTL (total 2.5s) print(f"Status after 2.5s for 'sensor_feed_alpha': {validator.get_freshness_status('sensor_feed_alpha')}") print(f"Is 'sensor_feed_alpha' stale? {validator.is_stale('sensor_feed_alpha')}") print(f"Is 'config_setting_beta' stale? {validator.is_stale('config_setting_beta')}") # Should still be fresh # --- Acceptance Criteria 2: Outdated info flagged --- print("\n[TEST] Outdated Info Flagging:") outdated_alpha = validator.flag_outdated_data("sensor_feed_alpha") if outdated_alpha: print(f"FLAGGED: sensor_feed_alpha is outdated. Details: {outdated_alpha}") else: print("sensor_feed_alpha is not flagged as outdated (unexpected if stale).") not_outdated_beta = validator.flag_outdated_data("config_setting_beta") if not_outdated_beta: print("config_setting_beta was unexpectedly flagged as outdated.") else: print("config_setting_beta is correctly not flagged as outdated.") # --- Acceptance Criteria 3: Auto-refresh triggers --- print("\n[TEST] Auto-Refresh Triggers:") # Define a mock refresh callback def mock_sensor_refresh(data_id: str) -> Dict[str, Any]: print(f" --> Mocking external refresh for {data_id}...") return {"temp": 26.0, "humidity": 62, "last_refreshed": time.time()} # Try to get stale data with auto-refresh print(f"Current 'sensor_feed_alpha' value (stale): {validator.get_data('sensor_feed_alpha')}") refreshed_data = validator.get_data("sensor_feed_alpha", auto_refresh_if_stale=True, refresh_callback=mock_sensor_refresh) print(f"Retrieved 'sensor_feed_alpha' with auto-refresh: {refreshed_data}") print(f"Status after auto-refresh for 'sensor_feed_alpha': {validator.get_freshness_status('sensor_feed_alpha')}") print(f"Is 'sensor_feed_alpha' stale after refresh? {validator.is_stale('sensor_feed_alpha')}") # Test manual refresh print("\n[TEST] Manual Refresh:") time.sleep(2.5) # Make 'sensor_feed_alpha' stale again, 'config_setting_beta' still fresh print(f"Status before manual refresh for 'config_setting_beta': {validator.get_freshness_status('config_setting_beta')}") print(f"Is 'sensor_feed_alpha' stale again? {validator.is_stale('sensor_feed_alpha')}") validator.refresh_data("sensor_feed_alpha", {"temp": 25.8, "humidity": 61, "source": "manual"}) print(f"Status after manual refresh for 'sensor_feed_alpha': {validator.get_freshness_status('sensor_feed_alpha')}") print(f"Is 'sensor_feed_alpha' stale after manual refresh? {validator.is_stale('sensor_feed_alpha')}") print(f"Manually refreshed 'sensor_feed_alpha' value: {validator.get_data('sensor_feed_alpha')}") print("\n--- Information Currency Validator Test Complete ---")