diff --git a/pievo/group/evolveflow.py b/pievo/group/evolveflow.py index 4e1d612..877feaa 100644 --- a/pievo/group/evolveflow.py +++ b/pievo/group/evolveflow.py @@ -68,6 +68,7 @@ def __init__( initial_absolute_sigma: float = 0.01, llm_judge: bool = False, off_pievo: bool = False, + guard: Optional[Any] = None, ): self.submissions: List[Dict[str, Any]] = [] self.output_dir = output_dir @@ -156,6 +157,7 @@ def __init__( self.hypothesis_form = {} self.off_pievo = off_pievo + self.guard = guard # Optional SemanticGuard for variable-level alignment checks self.extra_clients: Dict[str, OpenAIChatCompletionClient] = {} def register_client(self, client: Any, name: str): @@ -851,6 +853,50 @@ async def update_pievo_state(self) -> None: # Extract the new principle from the self.submissions, and assign the initial belief for handling. new_principle_ids = self._extract_principles() + + # ── Semantic Guard: validate new principles before Bayesian update ── + if self.guard is not None and new_principle_ids: + from pievo.guard import Principle as GuardPrinciple + phenomenon = self.task # Use task description as phenomenon context + validated_ids = [] + blocked_ids = [] + for pid in new_principle_ids: + ptext = self.principles.get(pid, "") + new_p = GuardPrinciple(id=pid, text=ptext, phenomenon=phenomenon) + decisions = await self.guard.check_against_all( + new_p, self.principles, phenomenon + ) + if self.guard.any_blocked(decisions): + blocked_ids.append(pid) + blocked_reasons = [ + d.reason for d in decisions if d.action == "BLOCK" + ] + pievo_log( + f"SEMANTIC GUARD: BLOCKED principle '{pid}' — " + f"semantic conflict with existing principles. " + f"Reasons: {blocked_reasons}", + source="semantic_guard", + tag="blocked", + level="WARN", + ) + logger.warning( + f"SEMANTIC GUARD: BLOCKED principle '{pid}' — {blocked_reasons}" + ) + else: + validated_ids.append(pid) + pievo_log( + f"SEMANTIC GUARD: ACCEPTED principle '{pid}'", + source="semantic_guard", + tag="proceed", + ) + + if blocked_ids: + # Remove blocked principles from self.principles so they do not + # participate in the Bayesian update. + for pid in blocked_ids: + self.principles.pop(pid, None) + new_principle_ids = validated_ids + self._handle_new_principles(new_principle_ids) self._extract_history() diff --git a/pievo/guard/__init__.py b/pievo/guard/__init__.py new file mode 100644 index 0000000..486bb6a --- /dev/null +++ b/pievo/guard/__init__.py @@ -0,0 +1,50 @@ +""" +Semantic Guard for PiEvo -- Optional plugin for variable-level semantic alignment +detection between LLM-generated principles before Bayesian comparison. + +Usage: + from pievo.guard import SemanticGuard, StrictPolicy, LLMExtractor + guard = SemanticGuard(extractor=LLMExtractor(client=openai_client), policy=StrictPolicy()) + pievo = PiEvo(task=..., client=..., guard=guard) +""" + +from pievo.guard.signature import ( + OperationalDefinition, + VariableSignature, + EvidentialStatus, + Principle, + PerVariableAlignment, + AlignmentReport, + NO_SHARED_VARIABLES, +) +from pievo.guard.alignment import ( + SimilarityMethod, + text_similarity, + variable_alignment, + compute_alignment_vector, +) +from pievo.guard.policy import ( + GuardPolicy, + StrictPolicy, + PermissivePolicy, + ConservativePolicy, +) +from pievo.guard.extractor import ( + SignatureExtractor, + LLMExtractor, + ManualExtractor, + ExtractionError, +) +from pievo.guard.guard import ( + SemanticGuard, + AlignmentDecision, +) + +__all__ = [ + "OperationalDefinition", "VariableSignature", "EvidentialStatus", + "Principle", "PerVariableAlignment", "AlignmentReport", "NO_SHARED_VARIABLES", + "SimilarityMethod", "text_similarity", "variable_alignment", "compute_alignment_vector", + "GuardPolicy", "StrictPolicy", "PermissivePolicy", "ConservativePolicy", + "SignatureExtractor", "LLMExtractor", "ManualExtractor", "ExtractionError", + "SemanticGuard", "AlignmentDecision", +] diff --git a/pievo/guard/alignment.py b/pievo/guard/alignment.py new file mode 100644 index 0000000..eee8d9e --- /dev/null +++ b/pievo/guard/alignment.py @@ -0,0 +1,192 @@ +""" +Multi-dimensional alignment vector computation (Spec Section 3.3). +""" + +from __future__ import annotations + +import logging +from enum import Enum +from typing import Any, Callable, Dict, List, Optional, Set, Tuple + +from pievo.guard.signature import ( + VariableSignature, + OperationalDefinition, + PerVariableAlignment, + AlignmentReport, + NO_SHARED_VARIABLES, + op_def_eq, +) + +logger = logging.getLogger(__name__) + + +class SimilarityMethod(Enum): + EXACT = "exact" + TOKEN = "token" + EMBEDDING = "embedding" + + +def _tokenize(text: str) -> Set[str]: + return set(text.lower().split()) + + +def text_similarity( + text1: str, + text2: str, + method: SimilarityMethod = SimilarityMethod.EXACT, + threshold: float = 0.7, + embedding_fn: Optional[Callable[[str], List[float]]] = None, +) -> Tuple[float, str]: + """Compute similarity between two text strings.""" + if method == SimilarityMethod.EXACT: + return (1.0 if text1 == text2 else 0.0), "exact" + + elif method == SimilarityMethod.TOKEN: + tokens1 = _tokenize(text1) + tokens2 = _tokenize(text2) + if not tokens1 or not tokens2: + return 0.0, "token" + intersection = tokens1 & tokens2 + union = tokens1 | tokens2 + jaccard = len(intersection) / len(union) + return jaccard if jaccard >= threshold else 0.0, "token" + + elif method == SimilarityMethod.EMBEDDING: + if embedding_fn is None: + logger.warning("EMBEDDING mode requires embedding_fn, falling back to TOKEN") + return text_similarity(text1, text2, SimilarityMethod.TOKEN, threshold) + vec1 = embedding_fn(text1) + vec2 = embedding_fn(text2) + if not vec1 or not vec2: + return 0.0, "embedding" + dot = sum(a * b for a, b in zip(vec1, vec2)) + norm1 = sum(a * a for a in vec1) ** 0.5 + norm2 = sum(b * b for b in vec2) ** 0.5 + if norm1 == 0 or norm2 == 0: + return 0.0, "embedding" + cosine = max(0.0, min(1.0, dot / (norm1 * norm2))) + return cosine if cosine >= threshold else 0.0, "embedding" + + raise ValueError(f"Unknown similarity method: {method}") + + +def variable_alignment( + s1: VariableSignature, + s2: VariableSignature, + similarity_method: SimilarityMethod = SimilarityMethod.EXACT, + similarity_threshold: float = 0.7, + embedding_fn: Optional[Callable[[str], List[float]]] = None, +) -> Dict[str, Any]: + """Per-variable alignment with separate confidence reporting.""" + if similarity_method == SimilarityMethod.EXACT: + ref_match = 1.0 if s1.referent == s2.referent else 0.0 + else: + ref_match, _ = text_similarity( + s1.referent, s2.referent, + method=similarity_method, + threshold=similarity_threshold, + embedding_fn=embedding_fn, + ) + + op_match = 1.0 if op_def_eq(s1.operational_definition, s2.operational_definition) else 0.0 + ev_match = 1.0 if s1.evidential_status == s2.evidential_status else 0.0 + confidence = min(s1.confidence, s2.confidence) + uncertain = confidence < 0.8 + + return { + "referential": ref_match, + "operational": op_match, + "evidential": ev_match, + "confidence": confidence, + "uncertain": uncertain, + "referent_1": s1.referent, + "referent_2": s2.referent, + "op_1": s1.operational_definition.to_dict(), + "op_2": s2.operational_definition.to_dict(), + "ref_match_raw": ref_match, + "op_match_raw": op_match, + "ev_match_raw": ev_match, + } + + +def compute_alignment_vector( + signatures_1: List[VariableSignature], + signatures_2: List[VariableSignature], + phenomenon: str, + similarity_method: SimilarityMethod = SimilarityMethod.EXACT, + similarity_threshold: float = 0.7, + embedding_fn: Optional[Callable[[str], List[float]]] = None, +) -> AlignmentReport: + """Compute 3D alignment vector between two sets of variable signatures.""" + sig1_by_name: Dict[str, VariableSignature] = {s.name: s for s in signatures_1} + sig2_by_name: Dict[str, VariableSignature] = {s.name: s for s in signatures_2} + + shared_names = sorted(set(sig1_by_name.keys()) & set(sig2_by_name.keys())) + + if not shared_names: + return AlignmentReport( + vector=(0.0, 0.0, 0.0), + per_variable=[], + shared_count=0, + phenomenon=phenomenon, + policy_recommendation=NO_SHARED_VARIABLES, + has_uncertain=False, + ) + + per_variable_results: List[PerVariableAlignment] = [] + ref_scores: List[float] = [] + op_scores: List[float] = [] + ev_scores: List[float] = [] + has_uncertain = False + + for name in shared_names: + s1 = sig1_by_name[name] + s2 = sig2_by_name[name] + + result = variable_alignment( + s1, s2, + similarity_method=similarity_method, + similarity_threshold=similarity_threshold, + embedding_fn=embedding_fn, + ) + + ref_scores.append(result["referential"]) + op_scores.append(result["operational"]) + ev_scores.append(result["evidential"]) + if result["uncertain"]: + has_uncertain = True + + pva = PerVariableAlignment( + name=name, + referential=result["referential"], + operational=result["operational"], + evidential=result["evidential"], + uncertain=result["uncertain"], + confidence=result["confidence"], + details={ + "referent_1": result["referent_1"], + "referent_2": result["referent_2"], + "op_1": result["op_1"], + "op_2": result["op_2"], + "ref_match_raw": result["ref_match_raw"], + "op_match_raw": result["op_match_raw"], + "ev_match_raw": result["ev_match_raw"], + "extraction_confidence": result["confidence"], + }, + ) + per_variable_results.append(pva) + + n = len(shared_names) + vector = ( + sum(ref_scores) / n, + sum(op_scores) / n, + sum(ev_scores) / n, + ) + + return AlignmentReport( + vector=vector, + per_variable=per_variable_results, + shared_count=n, + phenomenon=phenomenon, + has_uncertain=has_uncertain, + ) diff --git a/pievo/guard/extractor.py b/pievo/guard/extractor.py new file mode 100644 index 0000000..83490d9 --- /dev/null +++ b/pievo/guard/extractor.py @@ -0,0 +1,331 @@ +""" +LLM-based signature extractor adapted for PiEvo's OpenAIChatCompletionClient. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import hashlib +import re +from abc import ABC, abstractmethod +from typing import Any, Dict, List, Optional + +from pievo.guard.signature import ( + VariableSignature, + OperationalDefinition, + EvidentialStatus, +) + +logger = logging.getLogger(__name__) + +# -- Prompt template with few-shot examples demonstrating the core concept -- +# The Phlogiston vs. Oxidation pair is critical: both principles use +# "mass_change" but the referent (physical meaning) is fundamentally different. +EXTRACTION_PROMPT = r"""You are a precise scientific variable extractor. + +Given a scientific principle, extract all observable variables that are used +to explain or measure the target phenomenon. + +CRITICAL RULES: +1. Use EXACT variable names from the text, not paraphrases. If the text says + "mass_change", extract "mass_change", not "weight". +2. The referent must be a precise one-sentence physical description. This is + the MOST IMPORTANT field -- it distinguishes theories that use the same + variable name but mean different things. Be very specific. +3. Extract EVERY variable that relates to the target phenomenon, not just the + main one. +4. Confidence MUST reflect how clearly the variable is described in the text: + - 0.9-1.0: explicitly named and defined + - 0.7-0.9: mentioned but definition is implicit + - 0.5-0.7: inferred from context + - Do NOT use confidence below 0.5. + +For each variable, provide: +- name: the exact term as used in the text +- referent: one-sentence physical description of what it denotes +- operational_definition: + - protocol: measurement protocol + - instrument: measuring instrument + - unit: unit of measurement + - environment: environmental constraints +- evidential_status: "primitive" if directly observed, "derived" if computed, + "auxiliary" if theoretical construct without direct measurement +- confidence: your confidence in this extraction, 0.0 to 1.0 + +FEW-SHOT EXAMPLES (study the referent specificity): + +Example 1 -- Phlogiston theory: +Text: "Metals contain phlogiston. When burned, phlogiston escapes, leaving +calx. Mass remains constant because phlogiston is replaced by air." +Phenomenon: combustion +Output: +[ + { + "name": "mass_change", + "referent": "weight of the dephlogisticated metal after burning, measured in open air", + "operational_definition": { + "protocol": "spring_scale_measurement", + "instrument": "balance", + "unit": "gram", + "environment": "open_air" + }, + "evidential_status": "primitive", + "confidence": 0.92 + }, + { + "name": "phlogiston", + "referent": "fire-like substance contained in combustible bodies, released upon burning", + "operational_definition": { + "protocol": "inferred_from_mass_change", + "instrument": "none", + "unit": "none", + "environment": "none" + }, + "evidential_status": "auxiliary", + "confidence": 0.85 + } +] + +Example 2 -- Oxidation theory (NOTICE different referent for "mass_change"): +Text: "Metals combine with oxygen during combustion to form metal oxides. +Mass increases because oxygen from the air bonds with the metal." +Phenomenon: combustion +Output: +[ + { + "name": "mass_change", + "referent": "weight of the metal oxide product after burning, measured in sealed vessel", + "operational_definition": { + "protocol": "spring_scale_measurement", + "instrument": "analytical_balance", + "unit": "gram", + "environment": "sealed_vessel" + }, + "evidential_status": "primitive", + "confidence": 0.95 + }, + { + "name": "oxygen", + "referent": "gaseous element from the air that combines with metal during combustion", + "operational_definition": { + "protocol": "volume_measurement", + "instrument": "gas_collector", + "unit": "milliliter", + "environment": "sealed_vessel" + }, + "evidential_status": "primitive", + "confidence": 0.93 + } +] + +Example 3 -- Caloric theory: +Text: "Heat is a self-repulsive fluid called caloric. Temperature directly measures caloric fluid density." +Phenomenon: heat_transfer +Output: +[ + { + "name": "temperature", + "referent": "density of caloric fluid in a substance, directly measured", + "operational_definition": { + "protocol": "thermometer_reading", + "instrument": "mercury_thermometer", + "unit": "celsius", + "environment": "thermal_equilibrium" + }, + "evidential_status": "primitive", + "confidence": 0.93 + } +] + +Example 4 -- Kinetic theory (NOTICE different evidential_status for same variable): +Text: "Heat is kinetic energy of molecular motion. Temperature is the average kinetic energy of particles." +Phenomenon: heat_transfer +Output: +[ + { + "name": "temperature", + "referent": "average kinetic energy of constituent particles, computed from velocity distribution", + "operational_definition": { + "protocol": "thermometer_reading", + "instrument": "mercury_thermometer", + "unit": "celsius", + "environment": "thermal_equilibrium" + }, + "evidential_status": "derived", + "confidence": 0.95 + } +] + +Return ONLY a JSON array. No markdown, no explanation. + +Principle text: +{principle_text} + +Target phenomenon: +{phenomenon}""" + + +class SignatureExtractor(ABC): + """Abstract interface for variable signature extraction.""" + + @abstractmethod + def extract(self, principle_text: str, phenomenon: str) -> List[VariableSignature]: ... + + @abstractmethod + async def extract_async(self, principle_text: str, phenomenon: str) -> List[VariableSignature]: ... + + +class ExtractionError(Exception): + """Raised when LLM extraction fails after all retries.""" + + +class ManualExtractor(SignatureExtractor): + """ + Manual/hand-crafted extractor for ablation studies. + + Returns preloaded signatures keyed by principle text, bypassing the LLM + entirely. Useful for controlled experiments where extraction drift must + be eliminated. + """ + + def __init__(self, preloaded_signatures: Dict[str, List[VariableSignature]]): + self._cache = preloaded_signatures + + def extract(self, principle_text: str, phenomenon: str) -> List[VariableSignature]: + return list(self._cache.get(principle_text, [])) + + async def extract_async(self, principle_text: str, phenomenon: str) -> List[VariableSignature]: + return self.extract(principle_text, phenomenon) + + +class LLMExtractor(SignatureExtractor): + """ + LLM-based signature extractor using PiEvo's OpenAIChatCompletionClient. + + Usage: + from autogen_ext.models.openai import OpenAIChatCompletionClient + client = OpenAIChatCompletionClient(model="gpt-4o", ...) + extractor = LLMExtractor(client=client) + sigs = await extractor.extract_async("Metals burn...", "combustion") + """ + + def __init__( + self, + client, + prompt_template: Optional[str] = None, + max_retries: int = 2, + retry_delay: float = 0.5, + ): + self._client = client + self._prompt = prompt_template or EXTRACTION_PROMPT + self._max_retries = max_retries + self._retry_delay = retry_delay + self._cache: Dict[str, List[VariableSignature]] = {} + + def _cache_key(self, text: str, phenomenon: str) -> str: + return hashlib.sha256(f"{phenomenon}::{text}".encode()).hexdigest()[:16] + + async def extract_async(self, principle_text: str, phenomenon: str) -> List[VariableSignature]: + """Async extraction using PiEvo's OpenAIChatCompletionClient.""" + key = self._cache_key(principle_text, phenomenon) + if key in self._cache: + return self._cache[key] + + prompt = self._prompt.format(principle_text=principle_text, phenomenon=phenomenon) + + for attempt in range(self._max_retries + 1): + try: + from autogen_core.models import UserMessage + + # Use autogen's UserMessage for proper client API compatibility + result = await self._client.create( + messages=[UserMessage(content=prompt, source="guard")], + ) + content = result.content if hasattr(result, "content") else str(result) + + signatures = self._parse_response(content) + if signatures: + self._cache[key] = signatures + logger.info( + "Extracted %d signatures (attempt %d)", + len(signatures), attempt + 1, + ) + return signatures + + if attempt < self._max_retries: + await asyncio.sleep(self._retry_delay) + + except Exception as e: + logger.warning("LLM extraction attempt %d failed: %s", attempt + 1, e) + if attempt < self._max_retries: + await asyncio.sleep(self._retry_delay) + + logger.error("LLM extraction failed after all retries") + return [] + + def extract(self, principle_text: str, phenomenon: str) -> List[VariableSignature]: + """Synchronous fallback (PiEvo is async by default; this is for testing).""" + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return asyncio.run(self.extract_async(principle_text, phenomenon)) + # Running inside an event loop: schedule via create_task (blocking the + # current thread is acceptable for sync testing scenarios). + import concurrent.futures + future = asyncio.run_coroutine_threadsafe( + self.extract_async(principle_text, phenomenon), loop + ) + return future.result(timeout=120) + + def _parse_response(self, content: str) -> List[VariableSignature]: + """Parse LLM JSON response into VariableSignature objects.""" + # Strip markdown code fences if present + content = content.strip() + if content.startswith("```"): + lines = content.split("\n") + content = "\n".join(lines[1:]) if len(lines) > 1 else "" + if content.endswith("```"): + content = content[:-3] + content = content.strip() + + try: + data = json.loads(content) + except json.JSONDecodeError: + # Try to find JSON array in the response + match = re.search(r"\[.*\]", content, re.DOTALL) + if match: + try: + data = json.loads(match.group()) + except json.JSONDecodeError: + return [] + else: + return [] + + if not isinstance(data, list): + return [] + + signatures = [] + for item in data: + try: + op_def = item.get("operational_definition", {}) + signatures.append(VariableSignature( + name=item["name"], + referent=item["referent"], + operational_definition=OperationalDefinition( + protocol=op_def.get("protocol", "unknown"), + instrument=op_def.get("instrument", "unknown"), + unit=op_def.get("unit", "unknown"), + environment=op_def.get("environment", "unknown"), + ), + evidential_status=EvidentialStatus( + item.get("evidential_status", "primitive") + ), + confidence=float(item.get("confidence", 0.8)), + )) + except (KeyError, TypeError, ValueError) as e: + logger.warning("Skipping malformed signature %s: %s", item, e) + continue + + return signatures diff --git a/pievo/guard/guard.py b/pievo/guard/guard.py new file mode 100644 index 0000000..34ece3c --- /dev/null +++ b/pievo/guard/guard.py @@ -0,0 +1,232 @@ +""" +Semantic Guard — main orchestration class (Spec Section 4.1). +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import Any, Callable, Dict, List, Optional + +from pievo.guard.signature import ( + Principle, + PerVariableAlignment, + AlignmentReport, + NO_SHARED_VARIABLES, +) +from pievo.guard.alignment import ( + SimilarityMethod, + compute_alignment_vector, +) +from pievo.guard.policy import GuardPolicy +from pievo.guard.extractor import SignatureExtractor + +logger = logging.getLogger(__name__) + + +@dataclass +class AlignmentDecision: + """Result of a guard check between two principles.""" + report: AlignmentReport + action: str # PROCEED | RESTRICT | BLOCK | ABSTAIN | NO_SHARED_VARIABLES + p1_id: str + p2_id: str + misaligned_variables: List[str] + reason: str = "" + + def to_dict(self) -> Dict[str, Any]: + return { + "principle_pair": [self.p1_id, self.p2_id], + "action": self.action, + "misaligned_variables": self.misaligned_variables, + "reason": self.reason, + "report": self.report.to_dict(), + } + + +class SemanticGuard: + """ + Semantic guard hook for PiEvo's Bayesian update pipeline. + + Usage within PiEvo.evolveflow.PiEvo: + self.guard = SemanticGuard(extractor, StrictPolicy()) + """ + + def __init__( + self, + extractor: SignatureExtractor, + policy: GuardPolicy, + similarity_method: SimilarityMethod = SimilarityMethod.EXACT, + similarity_threshold: float = 0.7, + embedding_fn: Optional[Callable] = None, + ): + self._extractor = extractor + self._policy = policy + self._similarity_method = similarity_method + self._similarity_threshold = similarity_threshold + self._embedding_fn = embedding_fn + self._log: List[Dict[str, Any]] = [] + # Cache: principle_id -> List[VariableSignature] + self._signature_cache: Dict[str, List] = {} + + async def extract_signatures( + self, principle_id: str, principle_text: str, phenomenon: str + ) -> List: + """Extract and cache variable signatures for a principle.""" + if principle_id not in self._signature_cache: + sigs = await self._extractor.extract_async(principle_text, phenomenon) + self._signature_cache[principle_id] = sigs + logger.info( + "Guard: extracted %d signatures for '%s'", + len(sigs), principle_id, + ) + return self._signature_cache[principle_id] + + async def check_pair( + self, + p1: Principle, + p2: Principle, + phenomenon: str = "", + ) -> AlignmentDecision: + """ + Check semantic alignment between two principles. + + Returns AlignmentDecision with action (PROCEED/RESTRICT/BLOCK/ABSTAIN/NO_SHARED_VARIABLES). + """ + phenom = phenomenon or p1.phenomenon or p2.phenomenon + logger.info("Guard: checking %s vs %s", p1.id, p2.id) + + # Extract signatures (cached) + sigs1 = await self.extract_signatures(p1.id, p1.text, phenom) + sigs2 = await self.extract_signatures(p2.id, p2.text, phenom) + + if not sigs1 or not sigs2: + decision = AlignmentDecision( + report=AlignmentReport( + vector=(0.0, 0.0, 0.0), + per_variable=[], + shared_count=0, + principle_pair=(p1.id, p2.id), + phenomenon=phenom, + policy_recommendation="BLOCK", + has_uncertain=True, + ), + action="BLOCK", + p1_id=p1.id, + p2_id=p2.id, + misaligned_variables=[], + reason="Signature extraction failed for one or both principles", + ) + self._log.append(decision.to_dict()) + return decision + + # Compute alignment vector + report = compute_alignment_vector( + sigs1, + sigs2, + phenom, + similarity_method=self._similarity_method, + similarity_threshold=self._similarity_threshold, + embedding_fn=self._embedding_fn, + ) + report.principle_pair = (p1.id, p2.id) + + # Handle NO_SHARED_VARIABLES sentinel + if report.policy_recommendation == NO_SHARED_VARIABLES: + decision = AlignmentDecision( + report=report, + action=NO_SHARED_VARIABLES, + p1_id=p1.id, + p2_id=p2.id, + misaligned_variables=[], + reason=f"No shared variables between {p1.id} and {p2.id}", + ) + else: + # Apply policy + action = self._policy.decide(report) + report.policy_recommendation = action + + # Collect misaligned variables + misaligned = self._collect_misaligned(report) + + decision = AlignmentDecision( + report=report, + action=action, + p1_id=p1.id, + p2_id=p2.id, + misaligned_variables=misaligned, + reason=self._reason_for(action, report), + ) + + self._log.append(decision.to_dict()) + return decision + + async def check_against_all( + self, + new_principle: Principle, + existing_principles: Dict[str, str], # id -> text + phenomenon: str = "", + ) -> List[AlignmentDecision]: + """ + Check a new principle against all existing principles. + Returns list of decisions (one per existing principle). + """ + decisions = [] + for pid, ptext in existing_principles.items(): + if pid == new_principle.id: + continue + existing = Principle(id=pid, text=ptext, phenomenon=phenomenon) + decision = await self.check_pair(new_principle, existing, phenomenon) + decisions.append(decision) + return decisions + + def any_blocked(self, decisions: List[AlignmentDecision]) -> bool: + """Check if any decision blocks the new principle.""" + return any(d.action == "BLOCK" for d in decisions) + + def collect_misaligned_variables(self, decisions: List[AlignmentDecision]) -> Dict[str, List[str]]: + """Collect misaligned variable names grouped by conflicting principle.""" + conflicts: Dict[str, List[str]] = {} + for d in decisions: + if d.action in ("BLOCK", "RESTRICT"): + conflicts[d.p2_id] = d.misaligned_variables + return conflicts + + def get_log(self) -> List[Dict[str, Any]]: + return list(self._log) + + def clear_log(self) -> None: + self._log.clear() + + def _collect_misaligned(self, report: AlignmentReport) -> List[str]: + """Collect variable names where any alignment dimension < 1.0.""" + misaligned = [] + for pva in report.per_variable: + if pva.referential < 1.0 or pva.operational < 1.0 or pva.evidential < 1.0: + misaligned.append(pva.name) + return misaligned + + @staticmethod + def _reason_for(action: str, report: AlignmentReport) -> str: + """Generate human-readable reason for the decision.""" + if action == "PROCEED": + return "All variable signatures aligned" + if action == "BLOCK": + mis = report.per_variable + issues = [] + for pva in mis: + parts = [] + if pva.referential < 1.0: + parts.append(f"referential={pva.referential:.0f}") + if pva.operational < 1.0: + parts.append(f"operational={pva.operational:.0f}") + if pva.evidential < 1.0: + parts.append(f"evidential={pva.evidential:.0f}") + if parts: + issues.append(f"{pva.name}({','.join(parts)})") + return f"Misaligned variables: {', '.join(issues)}" if issues else "Alignment vector below threshold" + if action == "RESTRICT": + return "Partial alignment — some variables comparable, some misaligned" + if action == "ABSTAIN": + return "Uncertainty in extraction prevents confident decision" + return "Unknown" diff --git a/pievo/guard/policy.py b/pievo/guard/policy.py new file mode 100644 index 0000000..b620051 --- /dev/null +++ b/pievo/guard/policy.py @@ -0,0 +1,47 @@ +""" +Guard policy decision strategies (Spec Section 4.3). +""" + +from abc import ABC, abstractmethod +from pievo.guard.signature import AlignmentReport + + +class GuardPolicy(ABC): + """Abstract policy interface. Returns one of: PROCEED, RESTRICT, BLOCK, ABSTAIN.""" + + @abstractmethod + def decide(self, report: AlignmentReport) -> str: ... + + +class StrictPolicy(GuardPolicy): + """Any dimension < 1.0 => BLOCK.""" + + def decide(self, report: AlignmentReport) -> str: + ref, op, ev = report.vector + if any(v < 1.0 for v in (ref, op, ev)): + return "BLOCK" + return "PROCEED" + + +class PermissivePolicy(GuardPolicy): + """All aligned => PROCEED, op >= 0.5 => RESTRICT, else BLOCK.""" + + def decide(self, report: AlignmentReport) -> str: + ref, op, ev = report.vector + if ref == 1.0 and op == 1.0 and ev == 1.0: + return "PROCEED" + if op >= 0.5: + return "RESTRICT" + return "BLOCK" + + +class ConservativePolicy(GuardPolicy): + """Uncertain => ABSTAIN, aligned => PROCEED, else BLOCK.""" + + def decide(self, report: AlignmentReport) -> str: + if report.has_uncertain: + return "ABSTAIN" + ref, op, ev = report.vector + if ref == 1.0 and op == 1.0 and ev == 1.0: + return "PROCEED" + return "BLOCK" diff --git a/pievo/guard/signature.py b/pievo/guard/signature.py new file mode 100644 index 0000000..c4e99ff --- /dev/null +++ b/pievo/guard/signature.py @@ -0,0 +1,143 @@ +""" +Core data models for the Semantic Guard Protocol. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field, asdict +from enum import Enum +from typing import Any, Dict, List, Tuple + +# Sentinel: principles share no variable names (Spec Section 3.3) +NO_SHARED_VARIABLES = "NO_SHARED_VARIABLES" + + +class EvidentialStatus(Enum): + """Evidential status of a variable in a principle.""" + PRIMITIVE = "primitive" + DERIVED = "derived" + AUXILIARY = "auxiliary" + + +@dataclass(frozen=True) +class OperationalDefinition: + """Structured operational definition (Spec Section 3.1).""" + protocol: str + instrument: str + unit: str + environment: str + + def to_dict(self) -> Dict[str, str]: + return asdict(self) + + @classmethod + def from_dict(cls, d: Dict[str, str]) -> "OperationalDefinition": + return cls(**d) + + +def op_def_eq(o1, o2) -> bool: + """Field-level strict matching of operational definitions.""" + if o1 is None or o2 is None: + return o1 is o2 + return ( + o1.protocol == o2.protocol + and o1.instrument == o2.instrument + and o1.unit == o2.unit + and o1.environment == o2.environment + ) + + +@dataclass(frozen=True) +class VariableSignature: + """Structured variable signature per Spec Section 3.1.""" + name: str + referent: str + operational_definition: OperationalDefinition + evidential_status: EvidentialStatus + confidence: float + + def __post_init__(self) -> None: + if not 0.0 <= self.confidence <= 1.0: + raise ValueError(f"Confidence must be in [0,1], got {self.confidence}") + + def to_dict(self) -> Dict[str, Any]: + return { + "name": self.name, + "referent": self.referent, + "operational_definition": self.operational_definition.to_dict(), + "evidential_status": self.evidential_status.value, + "confidence": self.confidence, + } + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "VariableSignature": + return cls( + name=d["name"], + referent=d["referent"], + operational_definition=OperationalDefinition.from_dict(d["operational_definition"]), + evidential_status=EvidentialStatus(d["evidential_status"]), + confidence=d["confidence"], + ) + + +@dataclass +class Principle: + """Principle container per Spec Section 3.1.""" + id: str + text: str + phenomenon: str + signatures: List[VariableSignature] = field(default_factory=list) + + def to_dict(self) -> Dict[str, Any]: + return { + "id": self.id, + "text": self.text, + "phenomenon": self.phenomenon, + "signatures": [s.to_dict() for s in self.signatures], + } + + +@dataclass +class PerVariableAlignment: + """Per-variable alignment details (Spec Section 5.3).""" + name: str + referential: float + operational: float + evidential: float + uncertain: bool + confidence: float = 1.0 + details: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + return { + "name": self.name, + "referential": self.referential, + "operational": self.operational, + "evidential": self.evidential, + "uncertain": self.uncertain, + "confidence": self.confidence, + "details": self.details, + } + + +@dataclass +class AlignmentReport: + """Alignment report per Spec Section 5.3.""" + vector: Tuple[float, float, float] # (referential, operational, evidential) + per_variable: List[PerVariableAlignment] + shared_count: int + principle_pair: Tuple[str, str] = ("", "") + phenomenon: str = "" + policy_recommendation: str = "PROCEED" + has_uncertain: bool = False + + def to_dict(self) -> Dict[str, Any]: + return { + "principle_pair": list(self.principle_pair), + "phenomenon": self.phenomenon, + "vector": list(self.vector), + "per_variable": [v.to_dict() for v in self.per_variable], + "shared_count": self.shared_count, + "policy_recommendation": self.policy_recommendation, + "has_uncertain": self.has_uncertain, + } diff --git a/pievo/main.py b/pievo/main.py index 8111944..b5d5e29 100644 --- a/pievo/main.py +++ b/pievo/main.py @@ -148,6 +148,27 @@ def __init__( self._set_util_client(cache_dir=self.cache_dir) # If off PiEvo, this term will be None. + # Configure optional Semantic Guard + _guard = None + if getattr(args, "enable_semantic_guard", False): + from pievo.guard import SemanticGuard, StrictPolicy, PermissivePolicy, ConservativePolicy, LLMExtractor + + policy_map = { + "strict": StrictPolicy(), + "permissive": PermissivePolicy(), + "conservative": ConservativePolicy(), + } + guard_policy = policy_map.get(args.guard_policy, StrictPolicy()) + + # Create guard using the util_client for LLM extraction + _guard = SemanticGuard( + extractor=LLMExtractor(client=self.util_client), + policy=guard_policy, + ) + logging.info( + f"GUARD: Semantic Guard enabled with policy='{args.guard_policy}'" + ) + self.pievo = PiEvo( task=self.task_config.get("task"), output_file=os.path.join(save_dir, f"submissions.json"), @@ -161,6 +182,7 @@ def __init__( initial_absolute_sigma=args.initial_absolute_sigma, llm_judge=args.enable_llm_judge, off_pievo=self.off_pievo, + guard=_guard, ) # Agents will be created asynchronously using the factory method @@ -449,6 +471,21 @@ async def run_pievo(): parser.add_argument("--off_pievo", action="store_true") + # Semantic Guard (optional plugin) + parser.add_argument( + "--enable-semantic-guard", + action="store_true", + help="Enable variable-level semantic alignment checks for principles", + ) + parser.add_argument( + "--guard-policy", + type=str, + default="strict", + choices=["strict", "permissive", "conservative"], + help="Guard policy: strict (any misalignment => BLOCK), " + "permissive (allow partial alignment), conservative (abstain on uncertain)", + ) + args = parser.parse_args() if args.debug: diff --git a/tests/test_semantic_guard.py b/tests/test_semantic_guard.py new file mode 100644 index 0000000..58ce0d5 --- /dev/null +++ b/tests/test_semantic_guard.py @@ -0,0 +1,686 @@ +""" +Integration tests for the Semantic Guard plugin. + +These tests validate the core guard logic — signature extraction, alignment +vector computation, policy decisions, and orchestration — without requiring +a full PiEvo runtime (no flask, no autogen agents). + +Import strategy: + The pievo package __init__.py triggers a flask import chain that is not + available in this environment. We load guard modules directly via + importlib, bypassing the parent package. +""" + +from __future__ import annotations + +import asyncio +import importlib.util +import sys +from typing import Any, Dict, List +from unittest.mock import Mock + +# ── Load guard modules without triggering pievo/__init__.py ────────────── +_GUARD_BASE = "pievo/guard" +_MODULE_MAP = { + "signature": "pievo.guard.signature", + "alignment": "pievo.guard.alignment", + "policy": "pievo.guard.policy", + "extractor": "pievo.guard.extractor", + "guard": "pievo.guard.guard", +} + +for _key, _full_name in _MODULE_MAP.items(): + _spec = importlib.util.spec_from_file_location( + _full_name, f"{_GUARD_BASE}/{_key}.py" + ) + _mod = importlib.util.module_from_spec(_spec) + sys.modules[_full_name] = _mod + _spec.loader.exec_module(_mod) + +# Local aliases for readability +from pievo.guard.signature import ( # noqa: E402 + AlignmentReport, + EvidentialStatus, + OperationalDefinition, + PerVariableAlignment, + Principle, + VariableSignature, + NO_SHARED_VARIABLES, +) +from pievo.guard.alignment import SimilarityMethod, compute_alignment_vector # noqa: E402 +from pievo.guard.policy import ( # noqa: E402 + ConservativePolicy, + PermissivePolicy, + StrictPolicy, +) +from pievo.guard.guard import AlignmentDecision, SemanticGuard # noqa: E402 +from pievo.guard.extractor import ManualExtractor, SignatureExtractor # noqa: E402 + + +# ═══════════════════════════════════════════════════════════════════════════ +# Test fixtures: pre-crafted signatures for the canonical Phlogiston vs. +# Oxidation example. +# ═══════════════════════════════════════════════════════════════════════════ + +PHLOGISTON_SIGS = [ + VariableSignature( + name="mass_change", + referent="weight of the dephlogisticated metal after burning, measured in open air", + operational_definition=OperationalDefinition( + protocol="spring_scale_measurement", + instrument="balance", + unit="gram", + environment="open_air", + ), + evidential_status=EvidentialStatus.PRIMITIVE, + confidence=0.92, + ), + VariableSignature( + name="phlogiston", + referent="fire-like substance contained in combustible bodies, released upon burning", + operational_definition=OperationalDefinition( + protocol="inferred_from_mass_change", + instrument="none", + unit="none", + environment="none", + ), + evidential_status=EvidentialStatus.AUXILIARY, + confidence=0.85, + ), +] + +OXIDATION_SIGS = [ + VariableSignature( + name="mass_change", + referent="weight of the metal oxide product after burning, measured in sealed vessel", + operational_definition=OperationalDefinition( + protocol="spring_scale_measurement", + instrument="analytical_balance", + unit="gram", + environment="sealed_vessel", + ), + evidential_status=EvidentialStatus.PRIMITIVE, + confidence=0.95, + ), + VariableSignature( + name="oxygen", + referent="gaseous element from the air that combines with metal during combustion", + operational_definition=OperationalDefinition( + protocol="volume_measurement", + instrument="gas_collector", + unit="milliliter", + environment="sealed_vessel", + ), + evidential_status=EvidentialStatus.PRIMITIVE, + confidence=0.93, + ), +] + +# Two principles that use the SAME variable names with SAME referents +# (should PASS alignment). +ALIGNED_SIGS_1 = [ + VariableSignature( + name="temperature", + referent="average kinetic energy of gas particles in the reaction chamber", + operational_definition=OperationalDefinition( + protocol="thermocouple_reading", + instrument="thermocouple", + unit="kelvin", + environment="reaction_chamber", + ), + evidential_status=EvidentialStatus.PRIMITIVE, + confidence=0.98, + ), +] + +ALIGNED_SIGS_2 = [ + VariableSignature( + name="temperature", + referent="average kinetic energy of gas particles in the reaction chamber", + operational_definition=OperationalDefinition( + protocol="thermocouple_reading", + instrument="thermocouple", + unit="kelvin", + environment="reaction_chamber", + ), + evidential_status=EvidentialStatus.PRIMITIVE, + confidence=0.96, + ), +] + + +# ── Helper: use ManualExtractor (bundled with guard) for controlled tests ── + + + +# ═══════════════════════════════════════════════════════════════════════════ +# Tests +# ═══════════════════════════════════════════════════════════════════════════ + +class TestAlignmentVector: + """Section 3.3: 3D alignment vector computation.""" + + def test_phlogiston_vs_oxidation_misaligned(self): + """Shared name 'mass_change' but different referent, op, env.""" + report = compute_alignment_vector( + PHLOGISTON_SIGS, + OXIDATION_SIGS, + phenomenon="combustion", + similarity_method=SimilarityMethod.EXACT, + ) + # mass_change is shared, but referent is different → ref=0 + assert report.shared_count == 1 + ref, op, ev = report.vector + assert ref == 0.0, f"Expected referential=0 (different physical meanings), got {ref}" + assert op == 0.0, f"Expected operational=0 (different protocols), got {op}" + # evidential: both PRIMITIVE → should match + assert ev == 1.0, f"Expected evidential=1 (both primitive), got {ev}" + + # policy_recommendation from the report (before policy applied) + pva = report.per_variable[0] + assert pva.name == "mass_change" + + def test_identical_signatures_produce_full_alignment(self): + """Same signatures → vector=(1, 1, 1).""" + report = compute_alignment_vector( + ALIGNED_SIGS_1, + ALIGNED_SIGS_2, + phenomenon="thermodynamics", + similarity_method=SimilarityMethod.EXACT, + ) + assert report.vector == (1.0, 1.0, 1.0) + assert report.shared_count == 1 + assert not report.has_uncertain + + def test_no_shared_variables_sentinel(self): + """Principles with zero overlapping variable names.""" + sigs_a = [ + VariableSignature( + name="velocity", + referent="speed of object", + operational_definition=OperationalDefinition( + protocol="radar", instrument="radar_gun", unit="m/s", environment="open_field", + ), + evidential_status=EvidentialStatus.PRIMITIVE, + confidence=0.9, + ), + ] + sigs_b = [ + VariableSignature( + name="pressure", + referent="force per unit area", + operational_definition=OperationalDefinition( + protocol="barometer", instrument="barometer", unit="pascal", environment="chamber", + ), + evidential_status=EvidentialStatus.PRIMITIVE, + confidence=0.9, + ), + ] + report = compute_alignment_vector(sigs_a, sigs_b, phenomenon="physics") + assert report.policy_recommendation == NO_SHARED_VARIABLES + assert report.shared_count == 0 + assert len(report.per_variable) == 0 + + +class TestPolicies: + """Section 4.3: Policy decision strategies.""" + + def _make_report(self, vector, has_uncertain=False): + return AlignmentReport( + vector=vector, + per_variable=[], + shared_count=1, + phenomenon="test", + has_uncertain=has_uncertain, + ) + + def test_strict_blocks_any_dimension_below_1(self): + strict = StrictPolicy() + assert strict.decide(self._make_report((1.0, 1.0, 1.0))) == "PROCEED" + assert strict.decide(self._make_report((1.0, 0.99, 1.0))) == "BLOCK" + assert strict.decide(self._make_report((0.0, 0.0, 1.0))) == "BLOCK" + + def test_permissive_allows_partial_operational(self): + permissive = PermissivePolicy() + assert permissive.decide(self._make_report((1.0, 1.0, 1.0))) == "PROCEED" + # op >= 0.5 but not full → RESTRICT + assert permissive.decide(self._make_report((0.0, 0.5, 1.0))) == "RESTRICT" + # op < 0.5 → BLOCK + assert permissive.decide(self._make_report((0.0, 0.0, 1.0))) == "BLOCK" + + def test_conservative_abstains_on_uncertain(self): + conservative = ConservativePolicy() + assert conservative.decide(self._make_report((1.0, 1.0, 1.0), has_uncertain=False)) == "PROCEED" + assert conservative.decide(self._make_report((1.0, 1.0, 1.0), has_uncertain=True)) == "ABSTAIN" + assert conservative.decide(self._make_report((0.0, 0.0, 0.0), has_uncertain=False)) == "BLOCK" + + +class TestSemanticGuardIntegration: + """ + End-to-end guard orchestration using a mock extractor. + + This is the closest we can get to testing the PiEvo integration without + a full PiEvo runtime. The same SemanticGuard.check_against_all() is + called in PiEvo.update_pievo_state(). + """ + + PHLOGISTON_TEXT = "Metals contain phlogiston which escapes upon burning." + OXIDATION_TEXT = "Metals combine with oxygen to form oxides; mass increases." + TEMPERATURE_TEXT_1 = "Temperature measures average kinetic energy." + TEMPERATURE_TEXT_2 = "Temperature measures average kinetic energy — identical." + + def _make_guard(self, policy_cls=StrictPolicy): + extractor = ManualExtractor({ + self.PHLOGISTON_TEXT: PHLOGISTON_SIGS, + self.OXIDATION_TEXT: OXIDATION_SIGS, + self.TEMPERATURE_TEXT_1: ALIGNED_SIGS_1, + self.TEMPERATURE_TEXT_2: ALIGNED_SIGS_2, + }) + return SemanticGuard(extractor=extractor, policy=policy_cls()) + + def test_check_pair_detects_misalignment(self): + """Phlogiston vs. Oxidation → BLOCK due to referential conflict.""" + guard = self._make_guard() + p1 = Principle(id="phlog", text=self.PHLOGISTON_TEXT, phenomenon="combustion") + p2 = Principle(id="oxi", text=self.OXIDATION_TEXT, phenomenon="combustion") + + decision = asyncio.run(guard.check_pair(p1, p2, "combustion")) + + assert decision.action == "BLOCK", f"Expected BLOCK, got {decision.action}" + assert decision.p1_id == "phlog" + assert decision.p2_id == "oxi" + assert "mass_change" in decision.misaligned_variables + assert decision.report.vector[0] == 0.0 # referential misalignment + # Log should contain the decision + assert len(guard.get_log()) == 1 + + def test_check_pair_allows_aligned_principles(self): + """Two temperature descriptions with identical signatures → PROCEED.""" + guard = self._make_guard() + p1 = Principle(id="temp1", text=self.TEMPERATURE_TEXT_1, phenomenon="heat") + p2 = Principle(id="temp2", text=self.TEMPERATURE_TEXT_2, phenomenon="heat") + + decision = asyncio.run(guard.check_pair(p1, p2, "heat")) + + assert decision.action == "PROCEED", f"Expected PROCEED, got {decision.action}: {decision.reason}" + assert len(decision.misaligned_variables) == 0 + + def test_check_against_all_skips_self(self): + """A principle should not be checked against itself.""" + guard = self._make_guard() + new_p = Principle(id="phlog", text=self.PHLOGISTON_TEXT, phenomenon="combustion") + existing = {"phlog": self.PHLOGISTON_TEXT} + + decisions = asyncio.run(guard.check_against_all(new_p, existing, "combustion")) + # The only entry is the new principle itself → skipped → empty list + assert len(decisions) == 0 + + def test_check_against_all_detects_conflict_in_many(self): + """New oxidation principle conflicts with existing phlogiston principle.""" + guard = self._make_guard() + new_p = Principle(id="oxi", text=self.OXIDATION_TEXT, phenomenon="combustion") + existing = { + "phlog": self.PHLOGISTON_TEXT, + "temp1": self.TEMPERATURE_TEXT_1, + } + + decisions = asyncio.run(guard.check_against_all(new_p, existing, "combustion")) + # One decision per existing principle (excluding self) + assert len(decisions) == 2 + + # phlog vs oxi → BLOCK + block_decision = [d for d in decisions if d.action == "BLOCK"] + assert len(block_decision) == 1 + assert block_decision[0].p1_id == "oxi" + assert block_decision[0].p2_id == "phlog" + + # temp1 vs oxi → no shared variables → NO_SHARED_VARIABLES + nsv_decision = [d for d in decisions if d.action == NO_SHARED_VARIABLES] + assert len(nsv_decision) == 1 + + def test_any_blocked_flag(self): + guard = self._make_guard() + new_p = Principle(id="oxi", text=self.OXIDATION_TEXT, phenomenon="combustion") + existing = {"phlog": self.PHLOGISTON_TEXT} + decisions = asyncio.run(guard.check_against_all(new_p, existing, "combustion")) + assert guard.any_blocked(decisions) is True + + def test_collect_misaligned_variables(self): + guard = self._make_guard() + new_p = Principle(id="oxi", text=self.OXIDATION_TEXT, phenomenon="combustion") + existing = {"phlog": self.PHLOGISTON_TEXT} + decisions = asyncio.run(guard.check_against_all(new_p, existing, "combustion")) + conflicts = guard.collect_misaligned_variables(decisions) + assert "phlog" in conflicts + assert "mass_change" in conflicts["phlog"] + + def test_signature_caching(self): + """Extract once, then cached — second call uses cache.""" + guard = self._make_guard() + p1 = Principle(id="phlog", text=self.PHLOGISTON_TEXT, phenomenon="combustion") + + # First extraction + sigs1 = asyncio.run(guard.extract_signatures("phlog", self.PHLOGISTON_TEXT, "combustion")) + assert len(sigs1) == 2 + + # Second extraction — should hit cache + sigs2 = asyncio.run(guard.extract_signatures("phlog", self.PHLOGISTON_TEXT, "combustion")) + assert sigs1 is sigs2 # Same list object (cached) + + def test_clear_log(self): + guard = self._make_guard() + p1 = Principle(id="phlog", text=self.PHLOGISTON_TEXT, phenomenon="combustion") + p2 = Principle(id="oxi", text=self.OXIDATION_TEXT, phenomenon="combustion") + asyncio.run(guard.check_pair(p1, p2, "combustion")) + assert len(guard.get_log()) == 1 + guard.clear_log() + assert len(guard.get_log()) == 0 + + def test_permissive_policy_restricts_partial(self): + """PermissivePolicy: partial operational conflict → RESTRICT.""" + guard = self._make_guard(policy_cls=PermissivePolicy) + p1 = Principle(id="phlog", text=self.PHLOGISTON_TEXT, phenomenon="combustion") + p2 = Principle(id="oxi", text=self.OXIDATION_TEXT, phenomenon="combustion") + + decision = asyncio.run(guard.check_pair(p1, p2, "combustion")) + # Referential=0, operational=0, evidential=1 → op=0 < 0.5 → BLOCK + # With PermissivePolicy, op=0 < 0.5 means BLOCK + assert decision.action == "BLOCK" + + def test_conservative_policy_on_uncertain_extraction(self): + """ConservativePolicy with uncertain signatures → ABSTAIN.""" + # Low-confidence signatures + uncertain_sigs = [ + VariableSignature( + name="mass_change", + referent="weight change of metal", + operational_definition=OperationalDefinition( + protocol="unknown", instrument="unknown", unit="unknown", environment="unknown", + ), + evidential_status=EvidentialStatus.PRIMITIVE, + confidence=0.6, # < 0.8 → uncertain + ), + ] + guard = SemanticGuard( + extractor=ManualExtractor({ + "u1": uncertain_sigs, + "u2": uncertain_sigs, + }), + policy=ConservativePolicy(), + ) + p1 = Principle(id="u1", text="u1", phenomenon="test") + p2 = Principle(id="u2", text="u2", phenomenon="test") + decision = asyncio.run(guard.check_pair(p1, p2, "test")) + # Both have low confidence → has_uncertain=True → Conservative → ABSTAIN + assert decision.action == "ABSTAIN", f"Expected ABSTAIN, got {decision.action}" + + +class TestAlignmentDecision: + """AlignmentDecision dataclass serialization.""" + + def test_to_dict_roundtrip_keys(self): + report = AlignmentReport( + vector=(1.0, 0.5, 1.0), + per_variable=[], + shared_count=2, + principle_pair=("a", "b"), + phenomenon="test", + ) + decision = AlignmentDecision( + report=report, + action="BLOCK", + p1_id="a", + p2_id="b", + misaligned_variables=["x"], + reason="test", + ) + d = decision.to_dict() + assert d["action"] == "BLOCK" + assert d["principle_pair"] == ["a", "b"] + assert "x" in d["misaligned_variables"] + assert d["report"] == report.to_dict() + + +# ═══════════════════════════════════════════════════════════════════════════ +# PiEvo Integration Simulation +# +# These tests replicate the EXACT guard hook logic from +# PiEvo.update_pievo_state() (evolveflow.py lines 857-898) without +# requiring a full PiEvo runtime. They validate that the integration +# between SemanticGuard and the Bayesian update pipeline is correct. +# ═══════════════════════════════════════════════════════════════════════════ + + +async def _simulate_guard_hook( + guard: SemanticGuard, + principles: Dict[str, str], # self.principles: id -> text + new_principle_ids: List[str], # IDs extracted by _extract_principles() + phenomenon: str, +) -> Dict[str, Any]: + """ + Replicate the guard hook logic from evolveflow.py lines 858-898. + + Returns a dict describing what happened so tests can inspect it. + """ + if guard is None or not new_principle_ids: + return {"validated": new_principle_ids, "blocked": [], "principles": principles} + + from pievo.guard.signature import Principle as GuardPrinciple + + validated_ids: List[str] = [] + blocked_ids: List[str] = [] + + for pid in new_principle_ids: + ptext = principles.get(pid, "") + new_p = GuardPrinciple(id=pid, text=ptext, phenomenon=phenomenon) + decisions = await guard.check_against_all(new_p, principles, phenomenon) + if guard.any_blocked(decisions): + blocked_ids.append(pid) + else: + validated_ids.append(pid) + + if blocked_ids: + for pid in blocked_ids: + principles.pop(pid, None) + new_principle_ids = validated_ids + + return { + "validated": validated_ids, + "blocked": blocked_ids, + "principles": principles, # mutated in place + } + + +class TestPiEvoIntegrationSimulation: + """Validate the guard hook logic as it runs in update_pievo_state().""" + + PHLOGISTON_TEXT = ( + "Metals contain phlogiston which escapes upon burning. " + "Mass remains constant because phlogiston is replaced by air." + ) + OXIDATION_TEXT = ( + "Metals combine with oxygen to form oxides; mass increases " + "because oxygen from the air bonds with the metal." + ) + KINETIC_TEXT = ( + "Heat is kinetic energy of molecular motion. Temperature " + "is the average kinetic energy of particles." + ) + + def _make_guard(self): + return SemanticGuard( + extractor=ManualExtractor({ + self.PHLOGISTON_TEXT: [ + VariableSignature( + name="mass_change", + referent="weight of dephlogisticated metal in open air", + operational_definition=OperationalDefinition( + protocol="scale", instrument="balance", + unit="gram", environment="open_air", + ), + evidential_status=EvidentialStatus.PRIMITIVE, + confidence=0.92, + ), + ], + self.OXIDATION_TEXT: [ + VariableSignature( + name="mass_change", + referent="weight of metal oxide in sealed vessel", + operational_definition=OperationalDefinition( + protocol="scale", instrument="analytical_balance", + unit="gram", environment="sealed_vessel", + ), + evidential_status=EvidentialStatus.PRIMITIVE, + confidence=0.95, + ), + ], + self.KINETIC_TEXT: [ + VariableSignature( + name="temperature", + referent="average kinetic energy of particles", + operational_definition=OperationalDefinition( + protocol="thermometer", instrument="thermocouple", + unit="kelvin", environment="chamber", + ), + evidential_status=EvidentialStatus.DERIVED, + confidence=0.95, + ), + ], + }), + policy=StrictPolicy(), + ) + + def test_blocked_principle_removed_from_dict(self): + """ + When a new principle conflicts with an existing one, it should be: + 1. Added to blocked_ids + 2. Removed from self.principles + 3. Excluded from validated_ids + """ + guard = self._make_guard() + + # Simulate PiEvo state: one existing principle (phlogiston), + # one newly extracted (oxidation) -- they share 'mass_change' with + # different referents. + principles = { + "phlog": self.PHLOGISTON_TEXT, + "oxi": self.OXIDATION_TEXT, # just added by _extract_principles() + } + new_ids = ["oxi"] + + result = asyncio.run( + _simulate_guard_hook(guard, principles, new_ids, "combustion") + ) + + # Oxidation conflicts with Phlogiston → BLOCKED + assert result["blocked"] == ["oxi"] + assert result["validated"] == [] + + # Blocked principle removed from principles dict + assert "oxi" not in result["principles"] + assert "phlog" in result["principles"] + + def test_non_conflicting_principle_proceeds(self): + """ + A new principle that shares no variable names with existing ones + should pass through without blocking. + """ + guard = self._make_guard() + + # Existing: phlogiston. New: kinetic (different variables entirely). + principles = { + "phlog": self.PHLOGISTON_TEXT, + "kinetic": self.KINETIC_TEXT, + } + new_ids = ["kinetic"] + + result = asyncio.run( + _simulate_guard_hook(guard, principles, new_ids, "combustion") + ) + + assert result["blocked"] == [] + assert result["validated"] == ["kinetic"] + assert "kinetic" in result["principles"] + + def test_mixed_batch_one_blocked_one_proceeds(self): + """ + When multiple principles are extracted simultaneously, blocked + ones are filtered out while valid ones survive. + """ + guard = self._make_guard() + + principles = { + "phlog": self.PHLOGISTON_TEXT, + "oxi": self.OXIDATION_TEXT, # conflicts with phlog + "kinetic": self.KINETIC_TEXT, # no shared vars with either + } + new_ids = ["oxi", "kinetic"] + + result = asyncio.run( + _simulate_guard_hook(guard, principles, new_ids, "combustion") + ) + + assert set(result["blocked"]) == {"oxi"} + assert set(result["validated"]) == {"kinetic"} + assert "oxi" not in result["principles"] + assert "kinetic" in result["principles"] + assert "phlog" in result["principles"] + + def test_guard_none_passes_all(self): + """When guard is None, all principles pass through unchanged.""" + principles = { + "phlog": self.PHLOGISTON_TEXT, + "oxi": self.OXIDATION_TEXT, + } + new_ids = ["oxi"] + + result = asyncio.run( + _simulate_guard_hook(None, principles, new_ids, "combustion") + ) + + assert result["blocked"] == [] + assert result["validated"] == ["oxi"] + assert "oxi" in result["principles"] + + def test_empty_new_ids_short_circuits(self): + """When no new principles extracted, guard is bypassed entirely.""" + guard = self._make_guard() + principles = {"phlog": self.PHLOGISTON_TEXT} + + result = asyncio.run( + _simulate_guard_hook(guard, principles, [], "combustion") + ) + + assert result["blocked"] == [] + assert result["validated"] == [] + assert principles == {"phlog": self.PHLOGISTON_TEXT} + + def test_self_not_checked(self): + """ + A new principle should never be checked against itself. + (check_against_all skips pid == new_principle.id) + """ + guard = self._make_guard() + + # Only existing principle is the same as the new one → no checks + principles = {"oxi": self.OXIDATION_TEXT} + new_ids = ["oxi"] + + result = asyncio.run( + _simulate_guard_hook(guard, principles, new_ids, "combustion") + ) + + # No existing principle to conflict with → passes + assert result["blocked"] == [] + assert result["validated"] == ["oxi"] + assert "oxi" in result["principles"] + + +# ═══════════════════════════════════════════════════════════════════════════ +# Run: py -m pytest tests/test_semantic_guard.py -v +# ═══════════════════════════════════════════════════════════════════════════ + +if __name__ == "__main__": + import pytest + + sys.exit(pytest.main([__file__, "-v", "--tb=short"]))