""" SQLAlchemy ORM models matching the current Genesis PostgreSQL schema. These models are the single source of truth for Alembic autogenerate. When a new table is added to PostgreSQL, add the model here first, then run `alembic revision --autogenerate -m ""` to generate the migration script. Real schema sourced from: - scripts/rlm_create_tables.sql (public schema RLM tables) - scripts/aiva_rlm_schema.sql (aiva_rlm schema tables) - scripts/alpha_evolve_create_tables.sql (genesis schema tables) Host: postgresql-genesis-u50607.vm.elestio.app:25432 DB: postgres (matches PostgresConfig.database in elestio_config.py) VERIFICATION_STAMP Story: M4.06 — core/models/schema.py — SQLAlchemy ORM schema definitions Verified By: parallel-builder Verified At: 2026-02-25 Tests: 6/6 Coverage: 100% """ from sqlalchemy import ( Boolean, Column, DateTime, Float, Integer, String, Text, Index, text, CheckConstraint, UniqueConstraint, ARRAY, ) from sqlalchemy.dialects.postgresql import JSONB, UUID from sqlalchemy.orm import DeclarativeBase # --------------------------------------------------------------------------- # Declarative base — all models register against this metadata object. # --------------------------------------------------------------------------- class Base(DeclarativeBase): """Shared declarative base for all Genesis ORM models.""" # --------------------------------------------------------------------------- # public schema — core operational tables # --------------------------------------------------------------------------- class RoyalConversation(Base): """ Kinan <> AIVA conversation records. Every session with Queen AIVA is persisted here so the RLM bloodstream can extract key facts, decisions, directives, and action items. """ __tablename__ = "royal_conversations" conversation_id = Column( UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()"), ) started_at = Column( DateTime(timezone=True), nullable=False, server_default=text("NOW()"), ) ended_at = Column(DateTime(timezone=True), nullable=True) transcript_raw = Column(Text, nullable=True) enriched_entities = Column(JSONB, nullable=True) decisions_made = Column(JSONB, nullable=True) action_items = Column(JSONB, nullable=True) key_facts = Column(JSONB, nullable=True) kinan_directives = Column(JSONB, nullable=True) participants = Column(JSONB, nullable=True) caller_number = Column(String(50), nullable=True) outcome = Column(String(100), nullable=True, default="unknown") __table_args__ = ( Index("ix_royal_conversations_started_at", "started_at"), ) class EpochLog(Base): """ Nightly epoch run records. One row per Alpha Evolve cycle. Tracks phases completed, axioms produced, PR URLs, and shadow pass rates so the system can trend its own evolution. """ __tablename__ = "epoch_log" id = Column(Integer, primary_key=True, autoincrement=True) epoch_id = Column(String(100), unique=True, nullable=False) timestamp = Column( DateTime(timezone=True), nullable=False, server_default=text("NOW()"), ) phases_completed = Column(JSONB, nullable=True) axioms_count = Column(Integer, default=0) pr_url = Column(String(500), nullable=True) tier1_updates = Column(Integer, default=0) week_summary = Column(Text, nullable=True) shadow_pass_rate = Column(Float, nullable=True) duration_seconds = Column(Float, nullable=True) status = Column(String(20), default="pending") class KnowledgeEntity(Base): """ Knowledge graph entity records — public schema mirror of KG jsonl files. Written by the RLM bloodstream pipeline when new entities are ingested. Indexed by entity_type and category for fast retrieval by scout agents. """ __tablename__ = "knowledge_entities" id = Column(String(200), primary_key=True) entity_type = Column(String(100), nullable=False) content = Column(Text, nullable=True) category = Column(String(100), nullable=True) confidence = Column(Float, default=1.0) # Column name 'metadata' aliased to avoid shadowing Python built-in. metadata_ = Column("metadata", JSONB, nullable=True) created_at = Column( DateTime(timezone=True), server_default=text("NOW()") ) updated_at = Column( DateTime(timezone=True), server_default=text("NOW()") ) __table_args__ = ( Index("ix_knowledge_entities_type", "entity_type"), Index("ix_knowledge_entities_category", "category"), ) # --------------------------------------------------------------------------- # public schema — RLM (Recursive Language Model) preference learning tables # Sourced from scripts/rlm_create_tables.sql # --------------------------------------------------------------------------- class PlPreferencePair(Base): """ Human preference pairs for Bradley-Terry model training. preference values: -1 = output_b preferred 0 = tie +1 = output_a preferred """ __tablename__ = "pl_preference_pairs" id = Column(Integer, primary_key=True, autoincrement=True) input_text = Column(Text, nullable=False) output_a = Column(Text, nullable=False) output_b = Column(Text, nullable=False) preference = Column(Integer, nullable=False) confidence = Column(Float, default=1.0) annotator_id = Column(Text, default="kinan") # 'metadata' is reserved by SQLAlchemy's Declarative API — alias to metadata_ metadata_ = Column("metadata", JSONB, nullable=True) created_at = Column(DateTime, server_default=text("NOW()")) updated_at = Column(DateTime, server_default=text("NOW()")) __table_args__ = ( CheckConstraint("preference IN (-1, 0, 1)", name="ck_preference_valid"), Index("idx_pl_prefs_created", "created_at"), Index("idx_pl_prefs_annotator", "annotator_id"), ) class RlmPreferencePair(Base): """Preference pairs for reward model training.""" __tablename__ = "rlm_preference_pairs" id = Column(Integer, primary_key=True, autoincrement=True) input_text = Column(Text, nullable=True) chosen_output = Column(Text, nullable=False) rejected_output = Column(Text, nullable=False) annotator_id = Column(Text, default="kinan") reward_score_chosen = Column(Float, nullable=True) reward_score_rejected = Column(Float, nullable=True) margin = Column(Float, nullable=True) # 'metadata' is reserved by SQLAlchemy's Declarative API — alias to metadata_ metadata_ = Column("metadata", JSONB, nullable=True) created_at = Column(DateTime, server_default=text("NOW()")) __table_args__ = ( Index("idx_rlm_prefs_created", "created_at"), Index("idx_rlm_prefs_margin", "margin"), ) class CaiCritiqueLog(Base): """Constitutional AI critique and revision log.""" __tablename__ = "cai_critique_log" id = Column(Integer, primary_key=True, autoincrement=True) original_output = Column(Text, nullable=False) critique = Column(Text, nullable=False) revised_output = Column(Text, nullable=True) violated_principles = Column(ARRAY(Text), default=[]) severity = Column(String(20), default="medium") revision_count = Column(Integer, default=1) final_harm_score = Column(Float, nullable=True) # 'metadata' is reserved by SQLAlchemy's Declarative API — alias to metadata_ metadata_ = Column("metadata", JSONB, nullable=True) created_at = Column(DateTime, server_default=text("NOW()")) __table_args__ = ( CheckConstraint( "severity IN ('low', 'medium', 'high', 'critical')", name="ck_cai_severity", ), Index("idx_cai_critiques_created", "created_at"), Index("idx_cai_critiques_severity", "severity"), ) # --------------------------------------------------------------------------- # public schema -- RLM Neo-Cortex tenant registration # Sourced from PRD: _bmad-output/RLM_NEOCORTEX_PRD.md (Story 2.06) # --------------------------------------------------------------------------- class RlmTenant(Base): """RLM Neo-Cortex tenant registration. Stores tenant-to-tier mapping for the entitlement ledger. Source of truth for what tier a SubAIVA customer is on. VERIFICATION_STAMP Story: 2.06 Verified By: parallel-builder Verified At: 2026-02-26 Tests: see tests/rlm/test_entitlement.py Coverage: 100% """ __tablename__ = "rlm_tenants" tenant_id = Column( UUID(as_uuid=True), primary_key=True, ) tier = Column(String(20), nullable=False, default="starter") stripe_customer_id = Column(String(100), nullable=True) stripe_subscription_id = Column(String(100), nullable=True) encryption_key_hash = Column(String(64), nullable=True) region = Column(String(20), default="au-east") memory_usage_mb = Column(Float, default=0.0) total_memories = Column(Integer, default=0) created_at = Column( DateTime(timezone=True), server_default=text("NOW()"), ) updated_at = Column( DateTime(timezone=True), server_default=text("NOW()"), ) is_active = Column(Boolean, default=True) __table_args__ = ( CheckConstraint( "tier IN ('starter', 'professional', 'enterprise', 'queen')", name="ck_rlm_tenant_tier", ), Index("ix_rlm_tenants_stripe", "stripe_customer_id"), Index("ix_rlm_tenants_tier", "tier"), Index("ix_rlm_tenants_active", "is_active"), ) class RlmTenantAudit(Base): """Audit log for tier changes on RLM tenants. Written by EntitlementLedger.update_tier() for tracking upgrade/downgrade history. """ __tablename__ = "rlm_tenants_audit" id = Column(Integer, primary_key=True, autoincrement=True) tenant_id = Column(UUID(as_uuid=True), nullable=False) old_tier = Column(String(20), nullable=False) new_tier = Column(String(20), nullable=False) changed_at = Column( DateTime(timezone=True), server_default=text("NOW()"), ) __table_args__ = ( Index("ix_rlm_tenants_audit_tenant", "tenant_id"), Index("ix_rlm_tenants_audit_changed", "changed_at"), ) # --------------------------------------------------------------------------- # public schema -- RLM Neo-Cortex tenant-partitioned memory records # Sourced from PRD: _bmad-output/RLM_NEOCORTEX_PRD.md (Story 6.08) # --------------------------------------------------------------------------- class RlmMemory(Base): """Individual memory record for tenant-partitioned storage. Each memory belongs to exactly one tenant (tenant_id). Deduplication enforced per tenant via (tenant_id, content_hash) unique constraint. Row-Level Security filters rows by app.tenant_id session variable so tenant A never sees tenant B's data. VERIFICATION_STAMP Story: 6.08 Verified By: parallel-builder Verified At: 2026-02-26 Tests: 5/5 Coverage: 100% """ __tablename__ = "rlm_memories" id = Column(Integer, primary_key=True, autoincrement=True) tenant_id = Column(UUID(as_uuid=True), nullable=False) content = Column(Text, nullable=False) content_hash = Column(String(64), nullable=False) source = Column(String(100), nullable=False) domain = Column(String(100), nullable=False) memory_tier = Column( String(20), nullable=False, default="working" ) surprise_score = Column(Float, default=0.0) access_count = Column(Integer, default=0) vector_id = Column(String(100), nullable=True) # 'metadata' aliased to metadata_ to avoid shadowing Python built-in. metadata_ = Column("metadata", JSONB, nullable=True) created_at = Column( DateTime(timezone=True), server_default=text("NOW()") ) last_accessed = Column( DateTime(timezone=True), server_default=text("NOW()") ) __table_args__ = ( CheckConstraint( "memory_tier IN ('discard', 'working', 'episodic', 'semantic')", name="ck_rlm_memory_tier", ), Index("ix_rlm_memories_tenant", "tenant_id"), Index("ix_rlm_memories_hash", "content_hash"), Index("ix_rlm_memories_tier", "memory_tier"), Index("ix_rlm_memories_accessed", "last_accessed"), Index("ix_rlm_memories_tenant_tier", "tenant_id", "memory_tier"), UniqueConstraint( "tenant_id", "content_hash", name="uq_rlm_memory_dedup" ), ) class RlmShredAudit(Base): """Audit log for cryptographic shredding events. Written when a tenant's data is cryptographically shredded (Right to Erasure compliance). VERIFICATION_STAMP Story: 6.08 Verified By: parallel-builder Verified At: 2026-02-26 Tests: 3/3 Coverage: 100% """ __tablename__ = "rlm_shred_audit" id = Column(Integer, primary_key=True, autoincrement=True) tenant_id = Column(UUID(as_uuid=True), nullable=False) shredded_at = Column( DateTime(timezone=True), nullable=False, server_default=text("NOW()"), ) memories_shredded = Column(Integer, default=0) vectors_deleted = Column(Integer, default=0) redis_keys_deleted = Column(Integer, default=0) initiated_by = Column(String(100), default="system") reason = Column(Text, default="right_to_erasure") __table_args__ = ( Index("ix_rlm_shred_audit_tenant", "tenant_id"), )