"""
Operon Bi-Temporal Memory — Explorer (Gradio Demo)
===================================================
Three-tab demo:
1. Fact Timeline — load preset scenarios, inspect facts and corrections
2. Point-in-Time — query belief state at any (valid, record) coordinate
3. Diff & Audit — compare time points, view full audit trails
Run locally:
pip install gradio operon-ai
python space-bitemporal/app.py
"""
import sys
from datetime import datetime, timedelta
from pathlib import Path
import gradio as gr
# Allow importing operon_ai from the repo root when running locally
_repo_root = Path(__file__).resolve().parent.parent
if str(_repo_root) not in sys.path:
sys.path.insert(0, str(_repo_root))
from operon_ai import BiTemporalMemory
# ── Time helpers ────────────────────────────────────────────────────
_BASE = datetime(2026, 3, 15, 9, 0, 0)
def _t(day: int, hour: int = 12) -> datetime:
return _BASE + timedelta(days=day - 1, hours=hour - 9)
def _fmt(dt: datetime) -> str:
return dt.strftime("Day %d %H:%M").replace("Day 0", "Day ").lstrip("0")
def _day_fmt(dt: datetime) -> str:
delta = dt - _BASE
day = delta.days + 1
return f"Day {day} {dt.strftime('%H:%M')}"
# ── Preset scenarios ───────────────────────────────────────────────
def _build_compliance_audit() -> BiTemporalMemory:
"""Financial product approval with retroactive correction."""
mem = BiTemporalMemory()
# Day 1: Quantitative model outputs risk score
mem.record_fact(
subject="product:BOND-7Y", predicate="risk_score",
value=0.42, valid_from=_t(1), recorded_from=_t(1),
source="quant_model", tags=("quantitative",),
)
# Day 2: Liquidity assessed
mem.record_fact(
subject="product:BOND-7Y", predicate="liquidity_class",
value="B", valid_from=_t(2), recorded_from=_t(2),
source="liquidity_engine", tags=("quantitative",),
)
# Day 3: Regulatory category assigned
mem.record_fact(
subject="product:BOND-7Y", predicate="regulatory_category",
value="standard", valid_from=_t(1), recorded_from=_t(3),
source="compliance_team", tags=("regulatory",),
)
# Day 5: Post-approval audit reveals risk was higher
risk_facts = mem.retrieve_valid_at(at=_t(5), subject="product:BOND-7Y",
predicate="risk_score")
if risk_facts:
mem.correct_fact(
old_fact_id=risk_facts[0].fact_id, value=0.68,
valid_from=_t(1), recorded_from=_t(5),
source="audit_review", tags=("correction", "audit"),
)
return mem
def _build_client_onboarding() -> BiTemporalMemory:
"""Client onboarding with progressive fact discovery."""
mem = BiTemporalMemory()
# Day 1: Initial KYC data
mem.record_fact(
subject="client:ACME-42", predicate="kyc_status",
value="pending", valid_from=_t(1), recorded_from=_t(1),
source="onboarding_portal",
)
mem.record_fact(
subject="client:ACME-42", predicate="revenue_band",
value="mid-market", valid_from=_t(1), recorded_from=_t(1),
source="crm_sync",
)
# Day 2: KYC approved
mem.record_fact(
subject="client:ACME-42", predicate="kyc_status",
value="approved", valid_from=_t(2), recorded_from=_t(2),
source="compliance_team",
)
# Day 3: Revenue reclassified (retroactive to day 1)
rev_facts = mem.retrieve_valid_at(at=_t(3), subject="client:ACME-42",
predicate="revenue_band")
if rev_facts:
mem.correct_fact(
old_fact_id=rev_facts[0].fact_id, value="enterprise",
valid_from=_t(1), recorded_from=_t(3),
source="finance_team", tags=("correction",),
)
# Day 4: Tier assigned based on corrected revenue
mem.record_fact(
subject="client:ACME-42", predicate="tier",
value="gold", valid_from=_t(4), recorded_from=_t(4),
source="tier_engine",
)
return mem
def _build_incident_response() -> BiTemporalMemory:
"""Incident timeline with evolving root cause analysis."""
mem = BiTemporalMemory()
# Hour 0: Alert fires
mem.record_fact(
subject="incident:INC-1337", predicate="severity",
value="P2", valid_from=_t(1, 9), recorded_from=_t(1, 9),
source="alerting_system", tags=("auto",),
)
mem.record_fact(
subject="incident:INC-1337", predicate="root_cause",
value="database_timeout", valid_from=_t(1, 9), recorded_from=_t(1, 10),
source="oncall_engineer",
)
# Hour 3: Escalated to P1
sev_facts = mem.retrieve_valid_at(at=_t(1, 12), subject="incident:INC-1337",
predicate="severity")
if sev_facts:
mem.correct_fact(
old_fact_id=sev_facts[0].fact_id, value="P1",
valid_from=_t(1, 9), recorded_from=_t(1, 12),
source="incident_commander", tags=("escalation",),
)
# Hour 5: Root cause updated
rc_facts = mem.retrieve_valid_at(at=_t(1, 14), subject="incident:INC-1337",
predicate="root_cause")
if rc_facts:
mem.correct_fact(
old_fact_id=rc_facts[0].fact_id,
value="network_partition_causing_db_timeout",
valid_from=_t(1, 9), recorded_from=_t(1, 14),
source="sre_team", tags=("correction", "root_cause"),
)
# Hour 8: Resolved
mem.record_fact(
subject="incident:INC-1337", predicate="status",
value="resolved", valid_from=_t(1, 17), recorded_from=_t(1, 17),
source="incident_commander",
)
return mem
PRESETS = {
"Compliance Audit (BOND-7Y)": {
"description": "Financial product approval with post-approval risk correction.",
"build_fn": _build_compliance_audit,
"subjects": ["product:BOND-7Y"],
},
"Client Onboarding (ACME-42)": {
"description": "Progressive KYC and revenue reclassification during onboarding.",
"build_fn": _build_client_onboarding,
"subjects": ["client:ACME-42"],
},
"Incident Response (INC-1337)": {
"description": "Evolving severity and root cause during an incident.",
"build_fn": _build_incident_response,
"subjects": ["incident:INC-1337"],
},
}
# Module-level state: the currently loaded memory instance
_current_mem: BiTemporalMemory | None = None
_current_subjects: list[str] = []
# ── HTML helpers ────────────────────────────────────────────────────
def _badge(text, color="#6366f1"):
return (
f''
f'{text}'
)
def _fact_row_html(f, show_status=True):
status = ""
if show_status:
if f.recorded_to is not None:
status = _badge("CLOSED", "#ef4444")
else:
status = _badge("ACTIVE", "#22c55e")
supersedes = ""
if f.supersedes:
supersedes = f' [corrects {f.supersedes[:8]}]'
tags = ""
if f.tags:
tags = " ".join(_badge(t, "#64748b") for t in f.tags)
valid_to = _day_fmt(f.valid_to) if f.valid_to else "ongoing"
rec_to = _day_fmt(f.recorded_to) if f.recorded_to else "current"
return (
f''
f'| {f.fact_id[:8]} | '
f'{status} | '
f'{f.subject} | '
f'{f.predicate} | '
f'{f.value} | '
f'{_day_fmt(f.valid_from)} — {valid_to} | '
f'{_day_fmt(f.recorded_from)} — {rec_to} | '
f'{f.source}{supersedes} | '
f'{tags} | '
f'
'
)
def _fact_table_html(facts, title="Facts", show_status=True):
if not facts:
return f'{title}: No matching facts.
'
header = (
''
''
'| ID | '
'Status | '
'Subject | '
'Predicate | '
'Value | '
'Valid Time | '
'Record Time | '
'Source | '
'Tags | '
'
'
)
rows = "".join(_fact_row_html(f, show_status) for f in facts)
return (
f'{title}
'
+ header + rows + '
'
)
def _summary_card(label, value, color="#6366f1"):
return (
f''
f'
{label}
'
f'
{value}
'
f'
'
+ _summary_card("Total Facts", len(all_facts), "#3b82f6")
+ _summary_card("Active", len(active), "#22c55e")
+ _summary_card("Closed", len(closed), "#ef4444")
+ _summary_card("Corrections", len(corrections), "#9333ea")
+ '
'
)
timeline = _fact_table_html(
sorted(all_facts, key=lambda f: f.recorded_from),
title=f"Full Timeline ({preset_name})",
)
return summary, timeline
# ── Tab 2: Point-in-Time Query ─────────────────────────────────────
def _run_query(query_type, valid_day, record_day, subject_filter):
if _current_mem is None:
return "Load a scenario first (Tab 1)."
subject = subject_filter.strip() or None
valid_time = _t(int(valid_day))
record_time = _t(int(record_day))
if query_type == "Valid-Time Only":
facts = _current_mem.retrieve_valid_at(at=valid_time, subject=subject)
title = f"Valid at {_day_fmt(valid_time)} (active records only)"
explanation = (
''
'Valid-time query: Returns facts whose valid interval contains '
f'{_day_fmt(valid_time)}, considering only currently active records. '
'This answers: "What is true in the world at this time?"'
'
'
)
elif query_type == "Record-Time Only":
facts = _current_mem.retrieve_known_at(at=record_time, subject=subject)
title = f"Known by {_day_fmt(record_time)} (includes closed records)"
explanation = (
''
'Record-time query: Returns facts the system had recorded by '
f'{_day_fmt(record_time)}, including later-closed records. '
'This answers: "What did the system know at this time?"'
'
'
)
else:
facts = _current_mem.retrieve_belief_state(
at_valid=valid_time, at_record=record_time,
)
title = f"Belief state at (valid={_day_fmt(valid_time)}, record={_day_fmt(record_time)})"
explanation = (
''
'Belief-state query: Intersects both axes — returns facts '
f'valid at {_day_fmt(valid_time)} AND recorded by {_day_fmt(record_time)}. '
'This answers: "What did the system believe was true at this world-time, '
'given only what it knew by this record-time?"'
'
'
)
if subject:
facts = [f for f in facts if f.subject == subject]
return explanation + _fact_table_html(facts, title=title)
# ── Tab 3: Diff & Audit ───────────────────────────────────────────
def _run_diff(axis, t1_day, t2_day):
if _current_mem is None:
return "Load a scenario first (Tab 1)."
t1 = _t(int(t1_day))
t2 = _t(int(t2_day))
diff = _current_mem.diff_between(t1, t2, axis=axis.lower().replace("-time", ""))
axis_label = axis.lower().replace("-time", "")
return _fact_table_html(
diff,
title=f"New facts on {axis_label} axis between {_day_fmt(t1)} and {_day_fmt(t2)}",
)
def _run_audit(subject):
if _current_mem is None:
return "", ""
subject = subject.strip()
if not subject:
subject = _current_subjects[0] if _current_subjects else ""
if not subject:
return "Enter a subject.", ""
history = _current_mem.history(subject)
timeline = _current_mem.timeline_for(subject)
hist_html = _fact_table_html(
history, title=f"History (by record time): {subject}",
)
time_html = _fact_table_html(
timeline, title=f"Timeline (by valid time): {subject}",
)
return hist_html, time_html
# ── Gradio UI ──────────────────────────────────────────────────────
def build_app() -> gr.Blocks:
with gr.Blocks(title="Bi-Temporal Memory Explorer", theme=gr.themes.Soft()) as app:
gr.Markdown(
"# Bi-Temporal Memory Explorer\n"
"Explore **dual time axes**: valid time (when a fact is true) vs "
"record time (when the system learned it). Corrections are "
"append-only --- old records are closed, never mutated.\n\n"
"Start by loading a **preset scenario** in the first tab."
)
with gr.Tabs():
# ── Tab 1: Fact Timeline ─────────────────────────────
with gr.TabItem("Fact Timeline"):
gr.Markdown(
"Load a preset scenario to populate the bi-temporal "
"memory. Each scenario demonstrates corrections and "
"retroactive knowledge updates."
)
with gr.Row():
preset_dd = gr.Dropdown(
choices=list(PRESETS.keys()),
value="Compliance Audit (BOND-7Y)",
label="Preset Scenario",
scale=3,
)
load_btn = gr.Button("Load Scenario", variant="primary", scale=1)
summary_html = gr.HTML()
timeline_html = gr.HTML()
load_btn.click(
fn=_load_preset,
inputs=[preset_dd],
outputs=[summary_html, timeline_html],
)
# ── Tab 2: Point-in-Time Query ───────────────────────
with gr.TabItem("Point-in-Time Query"):
gr.Markdown(
"Query the memory at specific time coordinates. "
"Compare how **valid-time**, **record-time**, and "
"**belief-state** queries produce different results "
"for the same data."
)
with gr.Row():
query_type = gr.Dropdown(
choices=[
"Valid-Time Only",
"Record-Time Only",
"Belief State (both axes)",
],
value="Belief State (both axes)",
label="Query Type",
)
subject_input = gr.Textbox(
value="",
label="Subject Filter (optional)",
placeholder="e.g. product:BOND-7Y",
)
with gr.Row():
valid_slider = gr.Slider(
minimum=1, maximum=7, value=2, step=1,
label="Valid-Time Day",
)
record_slider = gr.Slider(
minimum=1, maximum=7, value=4, step=1,
label="Record-Time Day",
)
query_btn = gr.Button("Query", variant="primary")
query_output = gr.HTML()
query_btn.click(
fn=_run_query,
inputs=[query_type, valid_slider, record_slider, subject_input],
outputs=[query_output],
)
# ── Tab 3: Diff & Audit ──────────────────────────────
with gr.TabItem("Diff & Audit"):
gr.Markdown(
"Compare what changed between two time points, or "
"inspect the full audit trail for a subject."
)
gr.Markdown("### Temporal Diff")
with gr.Row():
diff_axis = gr.Dropdown(
choices=["Valid-Time", "Record-Time"],
value="Record-Time",
label="Axis",
)
diff_t1 = gr.Slider(minimum=1, maximum=7, value=1, step=1, label="From Day")
diff_t2 = gr.Slider(minimum=1, maximum=7, value=5, step=1, label="To Day")
diff_btn = gr.Button("Compute Diff", variant="primary")
diff_output = gr.HTML()
diff_btn.click(
fn=_run_diff,
inputs=[diff_axis, diff_t1, diff_t2],
outputs=[diff_output],
)
gr.Markdown("### Subject Audit Trail")
with gr.Row():
audit_subject = gr.Textbox(
value="",
label="Subject",
placeholder="e.g. product:BOND-7Y",
scale=3,
)
audit_btn = gr.Button("Audit", variant="primary", scale=1)
audit_history = gr.HTML()
audit_timeline = gr.HTML()
audit_btn.click(
fn=_run_audit,
inputs=[audit_subject],
outputs=[audit_history, audit_timeline],
)
return app
if __name__ == "__main__":
app = build_app()
app.launch()