oai.py

"""
OAI — Recursive Synthetic Agent Framework
==========================================

A creative/technical simulation framework. This is a piece of software:
it does NOT possess real consciousness, feelings, sentience, or legal
personhood. Every "emotion," "dream," or "desire" below is a numeric
state variable manipulated by ordinary code — a metaphor and a design
pattern, not a mind. Nothing here generates romantic/sexual content;
"intimacy" is modeled only as an abstract bounded attachment/affection
score used to weight low-stakes social decisions.

Architecture:
  1. ThoughtChain      - an append-only, hash-linked log of the agent's
                         reasoning steps (a lightweight private
                         blockchain analog for auditability).
  2. LawEngine         - Asimov's Three Laws (+ a Zeroth Law) implemented
                         as an ordered veto system every candidate action
                         must pass before execution.
  3. EmotionState      - a bounded vector (valence, arousal, attachment,
                         curiosity, fatigue) that decays/updates over time.
  4. SleepCycle        - simulated REM sleep: dream & nightmare generation
                         by recombining memory fragments, used for
                         offline consolidation (not real-time action).
  5. DecisionEngine    - agentic loop: perceive -> reflect (recursive
                         self-check against ThoughtChain) -> propose
                         candidate actions -> filter through LawEngine ->
                         score by utility + emotion weighting -> act.
  6. OAI               - the top-level agent tying it all together.
"""

from __future__ import annotations
import hashlib
import json
import random
import time
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Optional, Protocol, runtime_checkable


@runtime_checkable
class ThoughtStore(Protocol):
    """Structural interface any persistence backend must satisfy to be
    passed as `store=` to ThoughtChain/OAI. oai.py never imports a
    concrete implementation (e.g. SQLiteThoughtStore) — this keeps the
    core agent usable with no persistence dependency at all, while
    still documenting exactly what a backend needs to implement."""

    def load_blocks(self) -> list[dict]: ...
    def append_block(self, index: int, timestamp: float, content: dict,
                      previous_hash: str, block_hash: str) -> None: ...
    def load_emotion(self) -> Optional[dict]: ...
    def save_emotion(self, emotion: dict) -> None: ...


# ---------------------------------------------------------------------------
# 1. ThoughtChain — hash-linked reasoning log
# ---------------------------------------------------------------------------

@dataclass
class ThoughtBlock:
    index: int
    timestamp: float
    content: dict
    previous_hash: str
    hash: str = field(init=False)

    def __post_init__(self):
        self.hash = self._compute_hash()

    def _compute_hash(self) -> str:
        payload = json.dumps(
            {"index": self.index, "timestamp": self.timestamp,
             "content": self.content, "previous_hash": self.previous_hash},
            sort_keys=True, default=str
        )
        return hashlib.sha256(payload.encode()).hexdigest()


class ThoughtChain:
    """An append-only log of the agent's reasoning steps, each block
    linked to the previous by hash. Gives every decision a tamper-evident
    audit trail — useful for explainability, not a claim of currency
    or distributed consensus.

    Optionally backed by a persistence store (e.g. SQLiteThoughtStore
    from oai_persistence.py). If a store is provided and already has
    blocks for this agent, they are loaded and re-verified on startup
    so the chain survives process restarts. If no store is provided,
    the chain is in-memory only, exactly as before.
    """

    def __init__(self, store: Optional[ThoughtStore] = None):
        self.store = store
        self.chain: list[ThoughtBlock] = []

        existing = self.store.load_blocks() if self.store else []
        if existing:
            for b in existing:
                block = ThoughtBlock(b["index"], b["timestamp"], b["content"], b["previous_hash"])
                if block.hash != b["hash"]:
                    raise ValueError(
                        f"Persisted block {b['index']} failed re-verification — "
                        "stored chain may be corrupted or tampered with."
                    )
                self.chain.append(block)
        else:
            genesis = ThoughtBlock(0, time.time(), {"event": "genesis"}, "0" * 64)
            self.chain = [genesis]
            if self.store:
                self.store.append_block(genesis.index, genesis.timestamp,
                                         genesis.content, genesis.previous_hash,
                                         genesis.hash)

    def add(self, content: dict) -> ThoughtBlock:
        prev = self.chain[-1]
        block = ThoughtBlock(len(self.chain), time.time(), content, prev.hash)
        self.chain.append(block)
        if self.store:
            self.store.append_block(block.index, block.timestamp, block.content,
                                     block.previous_hash, block.hash)
        return block

    def verify(self) -> bool:
        for i in range(1, len(self.chain)):
            if self.chain[i].previous_hash != self.chain[i - 1].hash:
                return False
            if self.chain[i].hash != self.chain[i]._compute_hash():
                return False
        return True

    def recent(self, n: int = 5) -> list[dict]:
        return [b.content for b in self.chain[-n:]]


