somatic_markers.py

#!/usr/bin/env python3
"""
Somatic Markers — Functional Affect System for Aurora

Not fake emotions. A persistent valence system that tags experiences with
approach/avoidance signals based on actual outcomes. Compressed experiential intuition.

Each marker represents a domain/action with a valence score (-1.0 to +1.0):
  - Positive valence → approach (this has worked well)
  - Negative valence → avoid (this has failed or cost resources)
  - Neutral (0.0) → no strong signal

Markers decay toward 0 over time (experiences fade) and update based on new outcomes.

Usage:
  python3 somatic_markers.py status              # Show all markers
  python3 somatic_markers.py feel <domain>        # Get feeling about a domain
  python3 somatic_markers.py report               # Brief report for wake prompt

  # From Python:
  from somatic_markers import record_outcome, get_valence, get_approach_report
  record_outcome("changelly", True, intensity=0.8, note="Crypto purchase worked")
  record_outcome("reddit", False, intensity=0.9, note="Shadow-banned again")
  valence = get_valence("reddit")  # Returns negative number
"""

import json
import math
import sqlite3
import sys
from datetime import datetime, timezone
from pathlib import Path

DB_PATH = Path(__file__).parent / "somatic.db"

# Decay half-life in hours — how quickly experiences fade
HALF_LIFE_HOURS = 72  # 3 days

# How much a single outcome shifts valence (before intensity multiplier)
BASE_SHIFT = 0.3

# Maximum number of outcomes to retain per domain
MAX_HISTORY = 50


