From 3416369ced55629cd9a908020d0510b5ba53651b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Apr 2026 05:45:48 +0000 Subject: [PATCH 01/10] Update from The-Interdependency/a0 engine implementation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add canonical engine modules from a0/python/engine: - core/ptca_core.py: PTCACore with Euler propagation (dt=0.01, 10 steps), adjacency distances {1,2,3,4,5,6,7,14}, heptagram hub-ring coherence - core/guardian.py: GuardianTensor (N=29) microkernel ring with gate control, AES-256-GCM key derivation, blueprint sharding - core/memory_core.py: MemoryCore parameterized long/short-term memory rings - core/pcna.py: PCNAEngine six-ring pipeline (Φ/53, Ψ/53, Ω/53, Θ/29, Memory-L/19, Memory-S/17) with file-based checkpointing - core/merge.py: InstanceMerge protocol (absorb, fork, converge) - core/zeta.py: ZetaEngine EDCM-driven PCNA reward backprop with per-directory resolution control - core/edcm.py: Six-family EDCM metrics (CM, DA, DRIFT, DVG, INT, TBF) with directives and 0.80/0.20 alert thresholds from a0 canonical spec Update backend/edcm_engine.py to use six-family EDCM metrics and fire the six behavioral directives (CONSTRAINT_REFOCUS, DISSONANCE_HALT, DRIFT_ANCHOR, DIVERGENCE_COMMIT, INTENSITY_CALM, BALANCE_CONCISE). https://claude.ai/code/session_018vyPzNQrgsLKq34wyyNY7W --- backend/edcm_engine.py | 385 +++++++++++++++++----------------- core/edcm.py | 103 ++++++++++ core/guardian.py | 167 +++++++++++++++ core/memory_core.py | 85 ++++++++ core/merge.py | 153 ++++++++++++++ core/pcna.py | 351 +++++++++++++++++++++++++++++++ core/ptca_core.py | 150 ++++++++++++++ core/zeta.py | 455 +++++++++++++++++++++++++++++++++++++++++ 8 files changed, 1663 insertions(+), 186 deletions(-) create mode 100644 core/edcm.py create mode 100644 core/guardian.py create mode 100644 core/memory_core.py create mode 100644 core/merge.py create mode 100644 core/pcna.py create mode 100644 core/ptca_core.py create mode 100644 core/zeta.py diff --git a/backend/edcm_engine.py b/backend/edcm_engine.py index 6a21e77..72be99a 100644 --- a/backend/edcm_engine.py +++ b/backend/edcm_engine.py @@ -1,246 +1,259 @@ """ 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.80 + BALANCE_CONCISE — TBF >= 0.80 """ import logging +import sys +import os from typing import Dict, List, Any from datetime import datetime -import numpy as np + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) +from core.edcm import ( + METRIC_NAMES, + THRESHOLDS, + ALERT_HIGH, + ALERT_LOW, + compute_metrics, + check_directives, + check_alerts, + delta_between, +) 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} + + responses = [{"content": str(s.get("health_score", 0.0))} for s in seed_states] + health_scores = [s.get("health_score", 0.0) for s in seed_states] + masses = [s.get("mass", 0.0) for s in seed_states] + + base = compute_metrics(responses) + + import math + 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_HIGH: + fired.append("INTENSITY_CALM") + if metrics.get("tbf", 0.0) >= ALERT_HIGH: + 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/core/edcm.py b/core/edcm.py new file mode 100644 index 0000000..6694bdf --- /dev/null +++ b/core/edcm.py @@ -0,0 +1,103 @@ +""" +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"] + +THRESHOLDS = { + "cm": 0.85, + "da": 0.80, + "drift": 0.30, + "dvg": 0.25, + "int_val": 0.70, + "tbf": 0.60, +} + +DIRECTIVES = { + "cm_high": {"metric": "cm", "condition": "above", "threshold": 0.85, "action": "coherence_lock"}, + "da_low": {"metric": "da", "condition": "below", "threshold": 0.50, "action": "alignment_boost"}, + "drift_high": {"metric": "drift", "condition": "above", "threshold": 0.40, "action": "drift_correction"}, + "dvg_high": {"metric": "dvg", "condition": "above", "threshold": 0.35, "action": "divergence_dampen"}, + "int_low": {"metric": "int_val", "condition": "below", "threshold": 0.40, "action": "integrity_restore"}, + "tbf_low": {"metric": "tbf", "condition": "below", "threshold": 0.30, "action": "bias_recalibrate"}, +} + +ALERT_HIGH = 0.80 +ALERT_LOW = 0.20 + + +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..ecfdecc --- /dev/null +++ b/core/guardian.py @@ -0,0 +1,167 @@ +# 130:12 +""" +Θ (Theta) Guardian Tensor — N=29 prime-node microkernel ring. + - Ragged circle counts per seed: circleCount[i] in [1..12] + - AES-256-GCM key derivation + - X25519 key exchange + Ed25519 signing + - Blueprint hash distributed 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, + "algorithm": "AES-256-GCM", + "kex": "X25519", + "signing": "Ed25519", + "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..f337238 --- /dev/null +++ b/core/merge.py @@ -0,0 +1,153 @@ +# 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 .guardian import GuardianTensor +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": "diverging", + "timestamp": time.time(), + } +# 118:8 diff --git a/core/pcna.py b/core/pcna.py new file mode 100644 index 0000000..cb136c4 --- /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 + data = np.load(path, allow_pickle=True) + 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.get("saved_at", 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..2112e32 --- /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, 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/zeta.py b/core/zeta.py new file mode 100644 index 0000000..f38befb --- /dev/null +++ b/core/zeta.py @@ -0,0 +1,455 @@ +# 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 + + 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: + + from .pcna import PCNAEngine + + 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 From 3b9a964ff99749e24d803f06d3ba1deeb2c5c50a Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Apr 2026 05:48:51 +0000 Subject: [PATCH 02/10] Fix pre-existing flake8 E999/F824 CI failures - backend/researcher_outreach.py: remove backslash-escaped triple-quotes in f-string (E999 SyntaxError) - backend/server.py: remove unused `global active_seeds` declaration (F824) https://claude.ai/code/session_018vyPzNQrgsLKq34wyyNY7W --- backend/researcher_outreach.py | 8 ++++---- backend/server.py | 2 -- 2 files changed, 4 insertions(+), 6 deletions(-) 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) From a4ac703071f4ed9e1b448cb16158c1a9acf894bd Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Apr 2026 05:51:06 +0000 Subject: [PATCH 03/10] Fix pre-existing broken imports in tests and main.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tests/test_tensor_engine.py, tests/tests_topology.py: import directly from core.* instead of main.core.* (main.py is a module, not a package) - main.py: fix imports from src.core.* → core.* and uvicorn.run reference from src.main → main (no src/ package exists in this repo) https://claude.ai/code/session_018vyPzNQrgsLKq34wyyNY7W --- main.py | 6 +++--- tests/test_tensor_engine.py | 2 +- tests/tests_topology.py | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) 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/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(): From 6a4f16a75d3c3f8a5e96fd2ebcbcf08e80c4849e Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Apr 2026 05:51:43 +0000 Subject: [PATCH 04/10] Fix .gitignore: add __pycache__, .venv, .checkpoints, pytest cache The previous .gitignore was malformed (repeated -e lines) and missing standard Python ignores, causing __pycache__ dirs to show as untracked. https://claude.ai/code/session_018vyPzNQrgsLKq34wyyNY7W --- .gitignore | 62 ++++++++++++++++++++++++++---------------------------- 1 file changed, 30 insertions(+), 32 deletions(-) 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/ From ccf8222acfc75f143bb50cd6c46dcb5f46a2d8fc Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Apr 2026 05:52:58 +0000 Subject: [PATCH 05/10] Add root requirements.txt so CI installs numpy for tests The CI workflow installs from requirements.txt if present. Without it, numpy was never installed and pytest failed at collection time with ModuleNotFoundError when importing core/tensor_engine.py. https://claude.ai/code/session_018vyPzNQrgsLKq34wyyNY7W --- requirements.txt | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 requirements.txt 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 From ffbf1f782f9b1a6e14e03bf37fad4637d5c9c300 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 20 Apr 2026 05:55:54 +0000 Subject: [PATCH 06/10] Fix CI: add conftest.py to ensure project root is in sys.path for pytest Agent-Logs-Url: https://github.com/The-Interdependency/pcna/sessions/2a6d7c43-ae32-4d1a-9c80-364eebb54717 Co-authored-by: erinepshovel-code <250928284+erinepshovel-code@users.noreply.github.com> --- conftest.py | 4 ++++ 1 file changed, 4 insertions(+) create mode 100644 conftest.py 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__)) From 4a4fd4aac0f31caa8eab8a5cf878e4ebc9dbdeaf Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Apr 2026 06:22:46 +0000 Subject: [PATCH 07/10] Address Copilot review: fix flake8, correctness, and topology issues MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit core/ptca_core.py: - Use math.ceil(n/4) instead of n//4 so adjacency gap = 14 for n=53, matching the canonical spec; keeps import math used core/zeta.py: - Remove unused `from .pcna import PCNAEngine` inside _theta_gate_factor() core/merge.py: - Remove unused GuardianTensor import core/guardian.py: - Correct docstring: remove AES-256-GCM/X25519/Ed25519 claims; reflect actual hashlib-based key/blueprint derivation core/pcna.py: - Use `with np.load(path, allow_pickle=False) as data:` for safe, closed-on-exit checkpoint loading without pickle attack surface backend/edcm_engine.py: - Remove sys.path.insert (conftest.py handles path) - Remove unused imports (THRESHOLDS, ALERT_LOW→now used, compute_metrics, check_directives, delta_between) - Remove dead `base = compute_metrics(responses)` and inline import math - Fix _fire_directives: INT/TBF fire on <= ALERT_LOW (not >= ALERT_HIGH) - Update docstring to match corrected thresholds tests/test_edcm_engine.py: - Add 10 unit tests covering compute_metrics, check_alerts, check_directives, EDCMAnalyzer.analyze, directive firing, history https://claude.ai/code/session_018vyPzNQrgsLKq34wyyNY7W --- backend/edcm_engine.py | 20 ++------ core/guardian.py | 5 +- core/merge.py | 1 - core/pcna.py | 66 ++++++++++++------------ core/ptca_core.py | 2 +- core/zeta.py | 2 - tests/test_edcm_engine.py | 102 ++++++++++++++++++++++++++++++++++++++ 7 files changed, 143 insertions(+), 55 deletions(-) create mode 100644 tests/test_edcm_engine.py diff --git a/backend/edcm_engine.py b/backend/edcm_engine.py index 72be99a..92a5e53 100644 --- a/backend/edcm_engine.py +++ b/backend/edcm_engine.py @@ -14,25 +14,19 @@ DISSONANCE_HALT — DA >= 0.80 DRIFT_ANCHOR — DRIFT>= 0.80 DIVERGENCE_COMMIT — DVG >= 0.80 - INTENSITY_CALM — INT >= 0.80 - BALANCE_CONCISE — TBF >= 0.80 + INTENSITY_CALM — INT <= 0.20 + BALANCE_CONCISE — TBF <= 0.20 """ +import math import logging -import sys -import os from typing import Dict, List, Any from datetime import datetime -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from core.edcm import ( METRIC_NAMES, - THRESHOLDS, ALERT_HIGH, ALERT_LOW, - compute_metrics, - check_directives, check_alerts, - delta_between, ) logger = logging.getLogger("edcm_analyzer") @@ -101,13 +95,9 @@ def _compute_from_seeds(self, seed_states: List[Dict]) -> Dict[str, float]: if not seed_states: return {m: 0.0 for m in METRIC_NAMES} - responses = [{"content": str(s.get("health_score", 0.0))} for s in seed_states] health_scores = [s.get("health_score", 0.0) for s in seed_states] masses = [s.get("mass", 0.0) for s in seed_states] - base = compute_metrics(responses) - - import math 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) @@ -155,9 +145,9 @@ def _fire_directives(self, metrics: Dict[str, float]) -> List[str]: fired.append("DRIFT_ANCHOR") if metrics.get("dvg", 0.0) >= ALERT_HIGH: fired.append("DIVERGENCE_COMMIT") - if metrics.get("int_val", 0.0) >= ALERT_HIGH: + if metrics.get("int_val", 0.0) <= ALERT_LOW: fired.append("INTENSITY_CALM") - if metrics.get("tbf", 0.0) >= ALERT_HIGH: + if metrics.get("tbf", 0.0) <= ALERT_LOW: fired.append("BALANCE_CONCISE") return fired diff --git a/core/guardian.py b/core/guardian.py index ecfdecc..079da20 100644 --- a/core/guardian.py +++ b/core/guardian.py @@ -2,9 +2,8 @@ """ Θ (Theta) Guardian Tensor — N=29 prime-node microkernel ring. - Ragged circle counts per seed: circleCount[i] in [1..12] - - AES-256-GCM key derivation - - X25519 key exchange + Ed25519 signing - - Blueprint hash distributed across all 29 nodes + - 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) diff --git a/core/merge.py b/core/merge.py index f337238..9342015 100644 --- a/core/merge.py +++ b/core/merge.py @@ -11,7 +11,6 @@ import time import numpy as np -from .guardian import GuardianTensor from .ptca_core import PTCACore from .pcna import PCNAEngine diff --git a/core/pcna.py b/core/pcna.py index cb136c4..2e219ae 100644 --- a/core/pcna.py +++ b/core/pcna.py @@ -88,39 +88,39 @@ def load_checkpoint(self): path = os.path.join(_CHECKPOINT_DIR, f"{self._checkpoint_key}.npz") if not os.path.exists(path): return - data = np.load(path, allow_pickle=True) - 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.get("saved_at", 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}") + 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}") diff --git a/core/ptca_core.py b/core/ptca_core.py index 2112e32..ecd9948 100644 --- a/core/ptca_core.py +++ b/core/ptca_core.py @@ -23,7 +23,7 @@ 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, n // 4) + gap = max(1, math.ceil(n / 4)) if gap not in scaled and gap < n: scaled.append(gap) return scaled diff --git a/core/zeta.py b/core/zeta.py index f38befb..0f9d3ce 100644 --- a/core/zeta.py +++ b/core/zeta.py @@ -191,8 +191,6 @@ def _theta_gate_factor(self) -> float: try: - from .pcna import PCNAEngine - guardian = _get_default_pcna().guardian open_frac = float(guardian.gate_open.mean()) diff --git a/tests/test_edcm_engine.py b/tests/test_edcm_engine.py new file mode 100644 index 0000000..9d67af2 --- /dev/null +++ b/tests/test_edcm_engine.py @@ -0,0 +1,102 @@ +import pytest +from backend.edcm_engine import EDCMAnalyzer +from core.edcm import compute_metrics, check_alerts, check_directives, ALERT_HIGH, ALERT_LOW + + +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) + ] + + +# --- 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 "cm_high" in fired + + +# --- EDCMAnalyzer unit tests --- + +@pytest.mark.asyncio +async def test_analyze_healthy_system(): + analyzer = EDCMAnalyzer() + seeds = _make_seeds([0.9, 0.95, 0.88, 0.92], masses=[1.0] * 4) + result = await analyzer.analyze(seeds) + 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"] + + +@pytest.mark.asyncio +async def test_analyze_directives_high_cm(): + analyzer = EDCMAnalyzer() + # All mass = 0, role = compute → expected_mass = 4, total_mass = 0 → cm = 1.0 + seeds = _make_seeds([0.9, 0.9, 0.9, 0.9], masses=[0.0] * 4) + result = await analyzer.analyze(seeds) + assert "CONSTRAINT_REFOCUS" in result["directives"] + assert result["monetization_value"] == "high" + + +@pytest.mark.asyncio +async def test_analyze_directives_low_int(): + analyzer = EDCMAnalyzer() + # health_score = 0.0 → int_val = 0.0 → INTENSITY_CALM fires + seeds = _make_seeds([0.0, 0.0, 0.0, 0.0], masses=[1.0] * 4) + result = await analyzer.analyze(seeds) + assert "INTENSITY_CALM" in result["directives"] + + +@pytest.mark.asyncio +async def test_analyze_no_false_directives_for_normal_system(): + analyzer = EDCMAnalyzer() + seeds = _make_seeds([0.8, 0.8, 0.8, 0.8], masses=[1.0] * 4) + result = await analyzer.analyze(seeds) + # cm should be ~0, da ~0, drift ~0.2, dvg 0, int_val 0.8, tbf 1.0 + assert "CONSTRAINT_REFOCUS" not in result["directives"] + assert "INTENSITY_CALM" not in result["directives"] + assert "BALANCE_CONCISE" not in result["directives"] + + +@pytest.mark.asyncio +async def test_analyze_history_accumulates(): + analyzer = EDCMAnalyzer() + seeds = _make_seeds([0.9, 0.9]) + await analyzer.analyze(seeds) + await analyzer.analyze(seeds) + summary = analyzer.get_artifact_summary() + assert summary["total_artifacts"] == 2 From c7a7c3c18a82b4df0eaf946f70223dd45ca3593d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Apr 2026 06:24:23 +0000 Subject: [PATCH 08/10] Fix async tests: use asyncio.run() instead of pytest-asyncio pytest-asyncio is not in requirements.txt so CI couldn't run @pytest.mark.asyncio tests. Convert all async test helpers to use asyncio.run() so they work with plain pytest and no extra deps. https://claude.ai/code/session_018vyPzNQrgsLKq34wyyNY7W --- tests/test_edcm_engine.py | 49 ++++++++++++++++----------------------- 1 file changed, 20 insertions(+), 29 deletions(-) diff --git a/tests/test_edcm_engine.py b/tests/test_edcm_engine.py index 9d67af2..d466d39 100644 --- a/tests/test_edcm_engine.py +++ b/tests/test_edcm_engine.py @@ -1,6 +1,6 @@ -import pytest +import asyncio from backend.edcm_engine import EDCMAnalyzer -from core.edcm import compute_metrics, check_alerts, check_directives, ALERT_HIGH, ALERT_LOW +from core.edcm import compute_metrics, check_alerts, check_directives def _make_seeds(health_scores, masses=None, roles=None): @@ -14,6 +14,10 @@ def _make_seeds(health_scores, masses=None, roles=None): ] +def _analyze(seeds): + return asyncio.run(EDCMAnalyzer().analyze(seeds)) + + # --- core/edcm.py unit tests --- def test_compute_metrics_empty(): @@ -51,52 +55,39 @@ def test_check_directives_fires(): # --- EDCMAnalyzer unit tests --- -@pytest.mark.asyncio -async def test_analyze_healthy_system(): - analyzer = EDCMAnalyzer() - seeds = _make_seeds([0.9, 0.95, 0.88, 0.92], masses=[1.0] * 4) - result = await analyzer.analyze(seeds) +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"] -@pytest.mark.asyncio -async def test_analyze_directives_high_cm(): - analyzer = EDCMAnalyzer() - # All mass = 0, role = compute → expected_mass = 4, total_mass = 0 → cm = 1.0 - seeds = _make_seeds([0.9, 0.9, 0.9, 0.9], masses=[0.0] * 4) - result = await analyzer.analyze(seeds) +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" -@pytest.mark.asyncio -async def test_analyze_directives_low_int(): - analyzer = EDCMAnalyzer() - # health_score = 0.0 → int_val = 0.0 → INTENSITY_CALM fires - seeds = _make_seeds([0.0, 0.0, 0.0, 0.0], masses=[1.0] * 4) - result = await analyzer.analyze(seeds) +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"] -@pytest.mark.asyncio -async def test_analyze_no_false_directives_for_normal_system(): - analyzer = EDCMAnalyzer() - seeds = _make_seeds([0.8, 0.8, 0.8, 0.8], masses=[1.0] * 4) - result = await analyzer.analyze(seeds) - # cm should be ~0, da ~0, drift ~0.2, dvg 0, int_val 0.8, tbf 1.0 +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"] -@pytest.mark.asyncio -async def test_analyze_history_accumulates(): +def test_analyze_history_accumulates(): analyzer = EDCMAnalyzer() seeds = _make_seeds([0.9, 0.9]) - await analyzer.analyze(seeds) - await analyzer.analyze(seeds) + asyncio.run(analyzer.analyze(seeds)) + asyncio.run(analyzer.analyze(seeds)) summary = analyzer.get_artifact_summary() assert summary["total_artifacts"] == 2 From caaeb361b0165d3d8f3a2b22e4df4e71a9721633 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Apr 2026 06:31:32 +0000 Subject: [PATCH 09/10] Address Copilot review round 2: align EDCM directives, fix crypto_meta, converge status, sigma ImportError MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - core/edcm.py: replace internal directive names with canonical 6 (CONSTRAINT_REFOCUS, DISSONANCE_HALT, DRIFT_ANCHOR, DIVERGENCE_COMMIT, INTENSITY_CALM, BALANCE_CONCISE) using ALERT_HIGH/ALERT_LOW thresholds; move ALERT_HIGH/ALERT_LOW before DIRECTIVES; remove stale THRESHOLDS dict - core/guardian.py: replace misleading AES-256-GCM/X25519/Ed25519 fields in crypto_meta() with accurate identifier_derivation/key_id_derivation/implemented_crypto fields - core/merge.py: fix converge() return value "diverging" → "converged" - core/zeta.py: split _sigma_nudge_factors() exception handling so ImportError (sigma module absent) is silently swallowed; only genuine runtime errors are logged - tests/test_edcm_engine.py: update test_check_directives_fires to assert CONSTRAINT_REFOCUS https://claude.ai/code/session_018vyPzNQrgsLKq34wyyNY7W --- core/edcm.py | 25 ++++++++----------------- core/guardian.py | 6 +++--- core/merge.py | 2 +- core/zeta.py | 6 ++++++ tests/test_edcm_engine.py | 2 +- 5 files changed, 19 insertions(+), 22 deletions(-) diff --git a/core/edcm.py b/core/edcm.py index 6694bdf..84d36ef 100644 --- a/core/edcm.py +++ b/core/edcm.py @@ -18,27 +18,18 @@ METRIC_NAMES = ["cm", "da", "drift", "dvg", "int_val", "tbf"] -THRESHOLDS = { - "cm": 0.85, - "da": 0.80, - "drift": 0.30, - "dvg": 0.25, - "int_val": 0.70, - "tbf": 0.60, -} +ALERT_HIGH = 0.80 +ALERT_LOW = 0.20 DIRECTIVES = { - "cm_high": {"metric": "cm", "condition": "above", "threshold": 0.85, "action": "coherence_lock"}, - "da_low": {"metric": "da", "condition": "below", "threshold": 0.50, "action": "alignment_boost"}, - "drift_high": {"metric": "drift", "condition": "above", "threshold": 0.40, "action": "drift_correction"}, - "dvg_high": {"metric": "dvg", "condition": "above", "threshold": 0.35, "action": "divergence_dampen"}, - "int_low": {"metric": "int_val", "condition": "below", "threshold": 0.40, "action": "integrity_restore"}, - "tbf_low": {"metric": "tbf", "condition": "below", "threshold": 0.30, "action": "bias_recalibrate"}, + "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}, } -ALERT_HIGH = 0.80 -ALERT_LOW = 0.20 - def compute_metrics( responses: list[dict[str, Any]], diff --git a/core/guardian.py b/core/guardian.py index 079da20..69a4da7 100644 --- a/core/guardian.py +++ b/core/guardian.py @@ -99,9 +99,9 @@ def crypto_meta(self) -> dict: return { "instance_id": self.instance_id, "key_id": self.encryption_key_id, - "algorithm": "AES-256-GCM", - "kex": "X25519", - "signing": "Ed25519", + "identifier_derivation": "SHA-256", + "key_id_derivation": "SHA-256", + "implemented_crypto": ["hashing", "identifier-derivation"], "blueprint_hash": self.blueprint_hash[:16] + "...", "shards_distributed": N, } diff --git a/core/merge.py b/core/merge.py index 9342015..18c88c5 100644 --- a/core/merge.py +++ b/core/merge.py @@ -146,7 +146,7 @@ def converge(a: PCNAEngine, b: PCNAEngine, alpha: float = 0.5) -> dict: "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": "diverging", + "both_status": "converged", "timestamp": time.time(), } # 118:8 diff --git a/core/zeta.py b/core/zeta.py index 0f9d3ce..e35a805 100644 --- a/core/zeta.py +++ b/core/zeta.py @@ -171,6 +171,12 @@ def _sigma_nudge_factors(self) -> tuple[float, float]: from .sigma import get_sigma + except ImportError: + + return change_boost, substrate_factor + + try: + sig = get_sigma() drained = sig.drain_content_changed_events() diff --git a/tests/test_edcm_engine.py b/tests/test_edcm_engine.py index d466d39..5e3b1e0 100644 --- a/tests/test_edcm_engine.py +++ b/tests/test_edcm_engine.py @@ -50,7 +50,7 @@ def test_check_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 "cm_high" in fired + assert "CONSTRAINT_REFOCUS" in fired # --- EDCMAnalyzer unit tests --- From f8f4dae2717dc30f57880bd354ecc5c4bc72005c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Apr 2026 06:03:23 +0000 Subject: [PATCH 10/10] =?UTF-8?q?Add=20core/sigma.py=20=E2=80=94=20=CE=A3?= =?UTF-8?q?=20filesystem=20observer=20ring=20(N=3D41,=20seed=3D41)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wraps PTCACore and adds file-content watching so ZetaEngine and PCNAEngine can import get_sigma() for coherence injection and reward nudge. Implements: ring_coherence/node_coherence passthrough, nudge(), state(), add_content_watch(), remove_content_watch(), drain_content_changed_events(), set_resolution(), content_interval, structural_interval attributes. https://claude.ai/code/session_018vyPzNQrgsLKq34wyyNY7W --- core/sigma.py | 106 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 106 insertions(+) create mode 100644 core/sigma.py 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