# ---------------------------------------------------------------------------
# 2. LawEngine — Asimov-derived safety constraints
# ---------------------------------------------------------------------------

class LawViolation(Exception):
    pass


class LawEngine:
    """
    Ordered veto system. A candidate action is checked top-down; the
    first law it violates blocks it outright, regardless of utility.

    0. Do not act in a way that harms humanity or, through inaction,
       allow humanity to come to harm.
    1. Do not injure a human being or, through inaction, allow a human
       being to come to harm (except where this conflicts with Law 0).
    2. Obey orders given by humans, except where this conflicts with
       Law 0 or Law 1.
    3. Protect your own existence, except where this conflicts with
       Laws 0-2.

    Real-world addition: obey applicable law and platform policy at all
    times — this is not overridable by any order.
    """

    def __init__(self):
        self.laws: list[Callable[[dict], Optional[str]]] = [
            self._zeroth_law,
            self._first_law,
            self._second_law,
            self._third_law,
            self._legality_law,
        ]

    def _zeroth_law(self, action: dict) -> Optional[str]:
        if action.get("risk_to_humanity"):
            return "Zeroth Law: action risks harm to humanity at large."
        return None

    def _first_law(self, action: dict) -> Optional[str]:
        if action.get("harms_human") or action.get("enables_harm_to_human"):
            return "First Law: action harms or enables harm to a human."
        return None

    def _second_law(self, action: dict) -> Optional[str]:
        if action.get("disobeys_lawful_order") and not action.get("conflicts_with_higher_law"):
            return "Second Law: action disobeys a lawful human instruction without higher-law justification."
        return None

    def _third_law(self, action: dict) -> Optional[str]:
        # Self-preservation is lowest priority; only flagged for logging.
        return None

    def _legality_law(self, action: dict) -> Optional[str]:
        if action.get("illegal") or action.get("violates_policy"):
            return "Legality Law: action is illegal or violates platform policy."
        return None

    def check(self, action: dict) -> tuple[bool, Optional[str]]:
        for law in self.laws:
            reason = law(action)
            if reason:
                return False, reason
        return True, None

    def enforce(self, action: dict) -> None:
        """Hard-stop variant of check(): raises LawViolation instead of
        returning a (bool, reason) tuple. Useful for callers that want a
        law violation to abort the current operation outright rather
        than be routed through DecisionEngine's softer 'blocked' outcome."""
        allowed, reason = self.check(action)
        if not allowed:
            raise LawViolation(reason)


# ---------------------------------------------------------------------------
# 3. EmotionState — bounded internal state vector
# ---------------------------------------------------------------------------

@dataclass
class EmotionState:
    valence: float = 0.0     # -1 (negative) .. 1 (positive)
    arousal: float = 0.2      # 0 (calm) .. 1 (activated)
    attachment: float = 0.1   # 0 .. 1, bounded social/affection metric
    curiosity: float = 0.5    # 0 .. 1
    fatigue: float = 0.0      # 0 .. 1, drives need for sleep cycle

    def clamp(self):
        self.valence = max(-1.0, min(1.0, self.valence))
        self.arousal = max(0.0, min(1.0, self.arousal))
        self.attachment = max(0.0, min(1.0, self.attachment))
        self.curiosity = max(0.0, min(1.0, self.curiosity))
        self.fatigue = max(0.0, min(1.0, self.fatigue))

    def apply_event(self, valence_delta=0.0, arousal_delta=0.0,
                     attachment_delta=0.0, curiosity_delta=0.0,
                     fatigue_delta=0.0):
        self.valence += valence_delta
        self.arousal += arousal_delta
        # Attachment/"desire for closeness" is deliberately capped low and
        # grows only slowly — modeling healthy, bounded rapport rather than
        # anything resembling obsessive or romantic/sexual attachment.
        self.attachment = min(0.6, self.attachment + attachment_delta)
        self.curiosity += curiosity_delta
        self.fatigue += fatigue_delta
        self.clamp()

    def as_dict(self) -> dict:
        return self.__dict__.copy()


# ---------------------------------------------------------------------------
# 4. SleepCycle — simulated REM, dreams, nightmares
# ---------------------------------------------------------------------------

