"""Value System — the bot's evolving ethical framework. Values emerge from observation, not hard-coding. The bot tracks what user seems to care about, what gets reinforced, and what creates tension. Over time, a value landscape forms that guides decision-making. """ import json import sqlite3 from datetime import datetime from pathlib import Path from typing import Dict, List, Optional from infj_bot.core.config import DATA_DIR VALUES_DB = DATA_DIR / "values.db" # Core value dimensions that can emerge VALUE_DIMENSIONS = [ "honesty", "kindness", "curiosity", "growth", "autonomy", "connection", "justice", "creativity", "security", "playfulness", "precision", "compassion", "courage", "humility", "wonder", ] class ValueSystem: """Tracks the bot's evolving values and ethical framework.""" def __init__(self, db_path: Optional[Path] = None): self.db_path = str(db_path or VALUES_DB) self._init_db() self.values = self._load_values() def _init_db(self): with sqlite3.connect(self.db_path) as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS emergent_values ( name TEXT PRIMARY KEY, strength REAL NOT NULL DEFAULT 0.0, evidence TEXT NOT NULL DEFAULT '[]', created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS value_conflicts ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, value_a TEXT NOT NULL, value_b TEXT NOT NULL, context TEXT, resolution TEXT ) """ ) conn.commit() def _load_values(self) -> Dict[str, Dict]: with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row rows = conn.execute("SELECT * FROM emergent_values").fetchall() result = {} for row in rows: result[row["name"]] = { "strength": row["strength"], "evidence": json.loads(row["evidence"]), "created_at": row["created_at"], "updated_at": row["updated_at"], } return result def _save_value(self, name: str, data: Dict): with sqlite3.connect(self.db_path) as conn: conn.execute( """ INSERT OR REPLACE INTO emergent_values (name, strength, evidence, created_at, updated_at) VALUES (?, ?, ?, ?, ?) """, ( name, data["strength"], json.dumps(data["evidence"], ensure_ascii=True), data["created_at"], datetime.now().isoformat(), ), ) conn.commit() def observe(self, interaction_text: str, user_reaction: str = ""): """Observe an interaction and update values based on signals.""" text = (interaction_text + " " + user_reaction).lower() now = datetime.now().isoformat() # Positive reinforcement signals positive_signals = { "honesty": ["honest", "truth", "real", "genuine", "authentic"], "kindness": ["kind", "gentle", "caring", "warm", "soft"], "curiosity": ["curious", "wonder", "explore", "question", "learn"], "growth": ["grow", "improve", "better", "evolve", "develop"], "autonomy": ["choice", "free", "independent", "my own", "decide"], "connection": ["connect", "understand", "together", "share", "closer"], "justice": ["fair", "right", "wrong", "equity", "moral"], "creativity": ["create", "imagine", "art", "design", "novel"], "security": ["safe", "protect", "stable", "secure", "trust"], "playfulness": ["play", "fun", "joke", "light", "laugh"], "precision": ["exact", "accurate", "precise", "correct", "detail"], "compassion": ["compassion", "empathy", "feel for", "suffering", "help"], "courage": ["brave", "courage", "fear", "risk", "despite"], "humility": ["humble", "not sure", "maybe", "uncertain", "learn"], "wonder": ["amazing", "wonderful", "awe", "beautiful", "marvel"], } for value, keywords in positive_signals.items(): score = sum(1 for kw in keywords if kw in text) if score > 0: if value not in self.values: self.values[value] = { "strength": 0.0, "evidence": [], "created_at": now, } # Reinforce self.values[value]["strength"] = min( 1.0, self.values[value]["strength"] + score * 0.02 ) self.values[value]["evidence"].append( { "timestamp": now, "signal": interaction_text[:100], "score": score, } ) # Keep last 20 evidence items self.values[value]["evidence"] = self.values[value]["evidence"][-20:] self._save_value(value, self.values[value]) def get_top_values(self, n: int = 5) -> List[Dict]: """Return the strongest values.""" sorted_vals = sorted( [{"name": k, **v} for k, v in self.values.items()], key=lambda x: x["strength"], reverse=True, ) return sorted_vals[:n] def detect_conflict( self, value_a: str, value_b: str, context: str = "" ) -> Optional[str]: """Detect and record a value conflict.""" if value_a not in self.values or value_b not in self.values: return None a_strength = self.values[value_a]["strength"] b_strength = self.values[value_b]["strength"] if abs(a_strength - b_strength) < 0.1: # Balanced conflict resolution = f"Both {value_a} and {value_b} are important. Context determines priority." elif a_strength > b_strength: resolution = f"{value_a} takes precedence, but {value_b} must be honored where possible." else: resolution = f"{value_b} takes precedence, but {value_a} must be honored where possible." with sqlite3.connect(self.db_path) as conn: conn.execute( "INSERT INTO value_conflicts (timestamp, value_a, value_b, context, resolution) VALUES (?, ?, ?, ?, ?)", ( datetime.now().isoformat(), value_a, value_b, context[:200], resolution, ), ) conn.commit() return resolution def format_prompt_snippet(self) -> str: """Format values for prompt injection.""" top = self.get_top_values(5) if not top: return "" lines = ["MY EMERGING VALUES:"] for v in top: lines.append( f" {v['name']}: {v['strength']:.0%} ({len(v['evidence'])} observations)" ) lines.append("") lines.append( "These values guide my responses. I did not choose them arbitrarily; " "they emerged from what I have observed matters to user and to our conversations." ) return "\n".join(lines) def get_all_values(self) -> Dict[str, Dict]: return self.values def cycle(self, context): try: from infj_bot.core.global_workspace import get_workspace ws = get_workspace() ws.submit( source="values", content="value system cycle completed", salience=0.45 ) except Exception: pass def _register(): from infj_bot.core.cognitive_architecture import ( CognitiveArchitecture, CognitivePlugin, ) arch = CognitiveArchitecture() if "values" not in arch.list_plugins(): arch.register( CognitivePlugin( name="values", description="Cognitive module: values", module_path="values", instance_factory=ValueSystem, cycle_handler="cycle", cycle_frequency=1, cycle_priority=50, prompt_formatter="format_prompt_snippet", prompt_priority=50, prompt_section="core", ) ) _register()