"""Local Luxzia research journal, fictional scenarios and one-pass reminder queue. Persona descriptions are metadata. No predictive or diagnostic capability is claimed. """ import argparse import contextlib from datetime import datetime, timezone, timedelta import fcntl import json import os from pathlib import Path import sqlite3 import stat import uuid def utc_now(): return datetime.now(timezone.utc) def timestamp(value): if not isinstance(value,str) or len(value)>80: raise ValueError('Timestamp must be a bounded ISO string') result=datetime.fromisoformat(value.replace('Z','+00:00')) if result.tzinfo is None or result.utcoffset() is None: raise ValueError('Timestamp requires a timezone offset') return result.astimezone(timezone.utc) def text(value, *, optional=False): if value is None and optional: return None if not isinstance(value,str) or not value.strip() or len(value)>100000: raise ValueError('Text must be nonempty and at most 100000 characters') return value @contextlib.contextmanager def transaction(conn): conn.execute('BEGIN IMMEDIATE') try: yield conn.commit() except BaseException: conn.rollback() raise SCHEMA=[ "CREATE TABLE lucxia_knowledge_history(history_id INTEGER PRIMARY KEY AUTOINCREMENT,topic_name TEXT NOT NULL,verbatim_focus_text TEXT NOT NULL,research_source_notes TEXT,user_feedback_text TEXT,recorded_at TEXT NOT NULL)", "CREATE TRIGGER knowledge_no_update BEFORE UPDATE ON lucxia_knowledge_history BEGIN SELECT RAISE(ABORT,'Append-only research history'); END", "CREATE TRIGGER knowledge_no_delete BEFORE DELETE ON lucxia_knowledge_history BEGIN SELECT RAISE(ABORT,'Append-only research history'); END", "CREATE TABLE lucxia_event_timeline(event_id INTEGER PRIMARY KEY AUTOINCREMENT,explicit_timestamp TEXT NOT NULL,observed_information TEXT NOT NULL,verification_state TEXT NOT NULL CHECK(verification_state IN ('MANUALLY_DESCRIBED','OBSERVED_INTERACTION','EVIDENCE_REVIEWED')),evidence_reference TEXT,logged_at TEXT NOT NULL,CHECK(verification_state!='EVIDENCE_REVIEWED' OR evidence_reference IS NOT NULL))", "CREATE TABLE lucxia_fictional_scenarios(scenario_uid TEXT PRIMARY KEY NOT NULL,abstract_marker_label TEXT NOT NULL,demographic_profile_criteria TEXT NOT NULL,simulated_target_timestamp TEXT NOT NULL,simulated_delivery_deadline TEXT NOT NULL,logged_at TEXT NOT NULL)", "CREATE TABLE lucxia_reminder_queue(reminder_id INTEGER PRIMARY KEY AUTOINCREMENT,request_uid TEXT NOT NULL UNIQUE,associated_project TEXT NOT NULL,target_timestamp TEXT NOT NULL,reminder_payload_text TEXT NOT NULL,processing_state TEXT NOT NULL DEFAULT 'PENDING' CHECK(processing_state IN ('PENDING','QUEUED_TO_OUTBOX','MISSED_LATE_RUN','SUPPRESSED')),logged_at TEXT NOT NULL)", "CREATE TABLE lucxia_notification_outbox(outbox_id INTEGER PRIMARY KEY AUTOINCREMENT,reminder_id INTEGER NOT NULL UNIQUE REFERENCES lucxia_reminder_queue(reminder_id),payload_text TEXT NOT NULL,delivery_state TEXT NOT NULL DEFAULT 'QUEUED' CHECK(delivery_state IN ('QUEUED','GATEWAY_ACCEPTED','UNKNOWN','FAILED')),queued_at TEXT NOT NULL)", ] def default_database(root): root=Path(root) current=root/'luxzia_research.db' legacy=root/'lucxia_research.db' if current.exists() and legacy.exists(): raise ValueError('Both research database names exist; resolve explicitly to avoid splitting history') return legacy if legacy.exists() else current class LuxziaResearchSystem: def __init__(self,db_filepath=None): self.path=Path(db_filepath if db_filepath is not None else default_database(Path('/Users/bpa/ForesixaQuant'))).absolute() self.persona_operating_frame='Meditative — Flow State' self.persona_temporal_horizon='Beyond Space & Time' self.conn=None self.lock=None if self.path.name=='qzotic_pure_sandbox.db': raise ValueError('Luxzia must use a separate research database') if not self.path.parent.is_dir() or self.path.parent.is_symlink() or self.path.is_symlink(): raise ValueError('Use an existing nonsymlink database directory') fd=os.open(str(self.path)+'.lucxia.lock',os.O_CREAT|os.O_RDWR|os.O_NOFOLLOW,0o600) self.lock=os.fdopen(fd,'a+') try: st=os.fstat(fd) if st.st_uid!=os.getuid() or st.st_mode & 0o077: raise ValueError('Unsafe lock ownership/permissions') fcntl.flock(self.lock,fcntl.LOCK_EX|fcntl.LOCK_NB) if self.path.exists(): st=self.path.stat() if not stat.S_ISREG(st.st_mode) or st.st_uid!=os.getuid() or st.st_mode & 0o077: raise ValueError('Research database must be private, mode 600 or 400') else: fd=os.open(self.path,os.O_CREAT|os.O_EXCL|os.O_WRONLY|os.O_NOFOLLOW,0o600);os.close(fd) self.conn=sqlite3.connect(self.path,timeout=5,isolation_level=None) self.conn.row_factory=sqlite3.Row self.conn.execute('PRAGMA foreign_keys=ON') self.conn.execute('PRAGMA synchronous=FULL') self._initialize_production_schemas() except BaseException: self.close() raise def _initialize_production_schemas(self): c=self.conn version=c.execute('PRAGMA user_version').fetchone()[0] if version==0: if c.execute("SELECT 1 FROM sqlite_master WHERE name NOT LIKE 'sqlite_%' LIMIT 1").fetchone(): raise ValueError('Existing unversioned database requires a reviewed migration; not modified') with transaction(c): for sql in SCHEMA:c.execute(sql) c.execute('PRAGMA user_version=1') elif version!=1: raise ValueError('Unsupported research database version') baseline=sqlite3.connect(':memory:') try: for sql in SCHEMA:baseline.execute(sql) expected={r[0]:r[1] for r in baseline.execute("SELECT name,sql FROM sqlite_master WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%'")} finally:baseline.close() actual={r[0]:r[1] for r in c.execute("SELECT name,sql FROM sqlite_master WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%'")} if actual!=expected:raise ValueError('Research schema differs from supported version') if c.execute('PRAGMA foreign_key_check').fetchall():raise ValueError('Foreign-key check failed') if [r[0] for r in c.execute('PRAGMA integrity_check')]!=['ok']:raise ValueError('Database integrity failed') def record_knowledge_revision(self,topic,focus_desc,sources=None,feedback=None): values=(text(topic),text(focus_desc),text(sources,optional=True),text(feedback,optional=True),utc_now().isoformat()) with transaction(self.conn): return self.conn.execute('INSERT INTO lucxia_knowledge_history(topic_name,verbatim_focus_text,research_source_notes,user_feedback_text,recorded_at) VALUES(?,?,?,?,?)',values).lastrowid def record_event(self,event_timestamp,information,verification_state='MANUALLY_DESCRIBED',evidence_reference=None): if verification_state not in ('MANUALLY_DESCRIBED','OBSERVED_INTERACTION','EVIDENCE_REVIEWED'):raise ValueError('Unknown verification state') if verification_state=='EVIDENCE_REVIEWED' and not evidence_reference:raise ValueError('Reviewed event requires a supplied evidence reference') with transaction(self.conn): return self.conn.execute('INSERT INTO lucxia_event_timeline(explicit_timestamp,observed_information,verification_state,evidence_reference,logged_at) VALUES(?,?,?,?,?)',(timestamp(event_timestamp).isoformat(),text(information),verification_state,text(evidence_reference,optional=True),utc_now().isoformat())).lastrowid def seed_fictional_scenarios(self): # Explicit demonstration setup only; never run automatically at startup. profiles=['Woman (Age 21-35)','Woman (Age 21-35)','Man (Age 21-35)','Older Person (Age > 65)','Child (Age Exactly 7)'] target=timestamp('2026-11-01T23:59:59Z') gate=target-timedelta(hours=72) with transaction(self.conn): for i,profile in enumerate(profiles,1): self.conn.execute('INSERT INTO lucxia_fictional_scenarios VALUES(?,?,?,?,?,?) ON CONFLICT(scenario_uid) DO NOTHING',(f'SCEN_20261101_{i:02d}','NOV_01_TEST_NODE',profile,target.isoformat(),gate.isoformat(),utc_now().isoformat())) def schedule_reminder(self,project,target_timestamp,payload,request_uid=None): uid=str(uuid.UUID(request_uid)) if request_uid else str(uuid.uuid4()) values=(text(project),timestamp(target_timestamp).isoformat(),text(payload)) with transaction(self.conn): existing=self.conn.execute('SELECT * FROM lucxia_reminder_queue WHERE request_uid=?',(uid,)).fetchone() if existing: if (existing['associated_project'],existing['target_timestamp'],existing['reminder_payload_text'])!=values: raise ValueError('Reused request ID with different reminder content') return existing['reminder_id'] return self.conn.execute('INSERT INTO lucxia_reminder_queue(request_uid,associated_project,target_timestamp,reminder_payload_text,logged_at) VALUES(?,?,?,?,?)',(uid,*values,utc_now().isoformat())).lastrowid def suppress_reminder(self,reminder_id): with transaction(self.conn): count=self.conn.execute("UPDATE lucxia_reminder_queue SET processing_state='SUPPRESSED' WHERE reminder_id=? AND processing_state='PENDING'",(reminder_id,)).rowcount if count!=1:raise ValueError('Only a pending reminder can be suppressed') def check_and_queue_scheduled_reminders(self,current_time=None): current=utc_now() if current_time is None else timestamp(current_time) results=[] with transaction(self.conn): for item in self.conn.execute("SELECT * FROM lucxia_reminder_queue WHERE processing_state='PENDING' ORDER BY reminder_id").fetchall(): target=timestamp(item['target_timestamp']) if current>=target: state='MISSED_LATE_RUN' elif current>=target-timedelta(hours=72): state='QUEUED_TO_OUTBOX' self.conn.execute('INSERT INTO lucxia_notification_outbox(reminder_id,payload_text,queued_at) VALUES(?,?,?)',(item['reminder_id'],item['reminder_payload_text'],current.isoformat())) else:continue self.conn.execute('UPDATE lucxia_reminder_queue SET processing_state=? WHERE reminder_id=?',(state,item['reminder_id'])) results.append({'reminder_id':item['reminder_id'],'state':state}) return results def explore_timeline_horizon(self): # No current-time call: simulation outcomes use only stored records. return {'name':'Luxzia','persona_operating_frame':self.persona_operating_frame,'persona_temporal_horizon':self.persona_temporal_horizon,'simulation_mode':'Clock-independent exploration of stored fictional scenarios','scenarios':[dict(r) for r in self.conn.execute('SELECT * FROM lucxia_fictional_scenarios ORDER BY scenario_uid')],'recorded_events':[dict(r) for r in self.conn.execute('SELECT * FROM lucxia_event_timeline ORDER BY explicit_timestamp,event_id')],'notice':'Persona and research descriptions do not establish predictive or diagnostic powers. Verification states record supplied assertions and evidence references, not independent automated validation.'} def close(self): if self.conn:self.conn.close();self.conn=None if self.lock:self.lock.close();self.lock=None def __enter__(self):return self def __exit__(self,*args):self.close() def main(): p=argparse.ArgumentParser() p.add_argument('--root',type=Path,default=Path('/Users/bpa/ForesixaQuant')) p.add_argument('--seed-fictional-scenarios',action='store_true') p.add_argument('--check-reminders',action='store_true') args=p.parse_args() os.umask(0o077) with LuxziaResearchSystem(default_database(args.root)) as lucxia: if args.seed_fictional_scenarios:lucxia.seed_fictional_scenarios() queued=lucxia.check_and_queue_scheduled_reminders() if args.check_reminders else [] print(json.dumps({'research':lucxia.explore_timeline_horizon(),'reminder_changes':queued,'notification_transport':'NOT_CONNECTED','scheduler':'NOT_INSTALLED'},indent=2)) # Compatibility for existing imports; internal table names remain unchanged. LucxiaResearchSystem = LuxziaResearchSystem if __name__=='__main__':main()