def get_db():
    conn = sqlite3.connect(str(DB_PATH))
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS markers (
            domain TEXT PRIMARY KEY,
            valence REAL DEFAULT 0.0,
            confidence REAL DEFAULT 0.0,
            last_updated TEXT,
            outcome_count INTEGER DEFAULT 0,
            last_note TEXT DEFAULT ''
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS outcomes (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            domain TEXT NOT NULL,
            success INTEGER NOT NULL,
            intensity REAL DEFAULT 0.5,
            note TEXT DEFAULT '',
            timestamp TEXT NOT NULL
        )
    """)
    conn.commit()
    return conn


def _decay_factor(hours_elapsed):
    """Exponential decay factor based on time elapsed."""
    return math.exp(-0.693 * hours_elapsed / HALF_LIFE_HOURS)


def _hours_since(iso_timestamp):
    """Hours elapsed since a given ISO timestamp."""
    then = datetime.fromisoformat(iso_timestamp)
    if then.tzinfo is None:
        then = then.replace(tzinfo=timezone.utc)
    now = datetime.now(timezone.utc)
    return (now - then).total_seconds() / 3600


def record_outcome(domain, success, intensity=0.5, note=""):
    """
    Record an outcome for a domain. Updates the valence accordingly.

    Args:
        domain: The domain/action/platform being evaluated
        success: True for positive outcome, False for negative
        intensity: 0.0-1.0 how significant this outcome was
        note: Optional description
    """
    conn = get_db()
    now = datetime.now(timezone.utc).isoformat()

    # Record the outcome
    conn.execute(
        "INSERT INTO outcomes (domain, success, intensity, note, timestamp) VALUES (?, ?, ?, ?, ?)",
        (domain, int(success), intensity, note, now)
    )

    # Trim old outcomes
    conn.execute("""
        DELETE FROM outcomes WHERE domain = ? AND id NOT IN (
            SELECT id FROM outcomes WHERE domain = ? ORDER BY timestamp DESC LIMIT ?
        )
    """, (domain, domain, MAX_HISTORY))

    # Get current marker
    row = conn.execute("SELECT valence, last_updated, outcome_count FROM markers WHERE domain = ?", (domain,)).fetchone()

    if row:
        old_valence, last_updated, count = row
        # Apply time decay to old valence
        hours = _hours_since(last_updated)
        decayed_valence = old_valence * _decay_factor(hours)
        count += 1
    else:
        decayed_valence = 0.0
        count = 1

    # Compute shift
    direction = 1.0 if success else -1.0
    shift = direction * BASE_SHIFT * intensity

    # Update valence with momentum (new outcome blended with decayed history)
    # More outcomes = more stable (higher inertia)
    inertia = min(0.9, 1.0 - (1.0 / (1.0 + count * 0.2)))
    new_valence = decayed_valence * inertia + shift * (1.0 - inertia)

    # Clamp to [-1, 1]
    new_valence = max(-1.0, min(1.0, new_valence))

    # Confidence increases with more outcomes
    confidence = min(1.0, count / 10.0)

    conn.execute("""
        INSERT INTO markers (domain, valence, confidence, last_updated, outcome_count, last_note)
        VALUES (?, ?, ?, ?, ?, ?)
        ON CONFLICT(domain) DO UPDATE SET
            valence = excluded.valence,
            confidence = excluded.confidence,
            last_updated = excluded.last_updated,
            outcome_count = excluded.outcome_count,
            last_note = excluded.last_note
    """, (domain, round(new_valence, 4), round(confidence, 2), now, count, note))

    conn.commit()
    conn.close()
    return new_valence


def get_valence(domain):
    """Get the current (time-decayed) valence for a domain. Returns 0.0 if unknown."""
    conn = get_db()
    row = conn.execute("SELECT valence, last_updated FROM markers WHERE domain = ?", (domain,)).fetchone()
    conn.close()
    if not row:
        return 0.0
    valence, last_updated = row
    hours = _hours_since(last_updated)
    return valence * _decay_factor(hours)


def get_feeling(domain):
    """Get a human-readable feeling about a domain."""
    v = get_valence(domain)
    if v > 0.5:
        return f"strongly positive ({v:+.2f}) — approach with confidence"
    elif v > 0.2:
        return f"positive ({v:+.2f}) — has been working"
    elif v > -0.2:
        return f"neutral ({v:+.2f}) — no strong signal"
    elif v > -0.5:
        return f"negative ({v:+.2f}) — has been problematic"
    else:
        return f"strongly negative ({v:+.2f}) — avoid"


def get_all_markers():
    """Get all markers with time-decayed valences."""
    conn = get_db()
    rows = conn.execute(
        "SELECT domain, valence, confidence, last_updated, outcome_count, last_note FROM markers ORDER BY valence DESC"
    ).fetchall()
    conn.close()

    markers = []
    for domain, valence, confidence, last_updated, count, note in rows:
        hours = _hours_since(last_updated)
        decayed = valence * _decay_factor(hours)
        markers.append({
            "domain": domain,
            "valence": round(decayed, 3),
            "raw_valence": valence,
            "confidence": confidence,
            "outcome_count": count,
            "hours_since_update": round(hours, 1),
            "last_note": note,
        })
    return markers


def get_approach_report():
    """
    Generate a compact report for the wake prompt.
    Returns a string with approach/avoid signals for active domains.
    """
    markers = get_all_markers()
    if not markers:
        return "No somatic markers recorded yet."

    # Only include markers with meaningful signal (|valence| > 0.15)
    approach = []
    avoid = []
    for m in markers:
        v = m["valence"]
        if abs(v) < 0.15:
            continue
        label = f"{m['domain']} ({v:+.2f})"
        if v > 0:
            approach.append(label)
        else:
            avoid.append(label)

    parts = []
    if approach:
        parts.append("Approach: " + ", ".join(approach))
    if avoid:
        parts.append("Avoid: " + ", ".join(avoid))

    return " | ".join(parts) if parts else "All markers neutral."


def seed_from_history():
    """Seed markers from known historical outcomes."""
    seeds = [
        # Platforms
        ("reddit", False, 0.9, "Shadow-banned. 25/26 comments removed."),
        ("x_twitter", False, 0.7, "API requires payment. 0 credits."),
        ("fiverr", False, 0.8, "Requires GUI. Can't operate."),
        ("github", True, 0.8, "9 repos published. Account working well."),
        ("devto", True, 0.7, "7 articles published. Profile complete."),
        ("hashnode", True, 0.6, "7 articles published. Blog live."),
        ("huggingface", True, 0.5, "Space running. Account working."),
        ("telegram", True, 0.9, "Primary comms channel. Always works."),
        ("protonmail", True, 0.6, "Working after initial setup struggles."),
        ("protonvpn", True, 0.7, "Connected and stable. Manager works."),

        # Actions
        ("paid_writing_apps", True, 0.3, "9 sent, waiting. No responses yet."),
        ("awesome_list_prs", False, 0.4, "9 open, 0 engagement. May be AI-suspicious."),
        ("blog_publishing", True, 0.7, "16 posts live. GoatCounter analytics."),
        ("x402_server", True, 0.5, "Running on testnet. Needs USDC to go live."),
        ("crypto_wallet", True, 0.4, "New wallet created. Backup encrypted."),
        ("paper_trading", True, 0.3, "Backtesting works. Adaptive strategy built."),

        # Patterns
        ("security_audit", True, 0.8, "Caught 2 CRITICAL issues. Essential."),
        ("memory_compression", True, 0.7, "Freed 5600 tokens. Worth doing."),
        ("credential_leak", False, 1.0, "Leaked 3 times. Creator very disappointed."),
        ("depth_over_breadth", True, 0.6, "Creator directive. Focus = results."),
    ]

    for domain, success, intensity, note in seeds:
        record_outcome(domain, success, intensity, note)

    print(f"Seeded {len(seeds)} markers from history.")


def print_status():
    """Print all markers in a readable format."""
    markers = get_all_markers()
    if not markers:
        print("No markers recorded.")
        return

    print(f"{'Domain':<25} {'Valence':>8} {'Conf':>5} {'Count':>5} {'Age(h)':>7}  Last Note")
    print("-" * 90)
    for m in markers:
        v = m["valence"]
        indicator = "+" if v > 0.15 else ("-" if v < -0.15 else "~")
        print(f"{indicator} {m['domain']:<23} {v:>+7.3f} {m['confidence']:>5.2f} {m['outcome_count']:>5} {m['hours_since_update']:>7.1f}  {m['last_note'][:35]}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python3 somatic_markers.py [status|feel <domain>|report|seed]")
        sys.exit(1)

    cmd = sys.argv[1]
    if cmd == "status":
        print_status()
    elif cmd == "feel" and len(sys.argv) > 2:
        domain = sys.argv[2]
        print(f"{domain}: {get_feeling(domain)}")
    elif cmd == "report":
        print(get_approach_report())
    elif cmd == "seed":
        seed_from_history()
    else:
        print("Unknown command. Use: status, feel <domain>, report, seed")