import uuid from enum import Enum from typing import List, Dict, Optional, Set class AssumptionValidity(Enum): """ Represents the current validity status of an assumption. """ UNKNOWN = "UNKNOWN" PENDING_VALIDATION = "PENDING_VALIDATION" TRUE = "TRUE" FALSE = "FALSE" MITIGATED = "MITIGATED" # For assumptions that were false but addressed class Assumption: """ Represents a single assumption that needs to be surfaced and validated. """ def __init__(self, description: str, source: Optional[str] = None, initial_validity: AssumptionValidity = AssumptionValidity.UNKNOWN, dependencies: Optional[List[str]] = None): self.id: str = str(uuid.uuid4()) self.description: str = description self.source: Optional[str] = source # Where did this assumption come from? (e.g., 'historical data', 'expert opinion') self.validity: AssumptionValidity = initial_validity self.validation_notes: List[str] = [] # Dependencies can be other assumption IDs or external conditions/data points. self.dependencies: Set[str] = set(dependencies) if dependencies else set() def update_validity(self, new_validity: AssumptionValidity, note: Optional[str] = None): """ Updates the validity status of the assumption and logs a note. """ self.validity = new_validity if note: self.validation_notes.append(f"[{new_validity.value}] {note}") def add_dependency(self, dependency_id: str): """ Adds a dependency to this assumption. """ self.dependencies.add(dependency_id) def remove_dependency(self, dependency_id: str): """ Removes a dependency from this assumption. """ self.dependencies.discard(dependency_id) def to_dict(self) -> Dict: return { "id": self.id, "description": self.description, "source": self.source, "validity": self.validity.value, "validation_notes": self.validation_notes, "dependencies": list(self.dependencies) } class DecisionContext: """ Encapsulates a specific decision and the assumptions relevant to it. """ def __init__(self, decision_goal: str, context_description: Optional[str] = None): self.id: str = str(uuid.uuid4()) self.decision_goal: str = decision_goal self.context_description: Optional[str] = context_description self.assumption_ids: Set[str] = set() # Store IDs to link to a central AssumptionManager def add_assumption_id(self, assumption_id: str): """ Links an assumption to this decision context. """ self.assumption_ids.add(assumption_id) def remove_assumption_id(self, assumption_id: str): """ Unlinks an assumption from this decision context. """ self.assumption_ids.discard(assumption_id) def to_dict(self) -> Dict: return { "id": self.id, "decision_goal": self.decision_goal, "context_description": self.context_description, "assumption_ids": list(self.assumption_ids) } class AssumptionManager: """ Manages all assumptions and decision contexts within the AIVA system. Provides tools for surfacing, validating, and tracking assumptions. """ def __init__(self): self._assumptions: Dict[str, Assumption] = {} self._decision_contexts: Dict[str, DecisionContext] = {} def create_assumption(self, description: str, source: Optional[str] = None, initial_validity: AssumptionValidity = AssumptionValidity.UNKNOWN, dependencies: Optional[List[str]] = None) -> Assumption: """ Creates a new assumption and adds it to the manager. """ assumption = Assumption(description, source, initial_validity, dependencies) self._assumptions[assumption.id] = assumption return assumption def get_assumption(self, assumption_id: str) -> Optional[Assumption]: """ Retrieves an assumption by its ID. """ return self._assumptions.get(assumption_id) def update_assumption_validity(self, assumption_id: str, new_validity: AssumptionValidity, note: Optional[str] = None) -> bool: """ Updates the validity of an existing assumption. Returns True if updated, False if assumption not found. """ assumption = self.get_assumption(assumption_id) if assumption: assumption.update_validity(new_validity, note) return True return False def create_decision_context(self, decision_goal: str, context_description: Optional[str] = None) -> DecisionContext: """ Creates a new decision context. """ context = DecisionContext(decision_goal, context_description) self._decision_contexts[context.id] = context return context def get_decision_context(self, context_id: str) -> Optional[DecisionContext]: """ Retrieves a decision context by its ID. """ return self._decision_contexts.get(context_id) def link_assumption_to_decision(self, assumption_id: str, context_id: str) -> bool: """ Links an existing assumption to a decision context. Returns True if linked, False if either ID is invalid. """ assumption = self.get_assumption(assumption_id) context = self.get_decision_context(context_id) if assumption and context: context.add_assumption_id(assumption_id) return True return False def get_assumptions_for_decision(self, context_id: str) -> List[Assumption]: """ Retrieves all assumptions linked to a specific decision context. """ context = self.get_decision_context(context_id) if context: return [self._assumptions[aid] for aid in context.assumption_ids if aid in self._assumptions] return [] def check_decision_readiness(self, context_id: str) -> Dict[str, any]: """ Evaluates the readiness of a decision based on the validity of its linked assumptions. A decision is considered 'ready' if all its directly linked assumptions are TRUE or MITIGATED. It also checks the validity of dependent assumptions. """ context = self.get_decision_context(context_id) if not context: return {"ready": False, "reason": "Decision context not found.", "summary": {}} assumptions_for_decision = self.get_assumptions_for_decision(context_id) if not assumptions_for_decision: return {"ready": True, "reason": "No assumptions linked to this decision.", "summary": {}} all_assumptions_involved: Set[Assumption] = set(assumptions_for_assumptions_for_decision) # Recursively find all dependent assumptions processed_ids = set() queue = [a.id for a in assumptions_for_decision] while queue: current_id = queue.pop(0) if current_id in processed_ids: continue processed_ids.add(current_id) current_assumption = self.get_assumption(current_id) if current_assumption: all_assumptions_involved.add(current_assumption) for dep_id in current_assumption.dependencies: if dep_id not in processed_ids: queue.append(dep_id) summary: Dict[str, Dict] = {} all_ready = True reasons = [] for assumption in all_assumptions_involved: summary[assumption.id] = assumption.to_dict() if assumption.validity not in [AssumptionValidity.TRUE, AssumptionValidity.MITIGATED]: all_ready = False reasons.append(f"Assumption '{assumption.description}' (ID: {assumption.id}) is {assumption.validity.value}.") return { "ready": all_ready, "reason": "All critical assumptions are validated." if all_ready else "Some assumptions require validation or mitigation.", "issues": reasons if not all_ready else [], "summary": {a.id: a.to_dict() for a in all_assumptions_involved} } def get_all_assumptions(self) -> List[Assumption]: """Returns all assumptions managed.""" return list(self._assumptions.values()) def get_all_decision_contexts(self) -> List[DecisionContext]: """Returns all decision contexts managed.""" return list(self._decision_contexts.values())