from __future__ import annotations
import os
from pathlib import Path
from typing import Any, Dict, List, Tuple
import gradio as gr
from orbit_core import (
CLAIM_TYPES,
DEFAULT_RELIABILITY,
RELATIONS,
SOURCE_TYPES,
OrbitStore,
decision_gate,
)
APP_DIR = Path(__file__).resolve().parent
DATA_DIR = Path(os.getenv("ORBIT_DATA_DIR", APP_DIR / "data"))
DATA_FILE = DATA_DIR / "beliefs.json"
DATA_DIR.mkdir(parents=True, exist_ok=True)
STORE = OrbitStore(DATA_FILE)
STORE.seed_if_empty()
def belief_row(belief) -> List[Any]:
return [
belief.id,
belief.statement,
belief.context,
belief.claim_type,
belief.status,
round(float(belief.support_weight), 3),
round(float(belief.contradiction_weight), 3),
round(float(belief.confidence), 3),
round(float(belief.pressure), 3),
len(belief.evidence),
round(float(getattr(belief, "source_diversity", 0.0)), 3),
belief.updated_at,
]
def queue_row(belief) -> List[Any]:
next_question = (
belief.revision_triggers[0]
if belief.revision_triggers
else f"What evidence would materially change '{belief.statement}'?"
)
return [
belief.id,
belief.statement,
belief.status,
round(float(belief.confidence), 3),
round(float(belief.pressure), 3),
next_question,
]
def recent_row(belief) -> List[Any]:
return [
belief.id,
belief.statement,
belief.status,
round(float(belief.confidence), 3),
round(float(belief.pressure), 3),
belief.updated_at,
]
def deck_rows(limit: int = 100) -> List[List[Any]]:
return [belief_row(item) for item in STORE.all()[:limit]]
def queue_rows(limit: int = 50) -> List[List[Any]]:
return [queue_row(item) for item in STORE.pressure_queue()[:limit]]
def recent_rows(limit: int = 25) -> List[List[Any]]:
return [recent_row(item) for item in STORE.recent(limit)]
def beliefs_missing_revision(limit: int = 25) -> List[List[Any]]:
rows = []
for belief in STORE.all():
if not belief.revision_triggers:
rows.append(
[
belief.id,
belief.statement,
belief.status,
round(float(belief.confidence), 3),
"No revision trigger recorded",
]
)
return rows[:limit]
def beliefs_missing_limits(limit: int = 25) -> List[List[Any]]:
rows = []
for belief in STORE.all():
if not belief.instrument_limits:
rows.append(
[
belief.id,
belief.statement,
belief.status,
round(float(belief.confidence), 3),
"No instrument limit recorded",
]
)
return rows[:limit]
def low_evidence_high_confidence(limit: int = 25) -> List[List[Any]]:
rows = []
for belief in STORE.all():
if len(belief.evidence) <= 1 and float(belief.confidence) >= 0.7:
rows.append(
[
belief.id,
belief.statement,
belief.status,
round(float(belief.confidence), 3),
len(belief.evidence),
"High confidence with sparse evidence",
]
)
return rows[:limit]
def compute_metrics() -> Dict[str, Any]:
beliefs = STORE.all()
total = len(beliefs)
evidence_count = sum(len(item.evidence) for item in beliefs)
status_counts: Dict[str, int] = {}
for belief in beliefs:
status_counts[belief.status] = status_counts.get(belief.status, 0) + 1
avg_confidence = (
sum(float(item.confidence) for item in beliefs) / total if total else 0.0
)
avg_pressure = (
sum(float(item.pressure) for item in beliefs) / total if total else 0.0
)
avg_diversity = (
sum(float(getattr(item, "source_diversity", 0.0)) for item in beliefs) / total
if total
else 0.0
)
high_pressure = sum(1 for item in beliefs if float(item.pressure) >= 0.6)
missing_revision = sum(1 for item in beliefs if not item.revision_triggers)
missing_limits = sum(1 for item in beliefs if not item.instrument_limits)
return {
"beliefs": total,
"evidence": evidence_count,
"supported": status_counts.get("supported", 0),
"contested": status_counts.get("contested", 0),
"provisional": status_counts.get("provisional", 0),
"contradicted": status_counts.get("contradicted", 0),
"avg_confidence": avg_confidence,
"avg_pressure": avg_pressure,
"avg_diversity": avg_diversity,
"high_pressure": high_pressure,
"missing_revision": missing_revision,
"missing_limits": missing_limits,
}
def serialize_belief(belief) -> Dict[str, Any]:
if belief is None:
return {"error": "belief not found"}
return {
"id": belief.id,
"statement": belief.statement,
"subject": belief.subject,
"predicate": belief.predicate,
"object": belief.obj,
"context": belief.context,
"claim_type": belief.claim_type,
"status": belief.status,
"support_weight": belief.support_weight,
"contradiction_weight": belief.contradiction_weight,
"confidence": belief.confidence,
"pressure": belief.pressure,
"source_diversity": getattr(belief, "source_diversity", 0.0),
"risk_flags": getattr(belief, "risk_flags", []),
"revision_triggers": belief.revision_triggers,
"instrument_limits": belief.instrument_limits,
"evidence": [
{
"id": item.id,
"relation": item.relation,
"source_type": item.source_type,
"source_ref": item.source_ref,
"speaker": item.speaker,
"quote": item.quote,
"note": item.note,
"reliability": item.reliability,
"observed_at": item.observed_at,
"submitted_at": item.submitted_at,
}
for item in belief.evidence
],
"created_at": belief.created_at,
"updated_at": belief.updated_at,
}
def metric_card(title: str, value: str, tone: str = "") -> str:
return f"""
"""
def metrics_html() -> str:
m = compute_metrics()
return f"""
{metric_card("Beliefs", str(m["beliefs"]))}
{metric_card("Evidence", str(m["evidence"]))}
{metric_card("Supported", str(m["supported"]), "success")}
{metric_card("Contested", str(m["contested"]), "warning")}
{metric_card("Contradicted", str(m["contradicted"]), "danger")}
{metric_card("High Pressure", str(m["high_pressure"]), "warning")}
{metric_card("Avg Confidence", f'{m["avg_confidence"]:.2f}', "info")}
{metric_card("Avg Pressure", f'{m["avg_pressure"]:.2f}', "info")}
{metric_card("Avg Source Diversity", f'{m["avg_diversity"]:.2f}', "info")}
"""
def fundamentals_html() -> str:
return """
1. Contextual Sanity
Keep contact with context. Ask when inquiry can change understanding.
2. Provisional Judgment
Form judgments, but bind them to conditions and keep them revisable.
3. Childlike Capacity
Preserve exploration, discovery, and play instead of collapsing too early.
4. Instrument-Bounded Understanding
Every conclusion is limited by the instruments used to produce it.
"""
def health_warnings_markdown() -> str:
m = compute_metrics()
warnings: List[str] = []
if m["missing_revision"] > 0:
warnings.append(
f"- **{m['missing_revision']} beliefs** have no revision trigger recorded."
)
if m["missing_limits"] > 0:
warnings.append(
f"- **{m['missing_limits']} beliefs** have no instrument limit recorded."
)
if m["high_pressure"] > 0:
warnings.append(
f"- **{m['high_pressure']} beliefs** are under elevated pressure."
)
sparse_rows = low_evidence_high_confidence()
if sparse_rows:
warnings.append(
f"- **{len(sparse_rows)} beliefs** have high confidence with sparse evidence."
)
if not warnings:
warnings.append("- No immediate governor warnings detected.")
return "### Governor Health Warnings\n" + "\n".join(warnings)
def overview_markdown() -> str:
m = compute_metrics()
return f"""
### Orbit State
Orbit is a shared belief-revision system. It does not store final truth.
It stores **claims**, **evidence**, **contradiction**, **pressure**, and **what would justify revision**.
**{m["beliefs"]} beliefs** · **{m["evidence"]} evidence records** ·
**{m["supported"]} supported** · **{m["contested"]} contested** ·
**{m["provisional"]} provisional** · **{m["contradicted"]} contradicted**
"""
def ask_orbit(query: str) -> Tuple[str, List[List[Any]], Dict[str, Any]]:
query = (query or "").strip()
if not query:
return (
"""
### Ask Orbit
Enter a question, claim, or topic to inspect what Orbit currently believes.
""",
[],
{"error": "empty query"},
)
matches = STORE.search(query)
if not matches:
return (
f"""
### No matching belief
Orbit has no stored belief for:
> {query}
That does **not** mean the claim is false.
It means Orbit does not yet have enough structured memory around it.
You can contribute evidence below.
""",
[],
{"query": query, "matches": 0},
)
top = matches[0]
evidence_count = len(top.evidence)
summary = f"""
### Orbit's current lean
**{top.statement}**
- **Status:** {top.status}
- **Confidence:** {top.confidence:.3f}
- **Pressure:** {top.pressure:.3f}
- **Evidence records:** {evidence_count}
- **Source diversity:** {getattr(top, "source_diversity", 0.0):.3f}
- **Context:** {top.context or "not declared"}
**Interpretation:** Orbit is showing a bounded, revisable belief — not a declaration of final truth.
"""
return summary, [belief_row(item) for item in matches[:20]], serialize_belief(top)
def orbit_inspect(belief_id: str) -> Dict[str, Any]:
return serialize_belief(STORE.get((belief_id or "").strip()))
def orbit_record(
subject: str,
predicate: str,
obj: str,
context: str,
claim_type: str,
relation: str,
source_type: str,
source_ref: str,
speaker: str,
quote: str,
reliability: float,
note: str,
observed_at: str,
revision_trigger: str,
instrument_limit: str,
) -> Tuple[str, str, str, str, List[List[Any]], List[List[Any]], List[List[Any]], List[List[Any]], List[List[Any]], List[List[Any]], Dict[str, Any]]:
try:
belief = STORE.record_evidence(
subject=subject,
predicate=predicate,
obj=obj,
context=context,
claim_type=claim_type,
relation=relation,
source_type=source_type,
source_ref=source_ref,
speaker=speaker,
quote=quote,
reliability=reliability,
note=note,
observed_at=observed_at,
revision_trigger=revision_trigger,
instrument_limit=instrument_limit,
)
except (TypeError, ValueError) as exc:
return (
f"### Record rejected\n{exc}",
overview_markdown(),
metrics_html(),
health_warnings_markdown(),
deck_rows(),
queue_rows(),
recent_rows(),
beliefs_missing_revision(),
beliefs_missing_limits(),
low_evidence_high_confidence(),
{"error": str(exc)},
)
return (
f"""
### Evidence recorded
Orbit updated belief **{belief.id}**
- **Statement:** {belief.statement}
- **Status:** {belief.status}
- **Confidence:** {belief.confidence:.3f}
- **Pressure:** {belief.pressure:.3f}
""",
overview_markdown(),
metrics_html(),
health_warnings_markdown(),
deck_rows(),
queue_rows(),
recent_rows(),
beliefs_missing_revision(),
beliefs_missing_limits(),
low_evidence_high_confidence(),
serialize_belief(belief),
)
def orbit_pressure_queue() -> List[List[Any]]:
return queue_rows()
def orbit_decision_gate(
confidence: float,
stakes: str,
reversibility: str,
time_pressure: str,
) -> Dict[str, Any]:
return decision_gate(confidence, stakes, reversibility, time_pressure)
def orbit_snapshot() -> Dict[str, Any]:
return STORE.export_snapshot()
def default_reliability(source_type: str) -> float:
return DEFAULT_RELIABILITY.get(source_type, 0.35)
def refresh_all():
return (
metrics_html(),
overview_markdown(),
health_warnings_markdown(),
deck_rows(),
queue_rows(),
recent_rows(),
beliefs_missing_revision(),
beliefs_missing_limits(),
low_evidence_high_confidence(),
)
TABLE_HEADERS = [
"Belief ID",
"Statement",
"Context",
"Claim type",
"Status",
"Support",
"Contradiction",
"Confidence",
"Pressure",
"Evidence",
"Source diversity",
"Updated",
]
QUEUE_HEADERS = [
"Belief ID",
"Statement",
"Status",
"Confidence",
"Pressure",
"Revision question",
]
RECENT_HEADERS = [
"Belief ID",
"Statement",
"Status",
"Confidence",
"Pressure",
"Updated",
]
WARNING_HEADERS = [
"Belief ID",
"Statement",
"Status",
"Confidence",
"Issue",
]
CSS = """
:root {
--bg: #060816;
--panel: rgba(15, 23, 42, 0.88);
--panel2: rgba(30, 41, 59, 0.78);
--border: rgba(148, 163, 184, 0.20);
--text: #e8eef9;
--muted: #9fb0c8;
}
.gradio-container {
background:
radial-gradient(circle at top left, rgba(59,130,246,.11), transparent 28%),
radial-gradient(circle at top right, rgba(139,92,246,.10), transparent 26%),
radial-gradient(circle at bottom left, rgba(6,182,212,.08), transparent 30%),
linear-gradient(180deg, #020617 0%, #0b1120 100%);
color: var(--text);
}
.orbit-shell {
max-width: 1400px;
margin: 0 auto;
}
.orbit-hero {
border: 1px solid var(--border);
border-radius: 24px;
padding: 30px;
background:
radial-gradient(circle at top right, rgba(59,130,246,.18), transparent 35%),
radial-gradient(circle at bottom left, rgba(16,185,129,.10), transparent 35%),
linear-gradient(180deg, rgba(15,23,42,.92), rgba(15,23,42,.74));
box-shadow: 0 20px 60px rgba(0,0,0,.35);
margin-bottom: 18px;
}
.orbit-kicker {
display: inline-block;
font-size: .8rem;
letter-spacing: .08em;
text-transform: uppercase;
color: #c7d2fe;
background: rgba(99,102,241,.15);
border: 1px solid rgba(129,140,248,.22);
border-radius: 999px;
padding: 6px 10px;
margin-bottom: 12px;
}
.orbit-subtitle {
color: var(--muted);
font-size: 1.05rem;
line-height: 1.7;
margin-top: 10px;
max-width: 980px;
}
.metrics-grid {
display: grid;
grid-template-columns: repeat(9, minmax(120px, 1fr));
gap: 12px;
margin: 8px 0 18px 0;
}
.metric-card {
background: linear-gradient(180deg, rgba(15,23,42,.90), rgba(30,41,59,.72));
border: 1px solid var(--border);
border-radius: 18px;
padding: 16px;
min-height: 92px;
display: flex;
flex-direction: column;
justify-content: space-between;
}
.metric-card.success { border-color: rgba(34,197,94,.35); }
.metric-card.warning { border-color: rgba(245,158,11,.35); }
.metric-card.danger { border-color: rgba(239,68,68,.35); }
.metric-card.info { border-color: rgba(6,182,212,.35); }
.metric-title {
color: var(--muted);
font-size: .8rem;
text-transform: uppercase;
letter-spacing: .04em;
}
.metric-value {
font-size: 1.7rem;
font-weight: 700;
color: #fff;
margin-top: 8px;
}
.fundamentals-grid {
display: grid;
grid-template-columns: repeat(4, minmax(180px, 1fr));
gap: 12px;
margin: 8px 0 20px 0;
}
.fundamental-card {
background: linear-gradient(180deg, rgba(15,23,42,.90), rgba(30,41,59,.70));
border: 1px solid var(--border);
border-radius: 18px;
padding: 16px;
min-height: 150px;
}
.fundamental-title {
font-weight: 700;
color: #eef2ff;
margin-bottom: 10px;
}
.fundamental-body {
color: var(--muted);
line-height: 1.6;
}
.gr-button-primary {
background: linear-gradient(90deg, #2563eb, #7c3aed) !important;
border: none !important;
}
thead tr th {
background: rgba(30,41,59,.95) !important;
color: #dbeafe !important;
}
tbody tr:nth-child(even) {
background: rgba(255,255,255,.02) !important;
}
footer { display: none !important; }
@media (max-width: 1250px) {
.metrics-grid {
grid-template-columns: repeat(4, minmax(120px, 1fr));
}
.fundamentals-grid {
grid-template-columns: repeat(2, minmax(180px, 1fr));
}
}
@media (max-width: 700px) {
.metrics-grid {
grid-template-columns: repeat(2, minmax(120px, 1fr));
}
.fundamentals-grid {
grid-template-columns: repeat(1, minmax(180px, 1fr));
}
.orbit-hero {
padding: 22px;
}
}
"""
with gr.Blocks(title="Orbit Command Deck", css=CSS, theme=gr.themes.Soft()) as demo:
gr.Markdown(
"""
Shared belief revision under uncertainty
🪐 Orbit Command Deck
Orbit is a governor for reasoning under uncertainty. It does not store final truth.
It stores claims, evidence, contradiction, pressure, and what would justify revision.
Anyone can query Orbit, inspect its current lean, and contribute new signal.
"""
)
fundamentals = gr.HTML(value=fundamentals_html())
metrics = gr.HTML(value=metrics_html())
overview = gr.Markdown(value=overview_markdown())
health_warnings = gr.Markdown(value=health_warnings_markdown())
with gr.Tab("Ask Orbit"):
gr.Markdown(
"""
Ask Orbit what it currently leans toward on a topic.
Orbit returns a **bounded, revisable belief**, not a declaration of final truth.
"""
)
with gr.Row():
ask_text = gr.Textbox(
label="Question, claim, or topic",
placeholder="What does Orbit currently believe about X?",
scale=6,
)
ask_button = gr.Button("Ask Orbit", variant="primary", scale=1)
ask_summary = gr.Markdown()
ask_table = gr.Dataframe(
headers=TABLE_HEADERS,
interactive=False,
wrap=True,
label="Matching beliefs",
)
ask_detail = gr.JSON(label="Strongest matching belief")
with gr.Tab("Contribute Evidence"):
gr.Markdown(
"""
Contribute support or contradiction to a claim already forming inside Orbit's memory.
Use the simple fields first. Advanced provenance fields keep the belief revisable instead of brittle.
"""
)
with gr.Group():
gr.Markdown("### 1) The claim")
with gr.Row():
subject = gr.Textbox(label="Subject", placeholder="The GSX-R750")
predicate = gr.Textbox(label="Predicate", placeholder="weighs about")
obj = gr.Textbox(label="Object", placeholder="330 lb")
context = gr.Textbox(
label="Context / scope",
placeholder="Thomas's 2001 motorcycle in current configuration",
)
with gr.Group():
gr.Markdown("### 2) The evidence")
with gr.Row():
relation = gr.Radio(
choices=list(RELATIONS),
value="support",
label="Relation",
)
source_type = gr.Dropdown(
choices=list(SOURCE_TYPES),
value="firsthand_report",
label="Source type",
)
claim_type = gr.Dropdown(
choices=list(CLAIM_TYPES),
value="world_claim",
label="Claim type",
)
with gr.Row():
source_ref = gr.Textbox(
label="Source reference",
placeholder="URL, conversation, message, document...",
)
speaker = gr.Textbox(
label="Speaker / observer",
placeholder="Thomas",
)
observed_at = gr.Textbox(
label="Observed at",
placeholder="2026-06-23 or ISO timestamp",
)
quote = gr.Textbox(
label="Exact quote / observation / measurement",
lines=4,
placeholder="Preserve the original wording or measurement here.",
)
note = gr.Textbox(
label="Analyst note",
lines=3,
placeholder="Optional interpretation or caution.",
)
reliability = gr.Slider(
minimum=0.0,
maximum=1.0,
value=DEFAULT_RELIABILITY["firsthand_report"],
step=0.05,
label="Reliability",
)
with gr.Group():
gr.Markdown("### 3) Governor constraints")
revision_trigger = gr.Textbox(
label="Revision trigger",
placeholder="What new signal would materially change this judgment?",
)
instrument_limit = gr.Textbox(
label="Instrument limit",
placeholder="What can this source, sensor, memory, or model not establish?",
)
record_button = gr.Button("Record evidence", variant="primary")
record_status = gr.Markdown()
record_detail = gr.JSON(label="Updated belief")
with gr.Tab("Pressure & Risk"):
gr.Markdown(
"""
These are the places where Orbit most needs attention:
- beliefs under pressure
- beliefs missing revision triggers
- beliefs missing instrument limits
- beliefs with high confidence but sparse evidence
"""
)
refresh_button = gr.Button("Refresh system state", variant="primary")
pressure_table = gr.Dataframe(
headers=QUEUE_HEADERS,
value=queue_rows(),
interactive=False,
wrap=True,
label="Pressure queue",
)
recent_table = gr.Dataframe(
headers=RECENT_HEADERS,
value=recent_rows(),
interactive=False,
wrap=True,
label="Recently updated beliefs",
)
missing_revision_table = gr.Dataframe(
headers=WARNING_HEADERS,
value=beliefs_missing_revision(),
interactive=False,
wrap=True,
label="Beliefs missing revision triggers",
)
missing_limits_table = gr.Dataframe(
headers=WARNING_HEADERS,
value=beliefs_missing_limits(),
interactive=False,
wrap=True,
label="Beliefs missing instrument limits",
)
sparse_confidence_table = gr.Dataframe(
headers=["Belief ID", "Statement", "Status", "Confidence", "Evidence", "Issue"],
value=low_evidence_high_confidence(),
interactive=False,
wrap=True,
label="High confidence / sparse evidence",
)
with gr.Tab("Inspect & Export"):
gr.Markdown(
"""
Inspect a belief directly by ID, or export Orbit's current snapshot.
"""
)
inspect_id = gr.Textbox(label="Belief ID")
with gr.Row():
inspect_button = gr.Button("Inspect", variant="primary")
snapshot_button = gr.Button("Export snapshot")
inspect_result = gr.JSON(label="Belief")
snapshot_result = gr.JSON(label="Orbit snapshot")
with gr.Tab("Decision Gate"):
gr.Markdown(
"""
A belief can be strong enough to use without being strong enough to act on.
Orbit separates judgment from action through stakes, reversibility, and time pressure.
"""
)
action_confidence = gr.Slider(
0.0, 1.0, value=0.60, step=0.01, label="Current confidence"
)
with gr.Row():
stakes = gr.Radio(["low", "medium", "high"], value="medium", label="Stakes")
reversibility = gr.Radio(
["high", "medium", "low"], value="medium", label="Reversibility"
)
time_pressure = gr.Radio(
["low", "medium", "high"], value="medium", label="Time pressure"
)
gate_button = gr.Button("Apply action threshold", variant="primary")
gate_result = gr.JSON(label="Decision gate result")
with gr.Tab("Belief Deck"):
gr.Markdown(
"Browse Orbit's current belief memory. Strong beliefs are still revisable."
)
deck_table = gr.Dataframe(
headers=TABLE_HEADERS,
value=deck_rows(),
interactive=False,
wrap=True,
label="Belief deck",
)
ask_button.click(
ask_orbit,
inputs=[ask_text],
outputs=[ask_summary, ask_table, ask_detail],
api_name="orbit_query",
)
source_type.change(
default_reliability,
inputs=[source_type],
outputs=[reliability],
api_name=False,
)
record_button.click(
orbit_record,
inputs=[
subject,
predicate,
obj,
context,
claim_type,
relation,
source_type,
source_ref,
speaker,
quote,
reliability,
note,
observed_at,
revision_trigger,
instrument_limit,
],
outputs=[
record_status,
overview,
metrics,
health_warnings,
deck_table,
pressure_table,
recent_table,
missing_revision_table,
missing_limits_table,
sparse_confidence_table,
record_detail,
],
api_name="orbit_record",
)
refresh_button.click(
refresh_all,
inputs=[],
outputs=[
metrics,
overview,
health_warnings,
deck_table,
pressure_table,
recent_table,
missing_revision_table,
missing_limits_table,
sparse_confidence_table,
],
api_name=False,
)
inspect_button.click(
orbit_inspect,
inputs=[inspect_id],
outputs=[inspect_result],
api_name="orbit_inspect",
)
snapshot_button.click(
orbit_snapshot,
inputs=[],
outputs=[snapshot_result],
api_name="orbit_snapshot",
)
gate_button.click(
orbit_decision_gate,
inputs=[action_confidence, stakes, reversibility, time_pressure],
outputs=[gate_result],
api_name="orbit_decision_gate",
)
demo.load(
refresh_all,
inputs=[],
outputs=[
metrics,
overview,
health_warnings,
deck_table,
pressure_table,
recent_table,
missing_revision_table,
missing_limits_table,
sparse_confidence_table,
],
)
if __name__ == "__main__":
demo.launch(mcp_server=True)