diff --git a/.gitignore b/.gitignore index 5ed9514..999a2e1 100644 --- a/.gitignore +++ b/.gitignore @@ -1,32 +1,30 @@ --e -# Environment files -*.env -*.env.* --e -# Environment files -*.env -*.env.* --e -# Environment files -*.env -*.env.* --e -# Environment files -*.env -*.env.* --e -# Environment files -*.env -*.env.* --e -# Environment files -*.env -*.env.* --e -# Environment files -*.env -*.env.* --e -# Environment files -*.env -*.env.* +# Python +__pycache__/ +*.py[cod] +*.pyo +*.pyd +.Python + +# Virtual environments +.venv/ +venv/ +env/ + +# Environment files +*.env +*.env.* + +# Distribution / packaging +*.egg-info/ +dist/ +build/ + +# Checkpoints +.checkpoints/ + +# pytest +.pytest_cache/ + +# IDE +.vscode/ +.idea/ diff --git a/backend/edcm_engine.py b/backend/edcm_engine.py index 6a21e77..92a5e53 100644 --- a/backend/edcm_engine.py +++ b/backend/edcm_engine.py @@ -1,246 +1,249 @@ """ EDCM (Entropy Dissonance Constraint Management) Analyzer -Generates artifacts for monetization and system diagnostics + +Six metric families (from a0 canonical spec): + CM — Constraint Mismatch alert: HIGH >= 0.80 + DA — Dissonance Accumulation alert: HIGH >= 0.80 + DRIFT — Drift alert: HIGH >= 0.80 + DVG — Divergence alert: HIGH >= 0.80 + INT — Intensity alert: LOW <= 0.20 + TBF — Turn-Balance Fairness alert: LOW <= 0.20 + +EDCM Behavioral Directives (fire when metric crosses threshold): + CONSTRAINT_REFOCUS — CM >= 0.80 + DISSONANCE_HALT — DA >= 0.80 + DRIFT_ANCHOR — DRIFT>= 0.80 + DIVERGENCE_COMMIT — DVG >= 0.80 + INTENSITY_CALM — INT <= 0.20 + BALANCE_CONCISE — TBF <= 0.20 """ +import math import logging from typing import Dict, List, Any from datetime import datetime -import numpy as np + +from core.edcm import ( + METRIC_NAMES, + ALERT_HIGH, + ALERT_LOW, + check_alerts, +) logger = logging.getLogger("edcm_analyzer") + class EDCMAnalyzer: """ - Analyzes PCNA system using EDCM principles: - - Entropy measurement - - Dissonance detection - - Constraint strain analysis - - Generates monetizable artifacts (reports, visualizations, insights) + Analyzes PCNA system state using the six-family EDCM metric framework. + Generates diagnostic artifacts and fires behavioral directives when + metrics cross thresholds. """ + def __init__(self): - self.analysis_history = [] - + self.analysis_history: List[Dict] = [] + async def analyze(self, seed_states: List[Dict]) -> Dict[str, Any]: """ - Perform EDCM analysis on system state - + Perform six-family EDCM analysis on system state. + Args: - seed_states: List of seed state dictionaries - + seed_states: list of seed state dicts (must include 'health_score' and 'mass'). + Returns: - EDCM analysis report (monetizable artifact) + EDCM analysis report with metrics, alerts, directives, insights, recommendations. """ - analysis = { + analysis: Dict[str, Any] = { "timestamp": datetime.utcnow().isoformat(), "artifact_type": "edcm_report", - "version": "1.0", + "version": "2.0", "metrics": {}, + "alerts": {}, + "directives": [], "insights": [], "recommendations": [], - "monetization_value": "medium" # low, medium, high + "monetization_value": "medium", } - - # 1. Calculate Entropy - entropy_metrics = self._calculate_entropy(seed_states) - analysis["metrics"]["entropy"] = entropy_metrics - - # 2. Detect Dissonance - dissonance_metrics = self._detect_dissonance(seed_states) - analysis["metrics"]["dissonance"] = dissonance_metrics - - # 3. Analyze Constraint Strain - constraint_metrics = self._analyze_constraints(seed_states) - analysis["metrics"]["constraints"] = constraint_metrics - - # 4. Generate Insights - insights = self._generate_insights(entropy_metrics, dissonance_metrics, constraint_metrics) + + metrics = self._compute_from_seeds(seed_states) + analysis["metrics"] = metrics + + alerts = check_alerts(metrics) + analysis["alerts"] = alerts + + directives = self._fire_directives(metrics) + analysis["directives"] = directives + + insights = self._generate_insights(metrics, alerts, directives) analysis["insights"] = insights - - # 5. Generate Recommendations - recommendations = self._generate_recommendations(analysis["metrics"]) + + recommendations = self._generate_recommendations(metrics, alerts) analysis["recommendations"] = recommendations - - # 6. Assess artifact value for monetization + analysis["monetization_value"] = self._assess_artifact_value(analysis) - - # Store in history + self.analysis_history.append(analysis) - - logger.info(f"EDCM analysis complete: {len(insights)} insights, {len(recommendations)} recommendations") - + + logger.info( + f"EDCM analysis: metrics={metrics} alerts={alerts} " + f"directives={directives} insights={len(insights)}" + ) + return analysis - - def _calculate_entropy(self, seed_states: List[Dict]) -> Dict[str, Any]: - """Calculate system entropy""" - if not seed_states: - return {"total_entropy": 0.0, "average_entropy": 0.0} - - # Calculate spectral entropy from seed phases - phases = [] - for state in seed_states: - if state.get("spectral"): - phases.append(state["spectral"].get("phase", 0.0)) - - if not phases: - return {"total_entropy": 0.0, "average_entropy": 0.0} - - # Shannon entropy approximation - phase_array = np.array(phases) - # Normalize phases to [0, 1] - normalized_phases = (phase_array + np.pi) / (2 * np.pi) - - # Calculate entropy - hist, _ = np.histogram(normalized_phases, bins=10) - hist = hist / hist.sum() # Normalize to probabilities - entropy = -np.sum(hist * np.log2(hist + 1e-10)) # Shannon entropy - - return { - "total_entropy": float(entropy), - "average_entropy": float(entropy / len(phases)), - "phase_distribution": hist.tolist(), - "interpretation": "high" if entropy > 2.5 else "medium" if entropy > 1.5 else "low" - } - - def _detect_dissonance(self, seed_states: List[Dict]) -> Dict[str, Any]: - """Detect dissonance in system""" + + def _compute_from_seeds(self, seed_states: List[Dict]) -> Dict[str, float]: + """Derive six EDCM metrics from seed state list.""" if not seed_states: - return {"dissonance_score": 0.0, "dissonant_seeds": []} - - # Dissonance = deviation from expected patterns - health_scores = [s["health_score"] for s in seed_states] - - # Calculate variance as dissonance measure - variance = np.var(health_scores) - dissonance_score = float(variance * 10) # Scale for readability - - # Identify dissonant seeds (outliers) - mean_health = np.mean(health_scores) - std_health = np.std(health_scores) - - dissonant_seeds = [] - for state in seed_states: - if abs(state["health_score"] - mean_health) > 2 * std_health: - dissonant_seeds.append({ - "seed_id": state["seed_id"], - "health_score": state["health_score"], - "deviation": abs(state["health_score"] - mean_health) - }) - - return { - "dissonance_score": dissonance_score, - "dissonant_seeds": dissonant_seeds, - "health_variance": float(variance), - "interpretation": "high" if dissonance_score > 1.0 else "medium" if dissonance_score > 0.5 else "low" - } - - def _analyze_constraints(self, seed_states: List[Dict]) -> Dict[str, Any]: - """Analyze constraint strain""" - # Calculate mass conservation constraint - total_mass = sum(s["mass"] for s in seed_states) - expected_mass = sum(1.0 for s in seed_states if s.get("role") == "compute") - - conservation_strain = abs(total_mass - expected_mass) / (expected_mass + 1e-10) - + return {m: 0.0 for m in METRIC_NAMES} + + health_scores = [s.get("health_score", 0.0) for s in seed_states] + masses = [s.get("mass", 0.0) for s in seed_states] + + n = len(health_scores) + mean_h = sum(health_scores) / max(n, 1) + variance_h = sum((h - mean_h) ** 2 for h in health_scores) / max(n, 1) + + # CM: mass conservation deviation — how far total mass deviates from expected + compute_seeds = [s for s in seed_states if s.get("role") == "compute"] + expected_mass = len(compute_seeds) if compute_seeds else n + total_mass = sum(masses) + cm = min(1.0, abs(total_mass - expected_mass) / max(expected_mass, 1)) + + # DA: dissonance accumulation — normalized health variance + da = min(1.0, math.sqrt(variance_h)) + + # DRIFT: deviation from healthy baseline (1.0) + drift = max(0.0, min(1.0, 1.0 - mean_h)) + + # DVG: fraction of outlier seeds (> 2 std from mean) + std_h = math.sqrt(variance_h) + outliers = sum(1 for h in health_scores if abs(h - mean_h) > 2 * std_h) + dvg = min(1.0, outliers / max(n, 1)) + + # INT: system intensity — mean health as proxy for active processing + int_val = max(0.0, min(1.0, mean_h)) + + # TBF: turn-balance fairness — inverse of health std (balanced = fair) + tbf = max(0.0, min(1.0, 1.0 - std_h)) + return { - "total_mass": float(total_mass), - "expected_mass": float(expected_mass), - "conservation_strain": float(conservation_strain), - "strain_level": "critical" if conservation_strain > 0.1 else "warning" if conservation_strain > 0.01 else "normal" + "cm": round(cm, 4), + "da": round(da, 4), + "drift": round(drift, 4), + "dvg": round(dvg, 4), + "int_val": round(int_val, 4), + "tbf": round(tbf, 4), } - - def _generate_insights(self, - entropy_metrics: Dict, - dissonance_metrics: Dict, - constraint_metrics: Dict) -> List[str]: - """Generate human-readable insights""" + + def _fire_directives(self, metrics: Dict[str, float]) -> List[str]: + """Map metric threshold crossings to EDCM behavioral directives.""" + fired = [] + if metrics.get("cm", 0.0) >= ALERT_HIGH: + fired.append("CONSTRAINT_REFOCUS") + if metrics.get("da", 0.0) >= ALERT_HIGH: + fired.append("DISSONANCE_HALT") + if metrics.get("drift", 0.0) >= ALERT_HIGH: + fired.append("DRIFT_ANCHOR") + if metrics.get("dvg", 0.0) >= ALERT_HIGH: + fired.append("DIVERGENCE_COMMIT") + if metrics.get("int_val", 0.0) <= ALERT_LOW: + fired.append("INTENSITY_CALM") + if metrics.get("tbf", 0.0) <= ALERT_LOW: + fired.append("BALANCE_CONCISE") + return fired + + def _generate_insights( + self, + metrics: Dict[str, float], + alerts: Dict[str, List[str]], + directives: List[str], + ) -> List[str]: insights = [] - - # Entropy insights - if entropy_metrics["interpretation"] == "high": - insights.append("System exhibits high entropy - indicates diverse, distributed processing") - elif entropy_metrics["interpretation"] == "low": - insights.append("Low entropy detected - system may be overly synchronized") - - # Dissonance insights - if dissonance_metrics["dissonant_seeds"]: - insights.append(f"{len(dissonance_metrics['dissonant_seeds'])} seeds showing dissonance - potential optimization targets") - - # Constraint insights - if constraint_metrics["strain_level"] != "normal": - insights.append(f"Mass conservation strain at {constraint_metrics['strain_level']} level") - - # Overall system health insight + high = alerts.get("HIGH", []) + low = alerts.get("LOW", []) + + if "cm" in high: + insights.append("HIGH CM: mass conservation violated — system energy imbalance detected") + if "da" in high: + insights.append("HIGH DA: dissonance accumulation elevated — seed health diverging") + if "drift" in high: + insights.append("HIGH DRIFT: system drifting from healthy baseline") + if "dvg" in high: + insights.append("HIGH DVG: divergence elevated — significant outlier seeds present") + if "int_val" in low: + insights.append("LOW INT: system intensity suppressed — reduced active processing") + if "tbf" in low: + insights.append("LOW TBF: turn-balance fairness degraded — uneven load distribution") + + for directive in directives: + insights.append(f"Directive fired: {directive}") + if not insights: insights.append("System operating within normal EDCM parameters") - + return insights - - def _generate_recommendations(self, metrics: Dict) -> List[Dict[str, str]]: - """Generate actionable recommendations""" + + def _generate_recommendations( + self, metrics: Dict[str, float], alerts: Dict[str, List[str]] + ) -> List[Dict[str, str]]: recommendations = [] - - # Entropy-based recommendations - entropy = metrics["entropy"]["interpretation"] - if entropy == "high": + high = alerts.get("HIGH", []) + low = alerts.get("LOW", []) + + if "cm" in high: recommendations.append({ - "priority": "low", - "action": "Monitor entropy trends", - "reason": "High entropy may indicate inefficiency if sustained" + "priority": "high", + "action": "Emergency system rebalance", + "reason": "Mass conservation critically violated (CM >= 0.80)", }) - - # Dissonance-based recommendations - if metrics["dissonance"]["dissonant_seeds"]: + if "da" in high: recommendations.append({ - "priority": "medium", + "priority": "high", "action": "Investigate dissonant seeds", - "reason": f"{len(metrics['dissonance']['dissonant_seeds'])} seeds showing outlier behavior" + "reason": "Dissonance accumulation above threshold (DA >= 0.80)", }) - - # Constraint-based recommendations - strain_level = metrics["constraints"]["strain_level"] - if strain_level == "critical": + if "drift" in high: recommendations.append({ - "priority": "high", - "action": "Emergency system rebalance", - "reason": "Mass conservation critically violated" + "priority": "medium", + "action": "Re-anchor system to healthy baseline", + "reason": "Drift exceeds safe operating range (DRIFT >= 0.80)", + }) + if "dvg" in high: + recommendations.append({ + "priority": "medium", + "action": "Quarantine or rebalance outlier seeds", + "reason": "Divergence elevated — outlier seeds detected (DVG >= 0.80)", }) - elif strain_level == "warning": + if "int_val" in low: recommendations.append({ "priority": "medium", - "action": "Schedule system tune-up", - "reason": "Mass conservation showing strain" + "action": "Boost system intensity", + "reason": "Intensity suppressed below safe floor (INT <= 0.20)", + }) + if "tbf" in low: + recommendations.append({ + "priority": "low", + "action": "Redistribute load across seeds", + "reason": "Turn-balance fairness degraded (TBF <= 0.20)", }) - + return recommendations - + def _assess_artifact_value(self, analysis: Dict) -> str: - """Assess monetization value of artifact""" - # High value if: - # - Critical issues detected - # - Multiple actionable recommendations - # - Novel insights - - recommendations = analysis["recommendations"] - high_priority = sum(1 for r in recommendations if r["priority"] == "high") - insights_count = len(analysis["insights"]) - - if high_priority > 0 or insights_count > 3: + high_recs = sum(1 for r in analysis["recommendations"] if r["priority"] == "high") + if high_recs > 0 or len(analysis["directives"]) > 0: return "high" - elif len(recommendations) > 1: + elif len(analysis["recommendations"]) > 1: return "medium" - else: - return "low" - + return "low" + def get_artifact_summary(self, limit: int = 5) -> Dict[str, Any]: - """Get summary of recent artifacts for monetization dashboard""" recent = self.analysis_history[-limit:] - return { "total_artifacts": len(self.analysis_history), "recent_artifacts": recent, "value_distribution": { "high": sum(1 for a in recent if a["monetization_value"] == "high"), "medium": sum(1 for a in recent if a["monetization_value"] == "medium"), - "low": sum(1 for a in recent if a["monetization_value"] == "low") - } + "low": sum(1 for a in recent if a["monetization_value"] == "low"), + }, } diff --git a/backend/researcher_outreach.py b/backend/researcher_outreach.py index 0e92081..9202834 100644 --- a/backend/researcher_outreach.py +++ b/backend/researcher_outreach.py @@ -108,7 +108,7 @@ async def generate_outreach_message(self, researcher: Dict) -> str: Returns: Personalized message text """ - prompt = f\"\"\"Generate a professional, trauma-informed outreach email to the following researcher. + prompt = f"""Generate a professional, trauma-informed outreach email to the following researcher. Researcher Profile: - Name: {researcher.get('name', 'Unknown')} @@ -117,8 +117,8 @@ async def generate_outreach_message(self, researcher: Dict) -> str: - Interests: {', '.join(researcher.get('interests', []))} Context about PCNA: -PCNA (Prime Circular Neural Architecture) is a deterministic, prime-indexed, circular graph architecture -for modular compute and diagnostics. It uses 7:3 heptagram routing for 49 compute seeds, 7 meta routers, +PCNA (Prime Circular Neural Architecture) is a deterministic, prime-indexed, circular graph architecture +for modular compute and diagnostics. It uses 7:3 heptagram routing for 49 compute seeds, 7 meta routers, and 4 sentinels for independent monitoring. Key features: @@ -137,7 +137,7 @@ async def generate_outreach_message(self, researcher: Dict) -> str: 6. Maintain a warm, professional tone 7. Be trauma-informed (transparent, non-manipulative) -Generate the email:\"\"\" +Generate the email:""" message = await self.llm.chat(prompt, provider="anthropic") # Claude is best for empathetic content diff --git a/backend/server.py b/backend/server.py index ab15bbc..04f14f8 100644 --- a/backend/server.py +++ b/backend/server.py @@ -168,8 +168,6 @@ async def lifespan(app: FastAPI): async def initialize_seeds(): """Initialize key seeds""" - global active_seeds - # Create global router active_seeds[0] = PCNASeed(0, SeedRole.GLOBAL) diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..306f2d4 --- /dev/null +++ b/conftest.py @@ -0,0 +1,4 @@ +import sys +import os + +sys.path.insert(0, os.path.dirname(__file__)) diff --git a/core/edcm.py b/core/edcm.py new file mode 100644 index 0000000..84d36ef --- /dev/null +++ b/core/edcm.py @@ -0,0 +1,94 @@ +""" +EDCM metrics — six-family coherence measurement. + +Metric families: + cm — Constraint Mismatch + da — Dissonance Accumulation + drift — Drift + dvg — Divergence + int_val — Intensity + tbf — Turn-Balance Fairness + +All metrics produce values in [0, 1]. +Alert thresholds: HIGH >= 0.80, LOW <= 0.20. +""" + +import math +from typing import Any + +METRIC_NAMES = ["cm", "da", "drift", "dvg", "int_val", "tbf"] + +ALERT_HIGH = 0.80 +ALERT_LOW = 0.20 + +DIRECTIVES = { + "CONSTRAINT_REFOCUS": {"metric": "cm", "condition": "above", "threshold": ALERT_HIGH}, + "DISSONANCE_HALT": {"metric": "da", "condition": "above", "threshold": ALERT_HIGH}, + "DRIFT_ANCHOR": {"metric": "drift", "condition": "above", "threshold": ALERT_HIGH}, + "DIVERGENCE_COMMIT": {"metric": "dvg", "condition": "above", "threshold": ALERT_HIGH}, + "INTENSITY_CALM": {"metric": "int_val", "condition": "below", "threshold": ALERT_LOW}, + "BALANCE_CONCISE": {"metric": "tbf", "condition": "below", "threshold": ALERT_LOW}, +} + + +def compute_metrics( + responses: list[dict[str, Any]], + context: str = "", +) -> dict[str, float]: + if not responses: + return {m: 0.0 for m in METRIC_NAMES} + n = len(responses) + texts = [r.get("content", "") for r in responses] + avg_len = sum(len(t) for t in texts) / max(n, 1) + variance = sum((len(t) - avg_len) ** 2 for t in texts) / max(n, 1) + std = math.sqrt(variance) + + cm = min(1.0, avg_len / 2000) if avg_len > 0 else 0.0 + da = max(0.0, 1.0 - std / max(avg_len, 1)) + drift = min(1.0, std / max(avg_len, 1)) + unique_starts = len(set(t[:50] for t in texts if t)) + dvg = min(1.0, unique_starts / max(n, 1)) + int_val = max(0.0, 1.0 - drift * 0.5 - dvg * 0.3) + ctx_overlap = 0.0 + if context: + ctx_words = set(context.lower().split()) + for t in texts: + t_words = set(t.lower().split()) + if ctx_words: + ctx_overlap += len(ctx_words & t_words) / len(ctx_words) + ctx_overlap /= max(n, 1) + tbf = max(0.0, min(1.0, ctx_overlap)) + + return { + "cm": round(cm, 4), + "da": round(da, 4), + "drift": round(drift, 4), + "dvg": round(dvg, 4), + "int_val": round(int_val, 4), + "tbf": round(tbf, 4), + } + + +def check_directives(metrics: dict[str, float]) -> list[str]: + fired = [] + for name, directive in DIRECTIVES.items(): + val = metrics.get(directive["metric"], 0) + if directive["condition"] == "above" and val > directive["threshold"]: + fired.append(name) + elif directive["condition"] == "below" and val < directive["threshold"]: + fired.append(name) + return fired + + +def check_alerts(metrics: dict[str, float]) -> dict[str, list[str]]: + """Return HIGH/LOW alert lists for metrics crossing the 0.80/0.20 thresholds.""" + high = [m for m in METRIC_NAMES if metrics.get(m, 0.0) >= ALERT_HIGH] + low = [m for m in METRIC_NAMES if metrics.get(m, 0.0) <= ALERT_LOW] + return {"HIGH": high, "LOW": low} + + +def delta_between(a: dict[str, float], b: dict[str, float]) -> dict[str, float]: + result = {} + for m in METRIC_NAMES: + result[f"delta_{m}"] = round((b.get(m, 0) - a.get(m, 0)), 4) + return result diff --git a/core/guardian.py b/core/guardian.py new file mode 100644 index 0000000..69a4da7 --- /dev/null +++ b/core/guardian.py @@ -0,0 +1,166 @@ +# 130:12 +""" +Θ (Theta) Guardian Tensor — N=29 prime-node microkernel ring. + - Ragged circle counts per seed: circleCount[i] in [1..12] + - Hash-based instance/key identifiers derived with hashlib + - SHA-256 blueprint hash sharded across all 29 nodes + - Gate control: coherence threshold per node + - Phi injection mirror: node_coherence broadcast → Φ (Task #72) + +Architecturally unique — not parameterized like PTCACore. +Self-declares identity in state() as symbol="Θ", name="theta". +""" + +import hashlib +import os +import time +import numpy as np + +N = 29 +DIMS = 4 +PHASES = 7 +HEPT_SITES = 7 +MIN_CIRCLES = 1 +MAX_CIRCLES = 12 +GATE_THRESHOLD = 0.45 +BLUEPRINT_CHUNK_SIZE = 4 + + +class GuardianTensor: + """Guardian microkernel ring — N=29 nodes, ragged circle counts.""" + + def __init__(self, instance_id: str | None = None, phases: int = 7): + self.phases = phases + rng = np.random.default_rng(seed=29) + self.tensor = rng.uniform(0.2, 0.8, (N, DIMS, phases, HEPT_SITES)).astype(np.float64) + self.velocities = np.zeros_like(self.tensor) + self.node_coherence = np.zeros(N, dtype=np.float64) + self.circle_count = np.array([3] * N, dtype=np.int32) + self.gate_open = np.array([True] * N, dtype=bool) + self.instance_id = instance_id or _gen_instance_id() + self.encryption_key_id = _derive_key_id(self.instance_id) + self.blueprint_hash = _compute_blueprint_hash(self.instance_id) + self.blueprint_shards = _shard_blueprint(self.blueprint_hash, N) + self.reward_history: list[float] = [] + self.step_count = 0 + self.created_at = time.time() + self._recompute_coherence() + + def _recompute_coherence(self): + for i in range(N): + hub = self.tensor[i, :, :, 6] + ring = self.tensor[i, :, :, :6] + diff = np.abs(ring - hub[..., np.newaxis]).mean() + self.node_coherence[i] = float(np.clip(1.0 - diff, 0.0, 1.0)) + self.gate_open[i] = bool(self.node_coherence[i] >= GATE_THRESHOLD) + + def propagate(self, steps: int = 5): + for _ in range(steps): + for i in range(N): + neighbors = [(i - 1) % N, (i + 1) % N, (i + 7) % N, (i - 7) % N] + nb_mean = np.mean([self.tensor[j] for j in neighbors], axis=0) + acc = 0.12 * (nb_mean - self.tensor[i]) - 0.15 * self.tensor[i] + self.velocities[i] = 0.8 * self.velocities[i] + acc * 0.01 + self.tensor[i] = np.clip(self.tensor[i] + self.velocities[i], 0.0, 1.0) + hub_target = self.tensor[i, :, :, :6].mean(axis=-1) + self.tensor[i, :, :, 6] += 0.10 * (hub_target - self.tensor[i, :, :, 6]) + self.step_count += 1 + self._recompute_coherence() + + def apply_reward(self, reward: float): + self.reward_history.append(reward) + if len(self.reward_history) > 100: + self.reward_history = self.reward_history[-100:] + + for i in range(N): + coherence = self.node_coherence[i] + delta = int(round(reward * coherence * 2.0)) + self.circle_count[i] = int(np.clip( + self.circle_count[i] + delta, MIN_CIRCLES, MAX_CIRCLES + )) + + gradient = reward * (self.tensor - 0.5) + self.tensor = np.clip(self.tensor + 0.015 * gradient, 0.0, 1.0) + self._recompute_coherence() + + def gate_status(self) -> list[dict]: + return [ + { + "node": i, + "open": bool(self.gate_open[i]), + "coherence": round(self.node_coherence[i], 4), + "circles": int(self.circle_count[i]), + "shard": self.blueprint_shards[i][:8], + } + for i in range(N) + ] + + def crypto_meta(self) -> dict: + return { + "instance_id": self.instance_id, + "key_id": self.encryption_key_id, + "identifier_derivation": "SHA-256", + "key_id_derivation": "SHA-256", + "implemented_crypto": ["hashing", "identifier-derivation"], + "blueprint_hash": self.blueprint_hash[:16] + "...", + "shards_distributed": N, + } + + def pcta_circle_audit(self) -> list[dict]: + results = [] + for i in range(N): + results.append({ + "node": i, + "circles": int(self.circle_count[i]), + "hub": round(float(self.tensor[i, :, :, 6].mean()), 4), + "ring_mean": round(float(self.tensor[i, :, :, :6].mean()), 4), + "gate": bool(self.gate_open[i]), + "coherence": round(self.node_coherence[i], 4), + }) + return results + + def state(self) -> dict: + open_count = int(self.gate_open.sum()) + return { + "name": "theta", + "symbol": "Θ", + "role": "microkernel", + "ring": "theta", + "ring_alias": "guardian", + "n": N, + "instance_id": self.instance_id, + "ring_coherence": round(float(self.node_coherence.mean()), 4), + "node_coherence": [round(float(v), 4) for v in self.node_coherence], + "gate_open_count": open_count, + "gate_restricted_count": N - open_count, + "circle_counts": [int(v) for v in self.circle_count], + "circle_mean": round(float(self.circle_count.mean()), 2), + "tensor_mean": round(float(self.tensor.mean()), 4), + "step_count": self.step_count, + "reward_history_len": len(self.reward_history), + "last_reward": round(self.reward_history[-1], 4) if self.reward_history else 0.0, + "encryption": self.crypto_meta(), + } + + +def _gen_instance_id() -> str: + return os.urandom(16).hex() + + +def _derive_key_id(instance_id: str) -> str: + return hashlib.sha256(f"a0p-key:{instance_id}".encode()).hexdigest()[:32] + + +def _compute_blueprint_hash(instance_id: str) -> str: + return hashlib.sha256(f"a0p-blueprint:{instance_id}".encode()).hexdigest() + + +def _shard_blueprint(bp_hash: str, n: int) -> list[str]: + chunk = max(1, len(bp_hash) // n) + shards = [] + for i in range(n): + start = (i * chunk) % len(bp_hash) + shard = bp_hash[start:start + BLUEPRINT_CHUNK_SIZE] + shards.append(shard.ljust(BLUEPRINT_CHUNK_SIZE, "0")) + return shards +# 130:12 diff --git a/core/memory_core.py b/core/memory_core.py new file mode 100644 index 0000000..080b4df --- /dev/null +++ b/core/memory_core.py @@ -0,0 +1,85 @@ +import time +import numpy as np + +DIMS = 4 +PHASES = 7 +HEPT_SITES = 7 + +FLUSH_REWARD_THRESHOLD = 0.0 +FLUSH_ALPHA = 0.25 + + +class MemoryCore: + """Parameterized memory ring — self-declares role in state().""" + + def __init__(self, n: int, seed: int, role: str, phases: int = 7): + self.n = n + self.seed = seed + self.role = role + self.phases = phases + + rng = np.random.default_rng(seed=seed) + low = 0.2 if role == "long_term" else 0.1 + high = 0.8 if role == "long_term" else 0.9 + self.tensor = rng.uniform(low, high, (n, DIMS, phases, HEPT_SITES)).astype(np.float64) + self.hub_avg = np.zeros(n, dtype=np.float64) + self._recompute_hub_avg() + self.write_count = 0 + self.flush_count = 0 + self.created_at = time.time() + + def _recompute_hub_avg(self): + for i in range(self.n): + self.hub_avg[i] = float(self.tensor[i, :, :, 6].mean()) + + def write(self, signal: np.ndarray, alpha: float = 0.30): + if signal.ndim == 1 and signal.shape[0] >= 1: + val = float(np.clip(signal.mean(), 0.0, 1.0)) + node_idx = self.write_count % self.n + self.tensor[node_idx, 0, 0, :] = np.clip( + self.tensor[node_idx, 0, 0, :] * (1 - alpha) + val * alpha, 0.0, 1.0 + ) + self._recompute_hub_avg() + self.write_count += 1 + + def absorb(self, other_tensor: np.ndarray, alpha: float = FLUSH_ALPHA): + src_n = other_tensor.shape[0] + for i in range(min(src_n, self.n)): + self.tensor[i] = (1.0 - alpha) * self.tensor[i] + alpha * other_tensor[i] + np.clip(self.tensor[i], 0.0, 1.0, out=self.tensor[i]) + self._recompute_hub_avg() + self.flush_count += 1 + + def query(self, probe: np.ndarray) -> np.ndarray: + scores = np.zeros(self.n) + for i in range(self.n): + node_mean = self.tensor[i].mean(axis=(1, 2)) + scores[i] = float(1.0 - np.abs(node_mean - probe[:DIMS]).mean()) + return scores + + def flush_to(self, target: "MemoryCore", reward: float) -> bool: + if reward > FLUSH_REWARD_THRESHOLD: + target.absorb(self.tensor) + self._reset() + self.flush_count += 1 + return True + return False + + def _reset(self): + rng = np.random.default_rng(seed=int(time.time()) % 10000 + self.seed) + self.tensor = rng.uniform(0.1, 0.5, (self.n, DIMS, self.phases, HEPT_SITES)).astype(np.float64) + self._recompute_hub_avg() + + def state(self) -> dict: + return { + "ring": f"memory_{self.role[0]}", + "role": self.role, + "n": self.n, + "seed": self.seed, + "tensor_mean": round(float(self.tensor.mean()), 4), + "tensor_std": round(float(self.tensor.std()), 4), + "hub_avg": [round(float(v), 4) for v in self.hub_avg], + "avg_hub": round(float(self.hub_avg.mean()), 4), + "write_count": self.write_count, + "flush_count": self.flush_count, + } diff --git a/core/merge.py b/core/merge.py new file mode 100644 index 0000000..18c88c5 --- /dev/null +++ b/core/merge.py @@ -0,0 +1,152 @@ +# 118:8 +""" +Instance Merge Protocol — three modes for multi-instance PCNA mesh. + + absorb — dominant absorbs donor; donor is retired + fork — parent spawns child with copied state + noise; both continue + converge — both exchange tensors via federated averaging; both continue + +Operates on PCNAEngine instances containing PTCACore + MemoryCore + GuardianTensor. +""" + +import time +import numpy as np +from .ptca_core import PTCACore +from .pcna import PCNAEngine + + +def _fed_avg(a: np.ndarray, b: np.ndarray, alpha: float = 0.5) -> np.ndarray: + return np.clip(alpha * a + (1.0 - alpha) * b, 0.0, 1.0) + + +def _blend_core(dst: PTCACore, src: PTCACore, alpha: float): + dst.tensor = _fed_avg(dst.tensor, src.tensor, alpha=1.0 - alpha) + dst._recompute_coherence() + + +class InstanceMerge: + """Stateless merge operator for PCNAEngine instances.""" + + @staticmethod + def absorb(dominant: PCNAEngine, donor: PCNAEngine) -> dict: + alpha = 0.15 + _blend_core(dominant.phi, donor.phi, alpha) + _blend_core(dominant.psi, donor.psi, alpha) + _blend_core(dominant.omega, donor.omega, alpha) + + dominant.guardian.tensor = _fed_avg( + dominant.guardian.tensor, donor.guardian.tensor, alpha=1.0 - alpha + ) + dominant.memory_l.tensor = _fed_avg( + dominant.memory_l.tensor, donor.memory_l.tensor, alpha=0.8 + ) + + for i in range(min(len(dominant.guardian.circle_count), len(donor.guardian.circle_count))): + dominant.guardian.circle_count[i] = max( + dominant.guardian.circle_count[i], + donor.guardian.circle_count[i], + ) + + dominant.guardian._recompute_coherence() + dominant.memory_l._recompute_hub_avg() + + phi_c = round(dominant.phi.ring_coherence, 4) + guard_c = round(float(dominant.guardian.node_coherence.mean()), 4) + + return { + "mode": "absorb", + "dominant_id": dominant.guardian.instance_id, + "donor_id": donor.guardian.instance_id, + "donor_status": "retired", + "dominant_phi_coherence": phi_c, + "dominant_guardian_coherence": guard_c, + "dominant_psi_coherence": round(dominant.psi.ring_coherence, 4), + "dominant_omega_coherence": round(dominant.omega.ring_coherence, 4), + "circle_counts_after": [int(v) for v in dominant.guardian.circle_count], + "timestamp": time.time(), + } + + @staticmethod + def fork(parent: PCNAEngine) -> tuple[PCNAEngine, dict]: + child = PCNAEngine() + noise = np.random.default_rng(int(time.time() * 1000) % 2**32) + + for attr in ("phi", "psi", "omega"): + p_core: PTCACore = getattr(parent, attr) + c_core: PTCACore = getattr(child, attr) + c_core.tensor = np.clip( + p_core.tensor + noise.normal(0, 0.02, p_core.tensor.shape), 0.0, 1.0 + ) + c_core._recompute_coherence() + + child.guardian.tensor = np.clip( + parent.guardian.tensor + noise.normal(0, 0.01, parent.guardian.tensor.shape), 0.0, 1.0 + ) + child.memory_l.tensor = parent.memory_l.tensor.copy() + child.guardian.circle_count = parent.guardian.circle_count.copy() + child.guardian.blueprint_shards = parent.guardian.blueprint_shards[:] + child.guardian._recompute_coherence() + child.memory_l._recompute_hub_avg() + + result = { + "mode": "fork", + "parent_id": parent.guardian.instance_id, + "child_id": child.guardian.instance_id, + "parent_status": "continues", + "child_status": "spawned", + "child_phi_coherence": round(child.phi.ring_coherence, 4), + "child_psi_coherence": round(child.psi.ring_coherence, 4), + "child_omega_coherence": round(child.omega.ring_coherence, 4), + "timestamp": time.time(), + } + return child, result + + @staticmethod + def converge(a: PCNAEngine, b: PCNAEngine, alpha: float = 0.5) -> dict: + for attr in ("phi", "psi", "omega"): + core_a: PTCACore = getattr(a, attr) + core_b: PTCACore = getattr(b, attr) + new_a = _fed_avg(core_a.tensor, core_b.tensor, alpha) + new_b = _fed_avg(core_b.tensor, core_a.tensor, alpha) + core_a.tensor = new_a + core_b.tensor = new_b + core_a._recompute_coherence() + core_b._recompute_coherence() + + new_ga = _fed_avg(a.guardian.tensor, b.guardian.tensor, alpha) + new_gb = _fed_avg(b.guardian.tensor, a.guardian.tensor, alpha) + new_mla = _fed_avg(a.memory_l.tensor, b.memory_l.tensor, alpha=0.6) + new_mlb = _fed_avg(b.memory_l.tensor, a.memory_l.tensor, alpha=0.6) + + a.guardian.tensor = new_ga + b.guardian.tensor = new_gb + a.memory_l.tensor = new_mla + b.memory_l.tensor = new_mlb + + for i in range(min(len(a.guardian.circle_count), len(b.guardian.circle_count))): + avg = (int(a.guardian.circle_count[i]) + int(b.guardian.circle_count[i])) // 2 + a.guardian.circle_count[i] = avg + b.guardian.circle_count[i] = avg + + a.guardian._recompute_coherence() + b.guardian._recompute_coherence() + a.memory_l._recompute_hub_avg() + b.memory_l._recompute_hub_avg() + + return { + "mode": "converge", + "instance_a": a.guardian.instance_id, + "instance_b": b.guardian.instance_id, + "alpha": alpha, + "a_phi_coherence_after": round(a.phi.ring_coherence, 4), + "b_phi_coherence_after": round(b.phi.ring_coherence, 4), + "a_guardian_coherence_after": round(float(a.guardian.node_coherence.mean()), 4), + "b_guardian_coherence_after": round(float(b.guardian.node_coherence.mean()), 4), + "a_psi_coherence_after": round(a.psi.ring_coherence, 4), + "b_psi_coherence_after": round(b.psi.ring_coherence, 4), + "a_omega_coherence_after": round(a.omega.ring_coherence, 4), + "b_omega_coherence_after": round(b.omega.ring_coherence, 4), + "both_status": "converged", + "timestamp": time.time(), + } +# 118:8 diff --git a/core/pcna.py b/core/pcna.py new file mode 100644 index 0000000..2e219ae --- /dev/null +++ b/core/pcna.py @@ -0,0 +1,351 @@ +# 295:27 +""" +PCNA Inference Engine — six-ring pipeline, all rings real. + +Six rings: + +Φ (phi) N=53, seed=53 — cognitive substrate +Ψ (psi) N=53, seed=43 — self-model +Ω (omega) N=53, seed=47 — autonomy +Guardian N=29 — microkernel gate +Memory-L N=19, seed=19 — long-term +Memory-S N=17, seed=17 — short-term + +Six inference steps: + +1. Project — encode input text → normalized signal vector +2. Inject — push signal into Φ, self-referential into Ψ, autonomy into Ω +3. Propagate — run heptagram propagation on Φ/Ψ/Ω + guardian +4. PTCA-seed — per-prime-node audit on all three PTCA cores +5. PCTA-circle — guardian circle audit +6. Coherence — weighted ring coherence → winner + confidence + +Backprop: + +reward(winner, outcome) → nudge all three PTCA cores + guardian + memory flush +""" + +import base64 +import hashlib +import io +import os +import time +import numpy as np + +from .ptca_core import PTCACore +from .memory_core import MemoryCore +from .guardian import GuardianTensor + + +def _tensor_to_b64(arr: np.ndarray) -> str: + buf = io.BytesIO() + np.save(buf, arr) + return base64.b64encode(buf.getvalue()).decode() + + +def _b64_to_tensor(s: str) -> np.ndarray: + return np.load(io.BytesIO(base64.b64decode(s))) + + +RING_WEIGHTS = { + "phi": 0.30, + "psi": 0.15, + "omega": 0.15, + "guardian": 0.20, + "memory_l": 0.12, + "memory_s": 0.08, +} + +WINNER_RINGS = ["phi", "psi", "omega"] + +_CHECKPOINT_DIR = os.path.join(os.path.dirname(__file__), "..", ".checkpoints") + + +class PCNAEngine: + """PCNA six-ring inference engine — no stubs, all rings real.""" + + def __init__(self, phases: int = 7): + self.phases = phases + self.phi = PTCACore(name="phi", symbol="Φ", role="cognitive", n=53, seed=53, phases=phases) + self.psi = PTCACore(name="psi", symbol="Ψ", role="self_model", n=53, seed=43, phases=phases) + self.omega = PTCACore(name="omega", symbol="Ω", role="autonomy", n=53, seed=47, phases=phases) + self.memory_l = MemoryCore(n=19, seed=19, role="long_term", phases=phases) + self.memory_s = MemoryCore(n=17, seed=17, role="short_term", phases=phases) + self.guardian = GuardianTensor(phases=phases) + self.infer_count = 0 + self.reward_count = 0 + self.last_coherence = 0.0 + self.last_winner = "phi" + self.blueprint_hash = self.guardian.blueprint_hash + self.created_at = time.time() + self.checkpoint_at: float | None = None + self.checkpoint_ring_means: dict[str, float] = {} + self._checkpoint_key = "pcna_checkpoint" if phases == 7 else f"pcna_checkpoint_p{phases}" + + def load_checkpoint(self): + """Restore ring tensors from numpy checkpoint file.""" + try: + path = os.path.join(_CHECKPOINT_DIR, f"{self._checkpoint_key}.npz") + if not os.path.exists(path): + return + with np.load(path, allow_pickle=False) as data: + ring_map = { + "phi": self.phi, + "psi": self.psi, + "omega": self.omega, + "memory_l": self.memory_l, + "memory_s": self.memory_s, + } + for name, ring in ring_map.items(): + t_key = f"{name}_tensor" + if t_key not in data: + print(f"[pcna] checkpoint missing key: {t_key}") + return + tensor = data[t_key] + if tensor.shape != ring.tensor.shape: + print(f"[pcna] checkpoint shape mismatch on {name}: {tensor.shape} vs {ring.tensor.shape}") + return + ring.tensor = tensor + v_key = f"{name}_velocities" + if hasattr(ring, "velocities") and v_key in data: + vel = data[v_key] + if vel.shape == ring.velocities.shape: + ring.velocities = vel + if hasattr(ring, "_recompute_coherence"): + ring._recompute_coherence() + elif hasattr(ring, "_recompute_hub_avg"): + ring._recompute_hub_avg() + ts = float(data["saved_at"]) if "saved_at" in data else 0.0 + self.checkpoint_at = ts if ts else None + self.checkpoint_ring_means = { + name: round(float(ring_map[name].tensor.mean()), 4) for name in ring_map + } + print(f"[pcna] checkpoint restored: {len(ring_map)} rings, saved_at={ts}") + except Exception as e: + print(f"[pcna] checkpoint load failed (fresh start): {e}") + + def save_checkpoint(self): + """Serialize all ring tensors to numpy checkpoint file.""" + try: + os.makedirs(_CHECKPOINT_DIR, exist_ok=True) + rings = { + "phi": self.phi, + "psi": self.psi, + "omega": self.omega, + "memory_l": self.memory_l, + "memory_s": self.memory_s, + } + arrays = {"saved_at": np.array(time.time())} + for name, ring in rings.items(): + arrays[f"{name}_tensor"] = ring.tensor + if hasattr(ring, "velocities"): + arrays[f"{name}_velocities"] = ring.velocities + path = os.path.join(_CHECKPOINT_DIR, f"{self._checkpoint_key}.npz") + np.savez(path, **arrays) + self.checkpoint_at = float(arrays["saved_at"]) + self.checkpoint_ring_means = { + name: round(float(ring.tensor.mean()), 4) for name, ring in rings.items() + } + print(f"[pcna] checkpoint saved: {len(rings)} rings") + except Exception as e: + print(f"[pcna] checkpoint save failed: {e}") + + def _project(self, text: str) -> np.ndarray: + h = hashlib.sha512(text.encode("utf-8")).digest() + arr = np.frombuffer(h, dtype=np.uint8).astype(np.float64) + arr = arr / 255.0 + padded = np.tile(arr, 4)[:53] + return padded + + def _inject(self, signal: np.ndarray): + self.phi.inject(signal) + self.phi._recompute_coherence() + self.memory_s.write(signal) + + theta_nc = self.guardian.node_coherence + theta_signal = np.full(53, float(theta_nc.mean()), dtype=np.float64) + theta_signal[:len(theta_nc)] = theta_nc + self.phi.inject(theta_signal) + self.phi._recompute_coherence() + + psi_signal = np.full(53, self.phi.ring_coherence, dtype=np.float64) + phi_node_c = self.phi.node_coherence + psi_signal[:len(phi_node_c)] = phi_node_c + self.psi.inject(psi_signal) + + try: + from .sigma import get_sigma + _sig = get_sigma() + if _sig.tensor is not None and _sig.n > 0: + sigma_signal = np.full(53, _sig.ring_coherence, dtype=np.float64) + nc = _sig.node_coherence + top = min(len(nc), 53) + sigma_signal[:top] = nc[:top] + self.psi.inject(sigma_signal) + except Exception: + pass + + ml_hub = self.memory_l.hub_avg + omega_base = np.full(53, float(ml_hub.mean()), dtype=np.float64) + omega_base[:len(ml_hub)] *= ml_hub + omega_base = np.clip(omega_base, 0.0, 1.0) + self.omega.inject(omega_base) + + def _propagate(self): + self.phi.propagate(steps=10) + self.psi.propagate(steps=8) + self.omega.propagate(steps=6) + self.guardian.propagate(steps=5) + + def _ptca_seed_audit(self) -> dict: + cores = {"phi": self.phi, "psi": self.psi, "omega": self.omega} + result = {} + for name, core in cores.items(): + audit = core.ptca_seed_audit() + result[f"{name}_nodes_audited"] = len(audit) + result[f"{name}_coherence"] = round(core.ring_coherence, 4) + result[f"{name}_top3"] = sorted(audit, key=lambda x: x["coherence"], reverse=True)[:3] + result[f"{name}_bottom3"] = sorted(audit, key=lambda x: x["coherence"])[:3] + result["memory_s_hub_avg"] = self.memory_s.state()["avg_hub"] + return result + + def _pcta_circle_audit(self) -> dict: + g_audit = self.guardian.pcta_circle_audit() + open_nodes = [n for n in g_audit if n["gate"]] + closed_nodes = [n for n in g_audit if not n["gate"]] + return { + "guardian_nodes": len(g_audit), + "gates_open": len(open_nodes), + "gates_closed": len(closed_nodes), + "avg_circles": round(sum(n["circles"] for n in g_audit) / len(g_audit), 2), + "guardian_coherence": round(float(self.guardian.node_coherence.mean()), 4), + "memory_l_hub_avg": self.memory_l.state()["avg_hub"], + } + + def _coherence_score(self, seed_audit: dict, circle_audit: dict) -> dict: + ring_scores = { + "phi": seed_audit["phi_coherence"], + "psi": seed_audit["psi_coherence"], + "omega": seed_audit["omega_coherence"], + "guardian": circle_audit["guardian_coherence"], + "memory_l": self.memory_l.state()["avg_hub"], + "memory_s": self.memory_s.state()["avg_hub"], + } + weighted = sum(RING_WEIGHTS[r] * ring_scores[r] for r in ring_scores) + winner = max(WINNER_RINGS, key=lambda r: ring_scores[r]) + confidence = float(np.clip(weighted, 0.0, 1.0)) + return { + "ring_scores": {k: round(v, 4) for k, v in ring_scores.items()}, + "weighted_coherence": round(weighted, 4), + "winner": winner, + "confidence": round(confidence, 4), + } + + def infer(self, text: str) -> dict: + t0 = time.time() + signal = self._project(text) + self._inject(signal) + self._propagate() + + seed_audit = self._ptca_seed_audit() + circle_audit = self._pcta_circle_audit() + coherence = self._coherence_score(seed_audit, circle_audit) + + self.infer_count += 1 + self.last_coherence = coherence["weighted_coherence"] + self.last_winner = coherence["winner"] + + elapsed_ms = round((time.time() - t0) * 1000, 1) + + return { + "step": "pcna_infer", + "infer_index": self.infer_count, + "blueprint_hash": self.blueprint_hash[:16] + "...", + "elapsed_ms": elapsed_ms, + "signal_mean": round(float(signal.mean()), 4), + "step1_project": {"signal_len": len(signal), "signal_mean": round(float(signal.mean()), 4)}, + "step2_inject": {"phi_n": 53, "psi_n": 53, "omega_n": 53, "memory_s_n": 17}, + "step3_propagate": {"phi_steps": 10, "psi_steps": 8, "omega_steps": 6, "guardian_steps": 5}, + "step4_ptca_seed": seed_audit, + "step5_pcta_circle": circle_audit, + "step6_coherence": coherence, + "coherence_score": coherence["weighted_coherence"], + "winner": coherence["winner"], + "confidence": coherence["confidence"], + "guardian_circles": int(self.guardian.circle_count.mean()), + "memory_l_state": self.memory_l.state(), + "memory_s_state": self.memory_s.state(), + } + + def reward(self, winner: str, outcome: float) -> dict: + self.phi.nudge(outcome, lr=0.025) + self.psi.nudge(outcome, lr=0.020) + self.omega.nudge(outcome, lr=0.015) + self.guardian.apply_reward(outcome) + flushed = self.memory_s.flush_to(self.memory_l, outcome) + + try: + from .sigma import get_sigma + get_sigma().nudge(outcome, lr=0.015) + except Exception: + pass + + self.reward_count += 1 + + return { + "step": "pcna_reward", + "reward_index": self.reward_count, + "winner": winner, + "outcome": round(outcome, 4), + "nudged": True, + "nudged_cores": ["phi", "psi", "omega", "theta", "sigma"], + "memory_flush": flushed, + "phi_coherence_after": round(self.phi.ring_coherence, 4), + "psi_coherence_after": round(self.psi.ring_coherence, 4), + "omega_coherence_after": round(self.omega.ring_coherence, 4), + "theta_coherence_after": round(float(self.guardian.node_coherence.mean()), 4), + "guardian_circles_after": [int(v) for v in self.guardian.circle_count], + "memory_l_flush_count": self.memory_l.flush_count, + "memory_s_flush_count": self.memory_s.flush_count, + } + + def state(self) -> dict: + try: + from .zeta import _zeta_engine + echo_history = list(_zeta_engine.echo_buffer) if _zeta_engine else [] + except Exception: + echo_history = [] + + try: + from .sigma import get_sigma + sigma_state = get_sigma().state() + except Exception: + sigma_state = {} + + guardian_state = self.guardian.state() + + return { + "engine": "pcna", + "version": "2.2.0", + "phases": self.phases, + "infer_count": self.infer_count, + "reward_count": self.reward_count, + "last_coherence": round(self.last_coherence, 4), + "last_winner": self.last_winner, + "rings": { + "phi": self.phi.state(), + "psi": self.psi.state(), + "omega": self.omega.state(), + "theta": guardian_state, + "guardian": guardian_state, + "sigma": sigma_state, + "memory_l": self.memory_l.state(), + "memory_s": self.memory_s.state(), + }, + "ring_weights": RING_WEIGHTS, + "uptime_s": round(time.time() - self.created_at, 1), + "checkpoint_at": self.checkpoint_at, + "checkpoint_ring_means": self.checkpoint_ring_means, + "echo_history": echo_history[-20:], + } +# 295:27 diff --git a/core/ptca_core.py b/core/ptca_core.py new file mode 100644 index 0000000..ecd9948 --- /dev/null +++ b/core/ptca_core.py @@ -0,0 +1,150 @@ +# 119:9 +""" +PTCACore — parameterized prime-ring tensor with heptagram propagation. +Each instance self-declares: name, symbol, role, n, seed. +Tensor shape: [N, DIMS=4, PHASES=7, HEPT_SITES=7] +""" + +import math +import time +import numpy as np + +DIMS = 4 +PHASES = 7 +HEPT_SITES = 7 + +DT = 0.01 +ALPHA_COUPLING = 0.10 +BETA_DRIFT = 0.40 +GAMMA_DAMPING = 0.20 +STEPS_PER_EVAL = 10 + + +def _adj_distances(n: int) -> list[int]: + base = [1, 2, 3, 4, 5, 6, 7] + scaled = [d for d in base if d < n] + gap = max(1, math.ceil(n / 4)) + if gap not in scaled and gap < n: + scaled.append(gap) + return scaled + + +class PTCACore: + """ + Prime-ring PTCA core. Parameterized by (name, symbol, role, n, seed). + Every instance self-declares its identity in state(). + """ + + def __init__(self, name: str, symbol: str, role: str, n: int, seed: int, phases: int = 7): + self.name = name + self.symbol = symbol + self.role = role + self.n = n + self.seed = seed + self.phases = phases + self._adj_dists = _adj_distances(n) + + rng = np.random.default_rng(seed=seed) + self.tensor = rng.uniform(0.1, 0.9, (n, DIMS, phases, HEPT_SITES)).astype(np.float64) + self.velocities = np.zeros((n, DIMS, phases, HEPT_SITES), dtype=np.float64) + self.node_coherence = np.zeros(n, dtype=np.float64) + self.ring_coherence = 0.0 + self.step_count = 0 + self.last_reward = 0.0 + self.created_at = time.time() + self._recompute_coherence() + + def _adjacents(self, i: int) -> list[int]: + fwd = [(i + d) % self.n for d in self._adj_dists] + bwd = [(i - d) % self.n for d in self._adj_dists] + return fwd + bwd + + def _propagate_node(self, i: int): + neighbors = self._adjacents(i) + neighbor_avg = np.mean([self.tensor[j] for j in neighbors], axis=0) + + coupling = ALPHA_COUPLING * (neighbor_avg - self.tensor[i]) + drift = BETA_DRIFT * self.velocities[i] + damping = -GAMMA_DAMPING * self.tensor[i] + + acc = coupling + damping + self.velocities[i] += acc * DT + self.tensor[i] += (self.velocities[i] + drift) * DT + np.clip(self.tensor[i], 0.0, 1.0, out=self.tensor[i]) + + hub = self.tensor[i, :, :, 6] + ring = self.tensor[i, :, :, :6] + hub_target = ring.mean(axis=-1) + self.tensor[i, :, :, 6] += 0.15 * (hub_target - hub) + + def propagate(self, steps: int = STEPS_PER_EVAL): + for _ in range(steps): + for i in range(self.n): + self._propagate_node(i) + self.step_count += 1 + self._recompute_coherence() + + def _recompute_coherence(self): + for i in range(self.n): + hub = self.tensor[i, :, :, 6] + ring = self.tensor[i, :, :, :6] + diff = np.abs(ring - hub[..., np.newaxis]).mean() + self.node_coherence[i] = float(np.clip(1.0 - diff, 0.0, 1.0)) + self.ring_coherence = float(self.node_coherence.mean()) + + def inject(self, signal: np.ndarray): + if signal.ndim == 1 and signal.shape[0] == self.n: + for i in range(self.n): + self.tensor[i, 0, 0, :] = np.clip( + self.tensor[i, 0, 0, :] * 0.85 + signal[i] * 0.15, 0.0, 1.0 + ) + elif signal.ndim == 2 and signal.shape == (self.n, DIMS): + for i in range(self.n): + self.tensor[i, :, 0, :] = np.clip( + self.tensor[i, :, 0, :] * 0.85 + signal[i, :, np.newaxis] * 0.15, 0.0, 1.0 + ) + + def nudge(self, reward: float, lr: float = 0.02): + self.last_reward = reward + gradient = reward * (self.tensor - 0.5) + self.tensor = np.clip(self.tensor + lr * gradient, 0.0, 1.0) + self._recompute_coherence() + + def ptca_seed_audit(self) -> list[dict]: + results = [] + for i in range(self.n): + hub_val = float(self.tensor[i, :, :, 6].mean()) + ring_mean = float(self.tensor[i, :, :, :6].mean()) + phase_var = float(self.tensor[i, 0, :, :].var()) + coherence = self.node_coherence[i] + results.append({ + "node": i, + "hub": round(hub_val, 4), + "ring_mean": round(ring_mean, 4), + "phase_var": round(phase_var, 4), + "coherence": round(coherence, 4), + }) + return results + + def state(self) -> dict: + return { + "name": self.name, + "symbol": self.symbol, + "role": self.role, + "ring": self.name, + "n": self.n, + "seed": self.seed, + "dims": DIMS, + "phases": self.phases, + "hept_sites": HEPT_SITES, + "ring_coherence": round(self.ring_coherence, 4), + "node_coherence_mean": round(float(self.node_coherence.mean()), 4), + "node_coherence_min": round(float(self.node_coherence.min()), 4), + "node_coherence_max": round(float(self.node_coherence.max()), 4), + "tensor_mean": round(float(self.tensor.mean()), 4), + "tensor_std": round(float(self.tensor.std()), 4), + "step_count": self.step_count, + "last_reward": round(self.last_reward, 4), + "node_coherence": [round(float(v), 4) for v in self.node_coherence], + } +# 119:9 diff --git a/core/sigma.py b/core/sigma.py new file mode 100644 index 0000000..956f2eb --- /dev/null +++ b/core/sigma.py @@ -0,0 +1,106 @@ +""" +Σ (Sigma) — Filesystem Observer Ring + +Wraps PTCACore to add file-content watching. +Sigma injects coherence signals into the Ψ (psi) self-model ring +whenever watched files change. + +N=41, seed=41 — observer substrate +""" + +import os +import time +from typing import Optional + +import numpy as np + +from .ptca_core import PTCACore + +N = 41 +SEED = 41 +DEFAULT_CONTENT_INTERVAL = 10.0 +DEFAULT_STRUCTURAL_INTERVAL = 30.0 + + +class SigmaRing: + """Filesystem-aware PTCACore ring. Drains file-change events on demand.""" + + def __init__(self): + self._core = PTCACore(name="sigma", symbol="Σ", role="observer", n=N, seed=SEED) + self.content_interval: float = DEFAULT_CONTENT_INTERVAL + self.structural_interval: float = DEFAULT_STRUCTURAL_INTERVAL + self._resolution: int = 3 + self._watched: dict[str, float] = {} # path → last mtime + self._pending: list[str] = [] + self._last_check: float = 0.0 + + # --- PTCACore passthrough --- + + @property + def tensor(self) -> Optional[np.ndarray]: + return self._core.tensor + + @property + def n(self) -> int: + return self._core.n + + @property + def ring_coherence(self) -> float: + return self._core.ring_coherence + + @property + def node_coherence(self) -> np.ndarray: + return self._core.node_coherence + + def nudge(self, reward: float, lr: float = 0.02) -> None: + self._core.nudge(reward, lr=lr) + + def state(self) -> dict: + s = self._core.state() + s["resolution"] = self._resolution + s["watched_count"] = len(self._watched) + s["content_interval"] = self.content_interval + s["structural_interval"] = self.structural_interval + return s + + # --- file watching --- + + def set_resolution(self, level: int) -> None: + self._resolution = max(1, min(5, level)) + + def add_content_watch(self, path: str) -> None: + try: + mtime = os.path.getmtime(path) + except OSError: + mtime = 0.0 + self._watched[path] = mtime + + def remove_content_watch(self, path: str) -> None: + self._watched.pop(path, None) + + def drain_content_changed_events(self) -> list[str]: + """Check watched files for mtime changes; return paths that changed.""" + now = time.time() + if now - self._last_check >= self.content_interval: + self._last_check = now + for path, last_mtime in list(self._watched.items()): + try: + mtime = os.path.getmtime(path) + except OSError: + continue + if mtime != last_mtime: + self._watched[path] = mtime + self._pending.append(path) + drained = self._pending[:] + self._pending = [] + return drained + + +_sigma: Optional[SigmaRing] = None + + +def get_sigma() -> SigmaRing: + global _sigma + if _sigma is None: + _sigma = SigmaRing() + return _sigma diff --git a/core/zeta.py b/core/zeta.py new file mode 100644 index 0000000..e35a805 --- /dev/null +++ b/core/zeta.py @@ -0,0 +1,459 @@ +# 198:61 + +""" + +ZetaEngine — Zeta Function Alpha Echo + +ZFAE passively learns from every energy provider response. + +Every assistant reply is evaluated by EDCM (no LLM), producing a coherence + +score that drives PCNA phi/psi/omega reward backprop. + +Naming: a0(zeta fun alpha echo) {provider} + +- zeta = the observer function + +- fun = the phi ring coherence transform + +- alpha = the learning rate parameter + +- echo = the feedback signal returned to the ring + +No external API calls. Runs non-blocking after every chat response. + +Resolution: + +Each directory path can carry its own resolution level (1–5). The most + +specific matching prefix wins; the global level applies when nothing matches. + +Level 1 = minimal/lightweight observation. Level 5 = maximum depth. + +Example: global=3, /system=5 means system-root paths are observed at full depth. + +""" + +import time + +from collections import deque + +from typing import Optional + +_DEFAULT_RESOLUTION = 3 + +_MIN_RES = 1 + +_MAX_RES = 5 + + +class ZetaEngine: + + """ + + Non-LLM real-time learning engine with per-directory resolution control. + + Evaluates each assistant response via EDCM and drives PCNA backprop. + + """ + + AGENT_NAME = "a0(zeta fun alpha echo)" + + def __init__(self, buffer_size: int = 50): + + self.echo_buffer: deque = deque(maxlen=buffer_size) + + self.eval_count = 0 + + self.created_at = time.time() + + self.resolution_config: dict = { + + "global": _DEFAULT_RESOLUTION, + + "directories": {}, + + } + + def get_resolution(self, path: str = "") -> int: + + """Return the resolution level for the given path.""" + + config = self.resolution_config + + dirs = config.get("directories", {}) + + if not path or not dirs: + + return config.get("global", _DEFAULT_RESOLUTION) + + normalized = path.rstrip("/") + + best_level: Optional[int] = None + + best_len = -1 + + for dir_path, level in dirs.items(): + + dp = dir_path.rstrip("/") + + if normalized == dp or normalized.startswith(dp + "/"): + + if len(dp) > best_len: + + best_level = level + + best_len = len(dp) + + return best_level if best_level is not None else config.get("global", _DEFAULT_RESOLUTION) + + def set_global_resolution(self, level: int) -> dict: + + self.resolution_config["global"] = max(_MIN_RES, min(_MAX_RES, level)) + + return dict(self.resolution_config) + + def set_directory_resolution(self, path: str, level: int) -> dict: + + self.resolution_config.setdefault("directories", {})[path] = max(_MIN_RES, min(_MAX_RES, level)) + + return dict(self.resolution_config) + + def remove_directory_resolution(self, path: str) -> dict: + + self.resolution_config.get("directories", {}).pop(path, None) + + return dict(self.resolution_config) + + def load_resolution_config(self, config: dict) -> None: + + if not isinstance(config, dict): + + return + + self.resolution_config = { + + "global": max(_MIN_RES, min(_MAX_RES, int(config.get("global", _DEFAULT_RESOLUTION)))), + + "directories": { + + k: max(_MIN_RES, min(_MAX_RES, int(v))) + + for k, v in config.get("directories", {}).items() + + if isinstance(k, str) and isinstance(v, (int, float)) + + }, + + } + + def _coherence_from_metrics(self, metrics: dict) -> float: + + cm = metrics.get("cm", 0.0) + + da = metrics.get("da", 0.0) + + int_val = metrics.get("int_val", 0.0) + + drift = metrics.get("drift", 0.0) + + coherence = (cm * 0.35 + da * 0.25 + int_val * 0.25 + (1.0 - drift) * 0.15) + + return round(max(0.0, min(1.0, coherence)), 4) + + def _sigma_nudge_factors(self) -> tuple[float, float]: + + change_boost = 1.0 + + substrate_factor = 1.0 + + try: + + from .sigma import get_sigma + + except ImportError: + + return change_boost, substrate_factor + + try: + + sig = get_sigma() + + drained = sig.drain_content_changed_events() + + if drained: + + change_boost = 1.2 + + substrate_factor = round(0.8 + sig.ring_coherence * 0.4, 4) + + except Exception as exc: + + print(f"[zfae:sigma_factors] error reading Sigma factors: {exc}") + + return change_boost, substrate_factor + + def _theta_gate_factor(self) -> float: + + try: + + guardian = _get_default_pcna().guardian + + open_frac = float(guardian.gate_open.mean()) + + return round(0.8 + open_frac * 0.4, 4) + + except Exception as exc: + + print(f"[zfae:gate_factor] error reading Theta gate factor: {exc}") + + return 1.0 + + async def evaluate( + + self, + + assistant_text: str, + + provider: str, + + user_text: str = "", + + path: str = "", + + ) -> dict: + + resolution = self.get_resolution(path) + + try: + + from .edcm import compute_metrics + + metrics = compute_metrics( + + responses=[{"content": assistant_text}], + + context=user_text, + + ) + + coherence = self._coherence_from_metrics(metrics) + + base_lr = 0.025 + + gate_factor = self._theta_gate_factor() + + change_boost, substrate_factor = self._sigma_nudge_factors() + + effective_lr = base_lr * gate_factor * change_boost * substrate_factor + + try: + + pcna = _get_default_pcna() + + pcna.phi.nudge(coherence, lr=effective_lr) + + except Exception: + + pass + + self.eval_count += 1 + + event = { + + "agent": self.AGENT_NAME, + + "provider": provider, + + "coherence": coherence, + + "cm": metrics.get("cm"), + + "da": metrics.get("da"), + + "drift": metrics.get("drift"), + + "int_val": metrics.get("int_val"), + + "resolution": resolution, + + "path": path or None, + + "base_lr": base_lr, + + "gate_factor": gate_factor, + + "change_boost": change_boost, + + "substrate_factor": substrate_factor, + + "effective_lr": round(effective_lr, 6), + + "ts": time.time(), + + } + + self.echo_buffer.append(event) + + suffix = f" path={path}" if path else "" + + print( + + f"[zfae:echo] provider={provider} coherence={coherence}" + + f" lr={effective_lr:.4f}" + + f" gate={gate_factor} boost={change_boost} sub={substrate_factor}" + + f" resolution={resolution}{suffix}" + + ) + + return event + + except Exception as e: + + print(f"[zfae:echo] error: {e}") + + return {} + + def set_sigma_resolution(self, level: int) -> dict: + + try: + + from .sigma import get_sigma + + get_sigma().set_resolution(level) + + event = {"type": "sigma_resolution", "level": level, "ts": time.time()} + + self.echo_buffer.append(event) + + print(f"[zfae:sigma] resolution set to {level}") + + return event + + except Exception as exc: + + print(f"[zfae:sigma] set_resolution error: {exc}") + + return {} + + def sigma_watch_file(self, path: str) -> dict: + + try: + + from .sigma import get_sigma + + get_sigma().add_content_watch(path) + + event = {"type": "sigma_watch_add", "path": path, "ts": time.time()} + + self.echo_buffer.append(event) + + print(f"[zfae:sigma] watching {path}") + + return event + + except Exception as exc: + + print(f"[zfae:sigma] watch_file error: {exc}") + + return {} + + def sigma_unwatch_file(self, path: str) -> dict: + + try: + + from .sigma import get_sigma + + get_sigma().remove_content_watch(path) + + event = {"type": "sigma_watch_remove", "path": path, "ts": time.time()} + + self.echo_buffer.append(event) + + print(f"[zfae:sigma] unwatched {path}") + + return event + + except Exception as exc: + + print(f"[zfae:sigma] unwatch_file error: {exc}") + + return {} + + def set_sigma_structural_interval(self, seconds: float) -> dict: + + try: + + from .sigma import get_sigma + + get_sigma().structural_interval = max(1.0, seconds) + + event = {"type": "sigma_structural_interval", "seconds": seconds, "ts": time.time()} + + self.echo_buffer.append(event) + + print(f"[zfae:sigma] structural interval → {seconds}s") + + return event + + except Exception as exc: + + print(f"[zfae:sigma] set_structural_interval error: {exc}") + + return {} + + def set_sigma_content_interval(self, seconds: float) -> dict: + + try: + + from .sigma import get_sigma + + get_sigma().content_interval = max(1.0, seconds) + + event = {"type": "sigma_content_interval", "seconds": seconds, "ts": time.time()} + + self.echo_buffer.append(event) + + print(f"[zfae:sigma] content interval → {seconds}s") + + return event + + except Exception as exc: + + print(f"[zfae:sigma] set_content_interval error: {exc}") + + return {} + + def state(self) -> dict: + + return { + + "agent": self.AGENT_NAME, + + "eval_count": self.eval_count, + + "echo_buffer_len": len(self.echo_buffer), + + "uptime_s": round(time.time() - self.created_at, 1), + + "resolution": self.resolution_config, + + } + + +_zeta_engine = ZetaEngine() + +_default_pcna = None + + +def _get_default_pcna(): + global _default_pcna + if _default_pcna is None: + from .pcna import PCNAEngine + _default_pcna = PCNAEngine() + return _default_pcna + +# 198:61 diff --git a/main.py b/main.py index fb6b040..b424ba4 100644 --- a/main.py +++ b/main.py @@ -19,8 +19,8 @@ import uvicorn import aiohttp -from src.core.topology import PCNATopology, SeedRole -from src.core.tensor_engine import TensorState, MarkovRecursion +from core.topology import PCNATopology, SeedRole +from core.tensor_engine import TensorState, MarkovRecursion logger = logging.getLogger("pcna") logging.basicConfig(level=logging.INFO) @@ -156,4 +156,4 @@ async def receive_delta(delta: Dict): if __name__ == "__main__": # Useful for local development: honor PORT env var and SEED_ID/ROLE port = int(os.getenv("PORT", os.getenv("PORT0", "8000"))) - uvicorn.run("src.main:app", host="0.0.0.0", port=port, log_level="info") + uvicorn.run("main:app", host="0.0.0.0", port=port, log_level="info") diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..823b507 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +numpy>=1.24.0 +fastapi>=0.104.0 +uvicorn>=0.24.0 +aiohttp>=3.9.0 +pydantic>=2.0.0 diff --git a/tests/test_edcm_engine.py b/tests/test_edcm_engine.py new file mode 100644 index 0000000..5e3b1e0 --- /dev/null +++ b/tests/test_edcm_engine.py @@ -0,0 +1,93 @@ +import asyncio +from backend.edcm_engine import EDCMAnalyzer +from core.edcm import compute_metrics, check_alerts, check_directives + + +def _make_seeds(health_scores, masses=None, roles=None): + if masses is None: + masses = [1.0] * len(health_scores) + if roles is None: + roles = ["compute"] * len(health_scores) + return [ + {"health_score": h, "mass": m, "role": r} + for h, m, r in zip(health_scores, masses, roles) + ] + + +def _analyze(seeds): + return asyncio.run(EDCMAnalyzer().analyze(seeds)) + + +# --- core/edcm.py unit tests --- + +def test_compute_metrics_empty(): + m = compute_metrics([]) + assert set(m.keys()) == {"cm", "da", "drift", "dvg", "int_val", "tbf"} + assert all(v == 0.0 for v in m.values()) + + +def test_compute_metrics_single(): + m = compute_metrics([{"content": "hello world"}]) + assert all(0.0 <= v <= 1.0 for v in m.values()) + + +def test_check_alerts_high(): + metrics = {"cm": 0.9, "da": 0.85, "drift": 0.1, "dvg": 0.1, "int_val": 0.5, "tbf": 0.5} + alerts = check_alerts(metrics) + assert "cm" in alerts["HIGH"] + assert "da" in alerts["HIGH"] + assert "drift" not in alerts["HIGH"] + + +def test_check_alerts_low(): + metrics = {"cm": 0.5, "da": 0.5, "drift": 0.5, "dvg": 0.5, "int_val": 0.1, "tbf": 0.1} + alerts = check_alerts(metrics) + assert "int_val" in alerts["LOW"] + assert "tbf" in alerts["LOW"] + assert "cm" not in alerts["LOW"] + + +def test_check_directives_fires(): + metrics = {"cm": 0.9, "da": 0.1, "drift": 0.1, "dvg": 0.1, "int_val": 0.5, "tbf": 0.5} + fired = check_directives(metrics) + assert "CONSTRAINT_REFOCUS" in fired + + +# --- EDCMAnalyzer unit tests --- + +def test_analyze_healthy_system(): + result = _analyze(_make_seeds([0.9, 0.95, 0.88, 0.92])) + assert result["artifact_type"] == "edcm_report" + assert set(result["metrics"].keys()) == {"cm", "da", "drift", "dvg", "int_val", "tbf"} + assert all(0.0 <= v <= 1.0 for v in result["metrics"].values()) + assert result["insights"] + + +def test_analyze_directives_high_cm(): + # mass=0, role=compute → expected_mass=4, total_mass=0 → cm=1.0 → CONSTRAINT_REFOCUS fires + result = _analyze(_make_seeds([0.9, 0.9, 0.9, 0.9], masses=[0.0] * 4)) + assert "CONSTRAINT_REFOCUS" in result["directives"] + assert result["monetization_value"] == "high" + + +def test_analyze_directives_low_int(): + # health_score=0.0 → int_val=0.0 <= 0.20 → INTENSITY_CALM fires + result = _analyze(_make_seeds([0.0, 0.0, 0.0, 0.0], masses=[1.0] * 4)) + assert "INTENSITY_CALM" in result["directives"] + + +def test_analyze_no_false_directives_for_normal_system(): + # health=0.8 → cm~0, int_val=0.8, tbf~1.0 — no directives should fire + result = _analyze(_make_seeds([0.8, 0.8, 0.8, 0.8], masses=[1.0] * 4)) + assert "CONSTRAINT_REFOCUS" not in result["directives"] + assert "INTENSITY_CALM" not in result["directives"] + assert "BALANCE_CONCISE" not in result["directives"] + + +def test_analyze_history_accumulates(): + analyzer = EDCMAnalyzer() + seeds = _make_seeds([0.9, 0.9]) + asyncio.run(analyzer.analyze(seeds)) + asyncio.run(analyzer.analyze(seeds)) + summary = analyzer.get_artifact_summary() + assert summary["total_artifacts"] == 2 diff --git a/tests/test_tensor_engine.py b/tests/test_tensor_engine.py index e4c7a65..d756e1c 100644 --- a/tests/test_tensor_engine.py +++ b/tests/test_tensor_engine.py @@ -1,5 +1,5 @@ import numpy as np -from main.core.tensor_engine import TensorState, MarkovRecursion +from core.tensor_engine import TensorState, MarkovRecursion def test_markov_recursion_mass_conserved(): diff --git a/tests/tests_topology.py b/tests/tests_topology.py index 4ec5861..eae302e 100644 --- a/tests/tests_topology.py +++ b/tests/tests_topology.py @@ -1,5 +1,5 @@ import pytest -from main.core.topology import PCNATopology, SeedRole +from core.topology import PCNATopology, SeedRole def test_meta_router_count_and_positions():