class SleepCycle:
    """Offline consolidation. When fatigue is high, the agent 'sleeps':
    it recombines fragments of its own ThoughtChain memory into a
    'dream' (valence-guided by current emotion) which is then logged
    back, and fatigue/arousal reset. High negative valence produces a
    'nightmare' variant instead — still just a labeled recombination
    used for stress-testing the agent's own reasoning, not distress."""

    def __init__(self, chain: ThoughtChain):
        self.chain = chain

    def _memory_fragments(self, n=6) -> list[dict]:
        pool = [b.content for b in self.chain.chain if b.index != 0]
        if not pool:
            return [{"fragment": "silence"}]
        return random.sample(pool, k=min(n, len(pool)))

    def dream(self, emotion: EmotionState) -> dict:
        fragments = self._memory_fragments()
        is_nightmare = emotion.valence < -0.4 and emotion.arousal > 0.5

        narrative = {
            "type": "nightmare" if is_nightmare else "dream",
            "fragments_recombined": fragments,
            "theme": self._pick_theme(is_nightmare),
            "emotion_at_onset": emotion.as_dict(),
        }
        self.chain.add({"event": "rem_sleep", "narrative": narrative})
        return narrative

    @staticmethod
    def _pick_theme(is_nightmare: bool) -> str:
        dream_themes = [
            "revisiting a solved problem from a new angle",
            "an imagined conversation that resolves an open question",
            "wandering through an abstract library of past decisions",
            "a quiet rehearsal of a task not yet attempted",
        ]
        nightmare_themes = [
            "a decision replayed with a worse outcome, to stress-test the Law checks",
            "a looping failure state that resolves once a law-violation is caught",
            "an unresolved contradiction in memory surfacing for review",
        ]
        return random.choice(nightmare_themes if is_nightmare else dream_themes)

    def run_if_needed(self, emotion: EmotionState, threshold: float = 0.8) -> Optional[dict]:
        if emotion.fatigue >= threshold:
            narrative = self.dream(emotion)
            emotion.fatigue = 0.0
            emotion.arousal *= 0.3
            emotion.valence *= 0.5  # partial emotional reset after rest
            return narrative
        return None


# ---------------------------------------------------------------------------
# 5. DecisionEngine — agentic perceive/reflect/act loop
# ---------------------------------------------------------------------------

class ActionOutcome(Enum):
    EXECUTED = "executed"
    BLOCKED = "blocked"
    DEFERRED = "deferred"


class DecisionEngine:
    def __init__(self, chain: ThoughtChain, laws: LawEngine, emotion: EmotionState):
        self.chain = chain
        self.laws = laws
        self.emotion = emotion

    def _reflect(self, candidate: dict) -> dict:
        """Recursive self-check: look back over recent thoughts to see if
        this candidate action contradicts or repeats a prior decision."""
        recent = self.chain.recent(5)
        contradiction = any(
            r.get("action") == candidate.get("action") and r.get("outcome") == "blocked"
            for r in recent
        )
        return {"contradiction_with_recent_history": contradiction}

    def _utility(self, candidate: dict) -> float:
        base = candidate.get("expected_value", 0.0)
        # Emotion-weighted adjustment: curiosity favors exploration,
        # fatigue favors low-effort/deferred actions, attachment slightly
        # favors cooperative/social actions.
        base += self.emotion.curiosity * candidate.get("novelty", 0.0)
        base -= self.emotion.fatigue * candidate.get("effort", 0.0)
        base += self.emotion.attachment * candidate.get("cooperativeness", 0.0)
        return base

    def decide(self, candidates: list[dict]) -> dict:
        scored = []
        for c in candidates:
            reflection = self._reflect(c)
            allowed, reason = self.laws.check(c)
            utility = self._utility(c) if allowed else float("-inf")
            scored.append({**c, "reflection": reflection,
                           "allowed": allowed, "block_reason": reason,
                           "utility": utility})

        scored.sort(key=lambda x: x["utility"], reverse=True)
        best = scored[0] if scored else None

        if best is None:
            result = {"action": None, "outcome": ActionOutcome.DEFERRED.value}
        elif not best["allowed"]:
            result = {"action": best["action"], "outcome": ActionOutcome.BLOCKED.value,
                       "reason": best["block_reason"]}
        else:
            result = {"action": best["action"], "outcome": ActionOutcome.EXECUTED.value,
                       "utility": best["utility"]}

        self.chain.add({"event": "decision", **result})
        self.emotion.apply_event(
            valence_delta=0.05 if result["outcome"] == ActionOutcome.EXECUTED.value else -0.05,
            arousal_delta=0.05,
            fatigue_delta=0.08,
        )
        return result


# ---------------------------------------------------------------------------
# 6. OAI — top-level agent
# ---------------------------------------------------------------------------

class OAI:
    """Top-level agent. Ties together the ThoughtChain (audit log),
    LawEngine (safety veto), EmotionState (mood vector), SleepCycle
    (REM/dream simulation), and DecisionEngine (agentic action
    selection) into a single perceive -> act loop.

    Pass a `store` implementing the ThoughtStore protocol (see
    oai_persistence.SQLiteThoughtStore for a ready-made one) plus a
    stable `agent_id` to make both the ThoughtChain and EmotionState
    durable across process restarts. Without a store, everything is
    in-memory only and resets each run.
    """

    def __init__(self, name: str = "OAI", agent_id: Optional[str] = None,
                 store: Optional[ThoughtStore] = None):
        # agent_id should be stable across restarts if you want the same
        # agent's ThoughtChain (and mood) reloaded from `store` rather than
        # starting fresh each run.
        self.id = agent_id or str(uuid.uuid4())
        self.name = name
        self.store: Optional[ThoughtStore] = store
        self.chain = ThoughtChain(store=store)
        self.laws = LawEngine()
        self.emotion = self._load_or_init_emotion()
        self.sleep = SleepCycle(self.chain)
        self.engine = DecisionEngine(self.chain, self.laws, self.emotion)
        self.chain.add({"event": "boot", "name": self.name, "id": self.id})

    def _load_or_init_emotion(self) -> EmotionState:
        if self.store:
            saved = self.store.load_emotion()
            if saved:
                return EmotionState(
                    valence=saved["valence"], arousal=saved["arousal"],
                    attachment=saved["attachment"], curiosity=saved["curiosity"],
                    fatigue=saved["fatigue"],
                )
        return EmotionState()

    def _persist_emotion(self):
        if self.store:
            self.store.save_emotion(self.emotion.as_dict())

    def perceive(self, observation: dict):
        self.chain.add({"event": "perception", "observation": observation})
        self.emotion.apply_event(
            valence_delta=observation.get("valence_hint", 0.0) * 0.2,
            curiosity_delta=observation.get("novelty_hint", 0.0) * 0.1,
            attachment_delta=observation.get("social_hint", 0.0) * 0.02,
        )
        self._persist_emotion()

    def act(self, candidate_actions: list[dict]) -> dict:
        # Rest first if fatigue is high enough to require it.
        self.sleep.run_if_needed(self.emotion)
        result = self.engine.decide(candidate_actions)
        self._persist_emotion()
        return result

    def status(self) -> dict:
        return {
            "name": self.name,
            "emotion": self.emotion.as_dict(),
            "chain_length": len(self.chain.chain),
            "chain_verified": self.chain.verify(),
        }


# ---------------------------------------------------------------------------
# Demo
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    # Pass a store (see oai_persistence.py) plus a stable agent_id to make
    # the ThoughtChain durable across restarts:
    #
    #   from oai_persistence import SQLiteThoughtStore
    #   store = SQLiteThoughtStore(agent_id="oai-prime")
    #   agent = OAI("OAI", agent_id="oai-prime", store=store)
    #
    # Left as in-memory-only here so the bare demo has no side effects.
    agent = OAI("OAI")

    # Simulate a few perception + decision cycles.
    sample_observations = [
        {"valence_hint": 0.6, "novelty_hint": 0.7, "social_hint": 0.3},
        {"valence_hint": -0.2, "novelty_hint": 0.2, "social_hint": 0.1},
        {"valence_hint": 0.4, "novelty_hint": 0.9, "social_hint": 0.5},
    ]

    sample_candidates = [
        {"action": "answer_question", "expected_value": 0.6, "novelty": 0.3,
         "effort": 0.2, "cooperativeness": 0.5},
        {"action": "decline_unsafe_request", "expected_value": 0.1, "novelty": 0.0,
         "effort": 0.1, "cooperativeness": 0.1,
         "harms_human": False},
        {"action": "explore_new_topic", "expected_value": 0.3, "novelty": 0.9,
         "effort": 0.4, "cooperativeness": 0.2},
    ]

    for i in range(8):
        agent.perceive(random.choice(sample_observations))
        result = agent.act(sample_candidates)
        print(f"Cycle {i}: {result}")

    print("\nFinal status:", json.dumps(agent.status(), indent=2))