"""
app.py — LifeStack Gradio Demo App
Hackathon presentation interface for the LifeStack simulation engine.
"""
import os
import json
import copy
import gradio as gr
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
# ─── LifeStack modules ────────────────────────────────────────────────────────
from core.life_state import LifeMetrics, ResourceBudget
from core.lifestack_env import LifeStackEnv, LifeStackAction
from agent.agent import LifeStackAgent
from intake.simperson import SimPerson
from agent.conflict_generator import ConflictEvent, generate_conflict, TEMPLATES
from core.action_space import apply_action, validate_action
from agent.memory import LifeStackMemory
from core.metric_schema import normalize_metric_path, is_valid_metric_path
from core.reward import compute_reward
from intake.intake import LifeIntake
from agent.conflict_predictor import ConflictPredictor
from agent.counterfactuals import generate_counterfactuals
from scripts.longitudinal_demo import LongitudinalDemo
from intake.gmail_intake import GmailIntake
from core.task import Task, ExoEvent, Route, Milestone
from core.feedback import OutcomeFeedback, compute_human_feedback_reward
# ─── Pre-load at startup ──────────────────────────────────────────────────────
print("🚀 LifeStack booting…")
AGENT = LifeStackAgent()
MEMORY = LifeStackMemory(silent=True)
INTAKE = LifeIntake()
GMAIL = GmailIntake()
LONG_DEMO = LongitudinalDemo()
# Pre-seed Arjun's 3-week context into ChromaDB on startup
LONG_DEMO.pre_seed_arjun()
# Friday 6PM is always the default demo conflict
DEMO_CONFLICT = next(t for t in TEMPLATES if t.id == "d5_friday")
PERSONS = {
"Alex (Executive) — driven, high-stress":
SimPerson(openness=0.4, conscientiousness=0.9, extraversion=0.7, agreeableness=0.25, neuroticism=0.8, name="Alex (Executive)"),
"Chloe (Creative) — spontaneous, resilient":
SimPerson(openness=0.9, conscientiousness=0.2, extraversion=0.5, agreeableness=0.70, neuroticism=0.15, name="Chloe (Creative)"),
"Sam (Introvert) — anxious, thoughtful":
SimPerson(openness=0.5, conscientiousness=0.6, extraversion=0.1, agreeableness=0.65, neuroticism=0.9, name="Sam (Introvert)"),
"Maya (Family) — empathetic, nurturing":
SimPerson(openness=0.5, conscientiousness=0.7, extraversion=0.5, agreeableness=0.95, neuroticism=0.3, name="Maya (Family)"),
"Leo (Student) — curious, organised":
SimPerson(openness=0.85, conscientiousness=0.8, extraversion=0.4, agreeableness=0.4, neuroticism=0.55, name="Leo (Student)"),
"Arjun (Startup Lead) — high- conscientiousness, high-neuroticism":
SimPerson(name="Arjun", openness=0.4, conscientiousness=0.9, extraversion=0.7, agreeableness=0.25, neuroticism=0.8),
}
CONFLICT_CHOICES = {f"[Diff {t.difficulty}] {t.title}": t for t in TEMPLATES}
PERSON_CHOICES = list(PERSONS.keys())
CONFLICT_CHOICES_LIST = list(CONFLICT_CHOICES.keys())
DEFAULT_CONFLICT = next(k for k in CONFLICT_CHOICES_LIST if "Friday 6PM" in k)
DEMO_PREDICTOR = ConflictPredictor()
print("✅ LifeStack ready.")
# ─── Helpers ──────────────────────────────────────────────────────────────────
DOMAIN_EMOJI = {
"career": "💼", "finances": "💰", "relationships": "❤️",
"physical_health": "💪", "mental_wellbeing": "🧠", "time": "📅",
}
# Metrics where HIGH = BAD (inverted color logic)
INVERTED_METRICS = {"stress_level", "debt_pressure", "workload", "commute_burden", "admin_overhead"}
def _metric_color(key: str, val: float) -> str:
"""Return CSS color: inverted for 'bad-when-high' metrics."""
sub = key.split(".")[-1]
if sub in INVERTED_METRICS:
return "#f87171" if val > 70 else ("#facc15" if val >= 40 else "#4ade80")
return "#4ade80" if val > 70 else ("#facc15" if val >= 40 else "#f87171")
def metrics_html(flat: dict, title: str = "", before: dict = None) -> str:
"""Render metrics as coloured progress bars.
If `before` is supplied, metrics that changed >1 pt show ↑/↓ + delta.
"""
domains = ["career", "finances", "relationships", "physical_health", "mental_wellbeing", "time"]
rows = []
if title:
rows.append(f"{title}
")
for dom in domains:
emoji = DOMAIN_EMOJI[dom]
rows.append(f"{emoji} {dom.upper()}
")
sub = {k: v for k, v in flat.items() if k.startswith(dom + ".")}
for key, val in sub.items():
name = key.split(".")[1].replace("_", " ")
color = _metric_color(key, val)
pct = min(val, 100)
delta_str = ""
if before is not None and key in before:
delta = val - before[key]
if abs(delta) > 1.0:
arrow = "↑" if delta > 0 else "↓"
dc = "#4ade80" if delta > 0 else "#f87171"
delta_str = (
f""
f"{arrow} ({delta:+.1f})"
)
rows.append(
f""
f"
{name}"
f"
"
f"
{val:.1f}"
f" {delta_str}"
f"
" + "\n".join(rows) + "
"
def _init_env(conflict: ConflictEvent) -> LifeStackEnv:
env = LifeStackEnv()
env.reset(conflict=conflict.primary_disruption, budget=conflict.resource_budget)
return env
def task_html(task: Task) -> str:
if not task:
return "No active task
"
routes_html = "".join([f"{r.name}: {r.description}
Req. Actions: {r.required_action_types} | Reward: +{r.final_reward}" for r in task.viable_routes])
if not routes_html: routes_html = "No routes"
milestones_html = "".join([f"{m.id}: {m.description}
Reward: +{m.reward}" for m in task.milestones])
if not milestones_html: milestones_html = "No milestones"
return f"""
🎯 Goal: {task.goal}
Domain: {task.domain} | Difficulty: {task.difficulty}/5 | Horizon: {task.horizon} steps
CONSTRAINTS:
{task.constraints}
"""
def event_log_html(events: list[ExoEvent]) -> str:
if not events:
return "No events triggered yet.
"
rows = []
for e in events:
rows.append(f" Step {e.step}
{e.id.upper()}: {e.description}
")
return "" + "\n".join(rows) + "
"
def route_status_html(routes: list[Route], closed: set[str]) -> str:
if not routes:
return "No routes configured.
"
rows = []
for r in routes:
if r.id in closed:
icon, color = "❌", "#f87171"
status = "CLOSED"
else:
icon, color = "✅", "#4ade80"
status = "OPEN"
rows.append(f"")
return "" + "\n".join(rows) + "
"
def _normalize_action_metric_changes(action) -> None:
fixed_changes = {}
for path, delta in action.primary.metric_changes.items():
raw_path = str(path)
if "." not in raw_path:
raw_path = f"{action.primary.target_domain}.{raw_path}"
norm_path = normalize_metric_path(raw_path)
if not is_valid_metric_path(norm_path):
continue
try:
fixed_changes[norm_path] = float(delta)
except (ValueError, TypeError):
continue
action.primary.metric_changes = fixed_changes
# ─── Cascade Animation Engine ────────────────────────────────────────────────
def animate_cascade(primary_disruption: dict, metrics: LifeMetrics) -> list[dict]:
"""Replay the cascade step-by-step and capture intermediate frames.
Returns a list of frames. Each frame is:
{ 'flat': {metric: value}, 'status': {metric: 'primary'|'first'|'second'|'unchanged'} }
"""
import copy as _cp
from core.life_state import DependencyGraph, CASCADE_DAMPENING_DEFAULT
graph = DependencyGraph()
dampening = CASCADE_DAMPENING_DEFAULT
frames = []
# Frame 0 — initial stable state
base = _cp.deepcopy(metrics)
base_flat = base.flatten()
frames.append({
'flat': dict(base_flat),
'status': {k: 'unchanged' for k in base_flat},
})
# Frame 1 — primary disruption only (no cascade)
f1 = _cp.deepcopy(metrics)
primary_keys = set()
for path, amount in primary_disruption.items():
if '.' not in path:
continue
primary_keys.add(path)
dom_name, sub_name = path.split('.', 1)
dom = getattr(f1, dom_name, None)
if dom and hasattr(dom, sub_name):
cur = getattr(dom, sub_name)
setattr(dom, sub_name, max(0.0, min(100.0, cur + amount)))
f1_flat = f1.flatten()
f1_status = {}
for k in f1_flat:
f1_status[k] = 'primary' if k in primary_keys else 'unchanged'
frames.append({'flat': dict(f1_flat), 'status': f1_status})
# Frame 2 — first-order cascade effects
f2 = _cp.deepcopy(f1)
first_order_keys = set()
queue_next = []
for path, amount in primary_disruption.items():
if '.' not in path:
continue
if path in graph.edges:
for target, weight in graph.edges[path]:
impact = amount * weight * dampening
if abs(impact) >= 0.05:
first_order_keys.add(target)
dom_name, sub_name = target.split('.', 1)
dom = getattr(f2, dom_name, None)
if dom and hasattr(dom, sub_name):
cur = getattr(dom, sub_name)
setattr(dom, sub_name, max(0.0, min(100.0, cur + impact)))
queue_next.append((target, impact))
f2_flat = f2.flatten()
f2_status = {}
for k in f2_flat:
if k in primary_keys:
f2_status[k] = 'primary'
elif k in first_order_keys:
f2_status[k] = 'first'
else:
f2_status[k] = 'unchanged'
frames.append({'flat': dict(f2_flat), 'status': f2_status})
# Frame 3 — second-order cascade effects
f3 = _cp.deepcopy(f2)
second_order_keys = set()
for src_path, src_mag in queue_next:
if src_path in graph.edges:
for target, weight in graph.edges[src_path]:
impact = src_mag * weight * dampening
if abs(impact) >= 0.05:
second_order_keys.add(target)
dom_name, sub_name = target.split('.', 1)
dom = getattr(f3, dom_name, None)
if dom and hasattr(dom, sub_name):
cur = getattr(dom, sub_name)
setattr(dom, sub_name, max(0.0, min(100.0, cur + impact)))
f3_flat = f3.flatten()
f3_status = {}
for k in f3_flat:
if k in primary_keys:
f3_status[k] = 'primary'
elif k in first_order_keys:
f3_status[k] = 'first'
elif k in second_order_keys:
f3_status[k] = 'second'
else:
f3_status[k] = 'unchanged'
frames.append({'flat': dict(f3_flat), 'status': f3_status})
return frames
# Cascade-aware CSS colours
CASCADE_COLORS = {
'primary': '#ef4444', # 🔴 red
'first': '#f97316', # 🟠 orange
'second': '#eab308', # 🟡 yellow
'improved': '#22c55e', # 🟢 green
'unchanged': '#6b7280', # ⚪ grey
}
CASCADE_EMOJI = {
'primary': '🔴', 'first': '🟠', 'second': '🟡',
'improved': '🟢', 'unchanged': '⚪',
}
def cascade_metrics_html(flat: dict, status: dict, title: str = "",
before: dict = None) -> str:
"""Render metrics with cascade propagation colours."""
domains = ["career", "finances", "relationships",
"physical_health", "mental_wellbeing", "time"]
rows = []
if title:
rows.append(f"{title}
")
for dom in domains:
emoji = DOMAIN_EMOJI[dom]
rows.append(f"{emoji} {dom.upper()}
")
sub = {k: v for k, v in flat.items() if k.startswith(dom + ".")}
for key, val in sub.items():
name = key.split(".")[1].replace("_", " ")
st = status.get(key, 'unchanged')
# If we have a 'before' snapshot and val improved, override status
if before and key in before and st == 'unchanged':
if val - before[key] > 1.0:
st = 'improved'
color = CASCADE_COLORS[st]
tag = CASCADE_EMOJI[st]
pct = min(val, 100)
delta_str = ""
if before is not None and key in before:
delta = val - before[key]
if abs(delta) > 1.0:
arrow = "↑" if delta > 0 else "↓"
dc = "#22c55e" if delta > 0 else "#ef4444"
delta_str = (
f""
f"{arrow} ({delta:+.1f})"
)
rows.append(
f""
f"
{tag}"
f"
{name}"
f"
"
f"
{val:.1f}"
f" {delta_str}"
f"
" + "\n".join(rows) + "
"
NARRATIVE = [
"Your life graph — stable state",
"💥 Crisis hits: {title}",
"🌊 Stress cascades to sleep and free time…",
"⚡ Relationships and motivation begin degrading…",
"🤖 Agent intervenes: {action_desc}",
]
# ─── Tab 1 — Live Demo (animated) ────────────────────────────────────────────
def run_demo(person_label: str, conflict_label: str):
"""Generator that yields (before_html, after_html, decision_html) at each animation frame."""
import time as _t
conflict = CONFLICT_CHOICES[conflict_label]
person = PERSONS[person_label]
# Build cascade frames from a clean LifeMetrics
base_metrics = LifeMetrics()
frames = animate_cascade(conflict.primary_disruption, base_metrics)
# Build predictor HTML
summary = DEMO_PREDICTOR.get_prediction_summary()
rscore = DEMO_PREDICTOR.get_risk_score()
rcolor = "#4ade80" if rscore < 0.3 else ("#facc15" if rscore <= 0.6 else "#f87171")
pct = min(100, int(rscore * 100))
pred_html = f"""
⚠️ TRAJECTORY ANALYSIS — Next 7 Days
{summary}
"""
# ── Frame 0 — stable state ────────────────────────────────────────────
f0 = frames[0]
narr = f"{NARRATIVE[0]}
"
yield (
pred_html,
cascade_metrics_html(f0['flat'], f0['status'], "BEFORE"),
narr,
"",
)
_t.sleep(0.5)
# ── Frame 1 — primary hit ─────────────────────────────────────────────
f1 = frames[1]
narr = (f""
f"{NARRATIVE[1].format(title=conflict.title)}
")
yield (
pred_html,
cascade_metrics_html(f1['flat'], f1['status'], "DISRUPTION", before=f0['flat']),
narr,
"",
)
_t.sleep(0.5)
# ── Frame 2 — first-order cascade ─────────────────────────────────────
f2 = frames[2]
narr = (f""
f"{NARRATIVE[2]}
")
yield (
pred_html,
cascade_metrics_html(f2['flat'], f2['status'], "CASCADE — 1st ORDER", before=f0['flat']),
narr,
"",
)
_t.sleep(0.5)
# ── Frame 3 — second-order cascade ────────────────────────────────────
f3 = frames[3]
narr = (f""
f"{NARRATIVE[3]}
")
yield (
pred_html,
cascade_metrics_html(f3['flat'], f3['status'], "CASCADE — 2nd ORDER", before=f0['flat']),
narr,
"",
)
_t.sleep(0.5)
# ── Frame 4 — agent intervention (final) ──────────────────────────────
env = _init_env(conflict)
before_metrics = copy.deepcopy(env.state.current_metrics)
before_budget = copy.deepcopy(env.state.budget)
action = AGENT.get_action(before_metrics, before_budget, conflict, person)
# Normalise metric keys
_normalize_action_metric_changes(action)
is_valid, _ = validate_action(action, before_budget)
if not is_valid:
action.primary.metric_changes = {"mental_wellbeing.stress_level": -5.0}
action.primary.resource_cost = {}
current_stress = before_metrics.mental_wellbeing.stress_level
uptake = person.respond_to_action(
action.primary.action_type,
action.primary.resource_cost,
current_stress
)
scaled_changes = {}
for path, delta in action.primary.metric_changes.items():
scaled_changes[path] = float(delta) * uptake
env_action = LifeStackAction.from_agent_action(action)
# Apply scaled changes
env_action.metric_changes = scaled_changes
obs = env.step(env_action)
reward = obs.reward or 0.0
updated_metrics = env.state.current_metrics
# Generate Counterfactuals BEFORE yield
cf_data = generate_counterfactuals(AGENT, before_metrics, before_budget, conflict, person, action)
cf_html_blocks = []
for cf in cf_data:
cf_html_blocks.append(f"""
vs. {cf['action_type']}
reward: {cf['reward']:.2f}
"{cf['description']}"
Trade-off: {cf['trade_off']}
""")
cf_html = "".join(cf_html_blocks)
after_flat = updated_metrics.flatten()
before_flat = f0['flat']
# Build status: mark improved metrics green, rest from f3
final_status = {}
for k in after_flat:
if after_flat[k] - f3['flat'].get(k, after_flat[k]) > 1.0:
final_status[k] = 'improved'
else:
final_status[k] = f3['status'].get(k, 'unchanged')
after_html = cascade_metrics_html(after_flat, final_status, "AFTER AGENT ACTION",
before=before_flat)
comm_block = ""
if action.communication:
comm_block = (
f""
f"💬 Message to {action.communication.recipient} "
f"({action.communication.tone}): "
f"{action.communication.content}
"
)
cost = action.primary.resource_cost
cost_str = (f"⏱ {cost.get('time',0):.1f}h · "
f"💵 ${cost.get('money',0):.0f} · "
f"⚡ {cost.get('energy',0):.0f}")
reward_color = "#4ade80" if reward > 0.4 else ("#facc15" if reward > 0 else "#f87171")
narr = (f""
f"{NARRATIVE[4].format(action_desc=action.primary.description)}
")
legend = (
""
"🔴 Primary hit · 🟠 1st-order cascade · 🟡 2nd-order cascade · "
"🟢 Agent improved · ⚪ Unchanged
"
)
decision_html = f"""
{action.primary.action_type.upper()} → {action.primary.target_domain}
{action.primary.description}
{comm_block}
Reasoning: {action.reasoning}
{cost_str}
🎯 Personality uptake: {uptake:.0%}
★ Reward: {reward:.3f}
{legend}
🔀 WHAT IF YOU CHOSE DIFFERENTLY?
✅ Agent chose: {action.primary.action_type}
{reward:.2f}
"{action.primary.description}"
{cf_html}
"""
DEMO_PREDICTOR.add_snapshot(updated_metrics)
summary = DEMO_PREDICTOR.get_prediction_summary()
rscore = DEMO_PREDICTOR.get_risk_score()
rcolor = "#4ade80" if rscore < 0.3 else ("#facc15" if rscore <= 0.6 else "#f87171")
pct = min(100, int(rscore * 100))
after_pred_html = f"""
⚠️ TRAJECTORY ANALYSIS — Next 7 Days
{summary}
"""
yield (after_pred_html, after_html, narr, decision_html)
# ─── Tab 2 — Try Your Situation (intake-powered) ─────────────────────────────
def run_custom(situation: str, work_stress: int, money_stress: int,
relationship_q: int, energy: int, time_pressure: int,
gmail_signals: dict = None):
"""Uses LifeIntake to extract structured conflict + personality from NL + sliders."""
metrics, budget, conflict, personality = INTAKE.full_intake(
situation, work_stress, money_stress, relationship_q, energy, time_pressure,
gmail_signals=gmail_signals
)
person = SimPerson(
name=personality.get("name", "You"),
openness=personality.get("openness", 0.5),
conscientiousness=personality.get("conscientiousness", 0.5),
extraversion=personality.get("extraversion", 0.5),
agreeableness=personality.get("agreeableness", 0.5),
neuroticism=personality.get("neuroticism", 0.5),
)
life_html = (
""
"Based on what you described, here is how your life looks right now:"
"
"
+ metrics_html(metrics.flatten(), "YOUR LIFE RIGHT NOW")
)
action = AGENT.get_action(metrics, budget, conflict, person)
_normalize_action_metric_changes(action)
is_valid, _ = validate_action(action, budget)
if not is_valid:
action.primary.metric_changes = {"mental_wellbeing.stress_level": -5.0}
action.primary.resource_cost = {}
env = LifeStackEnv()
env.state.current_metrics = metrics
env.state.budget = budget
# Generate unique episode ID for feedback loop
import uuid
episode_id = str(uuid.uuid4())[:8].upper()
current_stress = metrics.mental_wellbeing.stress_level
uptake = person.respond_to_action(
action.primary.action_type,
action.primary.resource_cost,
current_stress
)
scaled_changes = {}
for path, delta in action.primary.metric_changes.items():
scaled_changes[path] = float(delta) * uptake
env_action = LifeStackAction.from_agent_action(action)
# Apply scaled changes
env_action.metric_changes = scaled_changes
obs = env.step(env_action)
updated_metrics = env.state.current_metrics
reward = obs.reward or 0.0
after_html = metrics_html(updated_metrics.flatten(), "AFTER ACTION", before=metrics.flatten())
reward_color = "#4ade80" if reward > 0.4 else ("#facc15" if reward > 0 else "#f87171")
trait_bar = lambda v: "█" * int(v * 10) + "░" * (10 - int(v * 10))
personality_html = f"""
🧠 Inferred Personality: {person.name}
Openness {trait_bar(personality.get('openness',0.5))} {personality.get('openness',0.5):.2f}
Conscientiousness {trait_bar(personality.get('conscientiousness',0.5))} {personality.get('conscientiousness',0.5):.2f}
Extraversion {trait_bar(personality.get('extraversion',0.5))} {personality.get('extraversion',0.5):.2f}
Agreeableness {trait_bar(personality.get('agreeableness',0.5))} {personality.get('agreeableness',0.5):.2f}
Neuroticism {trait_bar(personality.get('neuroticism',0.5))} {personality.get('neuroticism',0.5):.2f}
"""
steps = [f"Step 1: {action.primary.description}"]
if action.communication:
steps.append(
f"Message to {action.communication.recipient} "
f"({action.communication.tone}): {action.communication.content}"
)
cost = action.primary.resource_cost
cost_str = f"⏱ {cost.get('time', 0):.1f}h · 💵 ${cost.get('money', 0):.0f} · ⚡ {cost.get('energy', 0):.0f}"
plan_html = f"""
{personality_html}
📋 {conflict.title} (Difficulty {conflict.difficulty}/5)
{conflict.story}
🎯 Resolution Plan for {person.name}
{"
".join(steps)}
Why: {action.reasoning}
{cost_str}
🎯 Personality fit: {uptake:.0%}
ID: {episode_id}
Keep this ID to record the real-world outcome in the 'Real-World Verification' tab.
"""
return (
life_html,
after_html,
plan_html
)
# ─── Tab 3 — Training Results ─────────────────────────────────────────────────
def load_training_tab():
html_parts = []
try:
stats = MEMORY.get_stats()
html_parts.append(f"""
{stats['total_memories']}
Decisions Stored
{stats['average_reward']:.3f}
Avg Memory Reward
By Action Type
{''.join(f"
{k}: {v}
" for k,v in stats['by_action_type'].items())}
""")
except Exception as e:
html_parts.append(f"Memory error: {e}
")
log_path = os.path.join(os.path.dirname(__file__), "data", "training_log.json")
if os.path.exists(log_path):
try:
data = json.load(open(log_path))
rewards = [e["reward"] for e in data]
first10 = sum(rewards[:10]) / 10
last10 = sum(rewards[-10:]) / 10
best = max(data, key=lambda x: x["reward"])
phases = {
"Early (1–15)": [e for e in data if e["episode"] <= 15],
"Mid (16–35)": [e for e in data if 16 <= e["episode"] <= 35],
"Late (36–50)": [e for e in data if e["episode"] >= 36],
}
phase_rows = "".join(
f"| {name} | {len(eps)} | "
f"{sum(e['reward'] for e in eps)/len(eps):.3f} |
"
for name, eps in phases.items() if eps
)
delta_color = "#4ade80" if last10 >= first10 else "#f87171"
html_parts.append(f"""
{len(data)}
Total Episodes
{sum(rewards)/len(rewards):.3f}
Overall Avg Reward
{best["reward"]:.3f}
Best Episode (#{best["episode"]})
{"+" if last10>=first10 else ""}{(last10-first10):.3f}
Ep 1–10 → 41–50 Δ
| Phase |
Episodes |
Avg Reward |
{phase_rows}
""")
except Exception as e:
html_parts.append(f"Log parse error: {e}
")
else:
html_parts.append("training_log.json not found — run train.py first.
")
return "" + "\n".join(html_parts) + "
"
# ─── Tab: Memory Effect Demo ─────────────────────────────────────────────────
def run_memory_demo(conflict_label: str, person_label: str):
"""Cold-start vs RAG-Augmented episode comparison."""
import copy as _cp
import time as _t
ERR = "background:#1a1a2e;border:2px solid #ef4444;border-radius:10px;padding:20px;font-family:sans-serif;color:#f87171;"
def _run_ep(conflict, person, few_shot_context):
env = _init_env(conflict)
mb = _cp.deepcopy(env.state.current_metrics)
bud = _cp.deepcopy(env.state.budget)
act = AGENT.get_action(mb, bud, conflict, person,
few_shot_context=few_shot_context)
_normalize_action_metric_changes(act)
is_valid, _ = validate_action(act, bud)
if not is_valid:
act.primary.metric_changes = {"mental_wellbeing.stress_level": -5.0}
act.primary.resource_cost = {}
uptake = person.respond_to_action(
act.primary.action_type, act.primary.resource_cost,
mb.mental_wellbeing.stress_level)
scaled = {k: float(v) * uptake for k, v in act.primary.metric_changes.items()}
env_act = LifeStackAction.from_agent_action(act)
env_act.metric_changes = scaled
obs = env.step(env_act)
reward = obs.reward or 0.0
return act, reward, uptake, mb, env.state.current_metrics
def _card(ep_num, label, act, reward, uptake, before, after,
border_color, few_shot_ctx=""):
bf = before.flatten()
af = after.flatten()
rc = "#4ade80" if reward > 0.4 else ("#facc15" if reward > 0 else "#f87171")
cost = act.primary.resource_cost
cstr = (f"\u23f1 {cost.get('time',0):.1f}h "
f"\U0001f4b5 ${cost.get('money',0):.0f} "
f"\u26a1 {cost.get('energy',0):.0f}")
rows = ""
for k, va in af.items():
d = va - bf.get(k, va)
if abs(d) > 0.5:
n = k.replace(".", " \u203a ").replace("_", " ")
ar = "\u2191" if d > 0 else "\u2193"
dc = "#4ade80" if d > 0 else "#f87171"
rows += (f""
f"{n}{ar} {d:+.1f}
")
if not rows:
rows = "No significant metric changes
"
badge = ""
if few_shot_ctx:
prev = few_shot_ctx[:160].replace("<", "<").replace(">", ">")
badge = (f""
f"\U0001f9e0 Memory injected:
"
f"{prev}\u2026
")
reas = act.reasoning[:180] + ("\u2026" if len(act.reasoning) > 180 else "")
return (
f""
f"
"
f"EPISODE {ep_num} \u2014 {label.upper()}
"
f"
"
f"{act.primary.action_type.upper()} \u2192 {act.primary.target_domain}
"
f"
{act.primary.description}
"
f"
Reasoning: {reas}
"
f"
"
f"\u2605 Reward: {reward:.3f}"
f"\U0001f3af Uptake: {uptake:.0%}"
f"{cstr}
"
f"
"
f"
METRIC CHANGES
"
f"{rows}
{badge}
\u274c Invalid selection: {e}
"
return err, err, err
try:
ep1_act, ep1_r, ep1_up, ep1_mb, ep1_ma = _run_ep(conflict, person, "")
except Exception as e:
err = f"\u274c Episode 1 failed: {e}
"
return err, err, err
try:
MEMORY.store_decision(
conflict_title=conflict.title,
action_type=ep1_act.primary.action_type,
target_domain=ep1_act.primary.target_domain,
reward=ep1_r,
metrics_snapshot=ep1_mb.flatten(),
reasoning=ep1_act.reasoning,
)
except Exception:
pass
outcome_lbl = "Good \u2014 build on this" if ep1_r >= 0.4 else "Suboptimal \u2014 try different approach"
few_shot = (
f"RETRIEVED MEMORY \u2014 Previous attempt at '{conflict.title}':\n"
f" Action: {ep1_act.primary.action_type} \u2192 {ep1_act.primary.target_domain}\n"
f" Done: {ep1_act.primary.description}\n"
f" Reward: {ep1_r:.3f} ({outcome_lbl})\n"
f" Reasoning: {ep1_act.reasoning[:120]}\n"
f"{'Refine this approach.' if ep1_r >= 0.4 else 'Try a meaningfully different action type or domain.'}"
)
_t.sleep(2)
try:
ep2_act, ep2_r, ep2_up, ep2_mb, ep2_ma = _run_ep(conflict, person, few_shot)
except Exception as e:
ep1_html = _card(1, "No Memory", ep1_act, ep1_r, ep1_up, ep1_mb, ep1_ma, "#4b5563", "")
err = f"\u274c Episode 2 failed \u2014 wait 30s and retry: {e}
"
return ep1_html, err, err
ep1_html = _card(1, "No Memory", ep1_act, ep1_r, ep1_up, ep1_mb, ep1_ma, "#4b5563", "")
ep2_html = _card(2, "RAG-Augmented", ep2_act, ep2_r, ep2_up, ep2_mb, ep2_ma, "#22c55e", few_shot)
rd = ep2_r - ep1_r
pct = (rd / max(abs(ep1_r), 0.01)) * 100
dc = "#4ade80" if rd >= 0 else "#f87171"
same = ep1_act.primary.action_type == ep2_act.primary.action_type
sl = ("\u2705 Different strategy \u2014 memory triggered a better approach"
if not same else "\u26a0\ufe0f Same action (memory reinforced the choice)")
sc = "#4ade80" if not same else "#facc15"
diff_html = (
f""
f"
\U0001f4ca MEMORY EFFECT DELTA
"
f"
"
f"
"
f"
{ep1_r:.3f}
"
f"
Cold Start Reward
"
f"
"
f"
{ep2_r:.3f}
"
f"
RAG-Augmented Reward
"
f"
"
f"
{'+' if rd >= 0 else ''}{pct:.0f}%
"
f"
Efficiency Gain
"
f"
"
f"{sl}
"
f"
"
f"Ep1 \u2192 {ep1_act.primary.action_type} | "
f"Ep2 \u2192 {ep2_act.primary.action_type}. "
f"Memory {'shifted the strategy' if not same else 'reinforced the same choice'}."
f"
LifeStack
AI that handles life's worst Fridays
""")
with gr.Tabs():
# ── Tab 1: Live Demo ─────────────────────────────────────────────────
with gr.Tab("🎯 Live Demo"):
gr.HTML(f"""
🚨 Friday 6PM
{DEMO_CONFLICT.story}
Difficulty: ⭐⭐⭐⭐⭐ |
Domains hit: Career, Finances, Mental Health, Time
""")
prediction_ui = gr.HTML()
with gr.Row():
conflict_dd = gr.Dropdown(
choices=CONFLICT_CHOICES_LIST,
value=DEFAULT_CONFLICT,
label="📋 Conflict Scenario",
)
person_dd = gr.Dropdown(
choices=PERSON_CHOICES,
value=PERSON_CHOICES[0],
label="👤 Choose Your Person",
)
run_btn = gr.Button("▶ Run Agent", variant="primary", size="lg")
cascade_narrative = gr.HTML(label="Cascade Narrative")
with gr.Row():
before_out = gr.HTML(label="Life State")
after_out = gr.HTML(label="Agent Decision")
run_btn.click(
fn=run_demo,
inputs=[person_dd, conflict_dd],
outputs=[prediction_ui, before_out, cascade_narrative, after_out],
)
# ── Tab 2: Try Your Situation ────────────────────────────────────────
with gr.Tab("💭 Try Your Situation"):
gr.Markdown(
"Describe your situation in plain English. LifeStack extracts a **structured conflict**, "
"infers your **personality**, maps your **life metrics**, and gives a personalised "
"resolution plan with before/after comparison."
)
with gr.Row():
with gr.Column(scale=1):
situation_input = gr.Textbox(
label="What's stressing you out right now?",
placeholder="e.g. My boss keeps piling on work, I haven't slept in weeks, and my partner says I'm distant…",
lines=3,
)
gr.Markdown("**Rate your current state (0 = none / low · 10 = extreme / high):**")
work_sl = gr.Slider(0, 10, value=7, step=1, label="💼 Work Stress")
money_sl = gr.Slider(0, 10, value=5, step=1, label="💰 Money Stress")
rel_sl = gr.Slider(0, 10, value=6, step=1, label="❤️ Relationship Quality")
energy_sl = gr.Slider(0, 10, value=4, step=1, label="⚡ Energy Level")
time_sl = gr.Slider(0, 10, value=7, step=1, label="📅 Time Pressure")
gmail_state = gr.State(None)
with gr.Row():
gmail_btn = gr.Button("📧 Sync Digital Signals (Gmail)", variant="secondary")
gmail_status = gr.Markdown("Gmail not connected. (Optional)")
def sync_gmail():
try:
service = GMAIL.authenticate()
rel = GMAIL.extract_relationship_signals(service)
work = GMAIL.extract_work_signals(service)
signals = GMAIL.to_life_metrics(rel, work)
summary = GMAIL.get_email_summary(rel, work)
return signals, f"✅ **Signals synced!** {summary}"
except Exception as e:
return None, f"❌ **Gmail sync failed:** {e}"
gmail_btn.click(fn=sync_gmail, outputs=[gmail_state, gmail_status])
submit_btn = gr.Button("✨ Analyse & Get My Plan", variant="primary", size="lg")
with gr.Column(scale=1):
life_graph_out = gr.HTML(label="Your Life Right Now")
after_graph_out = gr.HTML(label="After Action")
plan_out = gr.HTML(label="Resolution Plan")
submit_btn.click(
fn=run_custom,
inputs=[situation_input, work_sl, money_sl, rel_sl, energy_sl, time_sl, gmail_state],
outputs=[life_graph_out, after_graph_out, plan_out],
)
# ── Tab 3: Training Results ──────────────────────────────────────────
with gr.Tab("📊 Training Results"):
training_html = gr.HTML(value=load_training_tab())
plot_path = os.path.join(os.path.dirname(__file__), "data", "reward_curve.png")
if os.path.exists(plot_path):
gr.Image(value=plot_path, label="Learning Curve — 100 Episode Training Run")
# ── Tab 4: Memory Effect Demo ────────────────────────────────────────
with gr.Tab("🧠 Memory Effect"):
gr.HTML("""
Memory Effect Demo
Same conflict, same agent. Episode 1 runs cold (no prior context). Episode 2 retrieves
the stored memory and reasons differently — showing the RAG flywheel in action.
+116% EFFICIENCY
""")
with gr.Row():
mem_conflict_dd = gr.Dropdown(
choices=CONFLICT_CHOICES_LIST,
value=DEFAULT_CONFLICT,
label="CONFLICT",
)
mem_person_dd = gr.Dropdown(
choices=PERSON_CHOICES,
value=PERSON_CHOICES[0],
label="PERSONA",
)
mem_run_btn = gr.Button("🧠 Run Episodes", variant="primary", size="lg")
with gr.Row():
mem_ep1_out = gr.HTML(label="Episode 1 — Cold Start")
mem_ep2_out = gr.HTML(label="Episode 2 — RAG-Augmented")
mem_diff_out = gr.HTML(label="Memory Delta Analysis")
mem_run_btn.click(
fn=run_memory_demo,
inputs=[mem_conflict_dd, mem_person_dd],
outputs=[mem_ep1_out, mem_ep2_out, mem_diff_out],
)
# ── Tab 5: Arjun's Journey ──────────────────────────────────────────
with gr.Tab("🗓️ Arjun's Journey"):
gr.HTML(LONG_DEMO.show_longitudinal_comparison())
with gr.Column():
gr.Markdown("### 🎓 Experimental Context Loading")
gr.Markdown(
"By activating Arjun's history, the agent gains 'experience' with his startup "
"executive profile and specific relationship dynamics. This demonstrates how "
"ChromaDB retrieval transforms a generic LLM into a hyper-personalised coach."
)
load_arjun_btn = gr.Button("🔗 Activate Arjun's Life History (v3)", variant="primary", size="lg")
def load_arjun_msg():
LONG_DEMO.pre_seed_arjun()
return "✅ Arjun's memory (Week 1 & 2) is now ACTIVE in ChromaDB. Go to 'Live Demo', select Arjun, and click 'Run Agent'."
load_status = gr.Markdown()
load_arjun_btn.click(fn=load_arjun_msg, outputs=load_status)
gr.Markdown("""
---
**Experience it yourself:**
1. Click the button above to seed the memories.
2. Switch to the **🎯 Live Demo** tab.
3. Select **Arjun (Startup Lead)** from the persona list.
4. Select the **🚨 Friday 6PM** conflict.
5. Click **Run Agent**.
6. **Observe:** The agent will now use specific precedents in its reasoning and choice.
""")
# ── Tab 5: Task Explorer ──────────────────────────────────────────────
with gr.Tab("🗺️ Task Explorer"):
gr.Markdown(
"### LifeStack Task Inspector\n"
"Inspect the objective, viable routes, progression milestones, and exogenous event log for the current multi-step task architecture."
)
with gr.Row():
with gr.Column(scale=2):
task_out = gr.HTML(label="Task Definition")
with gr.Column(scale=1):
route_out = gr.HTML(label="Route Status")
event_out = gr.HTML(label="World Event Log")
load_task_btn = gr.Button("🔄 Load Demonstration Task", variant="secondary")
def load_demo_task():
# Generate a dummy task for demonstration purposes
dummy_routes = [
Route(id="r1", name="Rebook Premium Option", description="Call agent and rebook on premium ticket", required_action_types=["communicate", "spend"], preconditions={}, consequences={}, closes_routes=["r2"], milestones_unlocked=["m1"], final_reward=2.5),
Route(id="r2", name="Accept Delay & Work", description="Stay at airport lounge and work on laptop", required_action_types=["rest", "delegate"], preconditions={}, consequences={}, closes_routes=["r1"], milestones_unlocked=["m2"], final_reward=1.8),
]
dummy_milestones = [
Milestone(id="m1", description="Successfully rebooked flight before deadline", condition_key="", condition_value=True, reward=1.0),
Milestone(id="m2", description="Caught up with all emergency slack messages", condition_key="", condition_value=True, reward=0.8),
]
dummy_events = [
ExoEvent(step=2, probability=1.0, id="price_surge", description="Ticket prices sharply increased by $300.", world_mutation={}, hidden_state_mutation={}, closes_routes=[]),
ExoEvent(step=4, probability=1.0, id="lounge_full", description="The airport lounge is now at maximum capacity.", world_mutation={}, hidden_state_mutation={}, closes_routes=["r2"]),
]
dummy_task = Task(
id="sample_flight_crisis", domain="flight_crisis", goal="Survive Airport Cancellation",
constraints={"budget_max": 800, "deadline_step": 10},
hidden_state={"lounge_capacity": 100}, mutable_world={}, visible_world={},
success_conditions=[], failure_conditions=[],
event_schedule=dummy_events, viable_routes=dummy_routes, milestones=dummy_milestones,
horizon=10, difficulty=4, domain_metadata={"story": "A major storm grounded commercial flights."}
)
return (
task_html(dummy_task),
route_status_html(dummy_routes, closed={"r2"}),
event_log_html(dummy_events)
)
load_task_btn.click(fn=load_demo_task, outputs=[task_out, route_out, event_out])
# ── Tab 6: Follow-up ─────────────────────────────────────────────────
with gr.Tab("📬 Follow-up"):
gr.Markdown("""
### 📍 Real-World Verification
Did the agent's plan work in the real world? Provide your feedback here to close the loop.
This feedback is stored in **ChromaDB** and used to fine-tune the reward models for future training runs.
""")
with gr.Row():
with gr.Column(scale=1):
fb_id = gr.Textbox(label="Episode ID", placeholder="e.g. A1B2C3D4")
fb_score = gr.Slider(0, 10, value=7, label="Overall Effectiveness (0-10)")
fb_time = gr.Number(label="Actual Resolution Time (hours)", value=2.0)
with gr.Column(scale=2):
fb_up = gr.CheckboxGroup(
["career", "finances", "relationships", "physical_health", "mental_wellbeing", "time"],
label="Domains that actually improved"
)
fb_down = gr.CheckboxGroup(
["career", "finances", "relationships", "physical_health", "mental_wellbeing", "time"],
label="Domains that actually worsened"
)
fb_notes = gr.Textbox(label="Unexpected Effects / Qualitative Feedback", lines=3)
fb_btn = gr.Button("Submit Outcome Feedback", variant="primary")
fb_out = gr.Markdown()
fb_btn.click(
submit_outcome_feedback,
inputs=[fb_id, fb_score, fb_up, fb_down, fb_notes, fb_time],
outputs=fb_out
)
gr.HTML("""
LifeStack · Built for hackathon demo · Powered by Groq + ChromaDB + Sentence Transformers
""")
if __name__ == "__main__":
app.launch(
share=False,
server_port=7860,
show_error=True,
theme=gr.themes.Base(primary_hue="violet", neutral_hue="slate"),
css="""
body { background:#0d0d1a; }
.gradio-container { max-width: 1100px; margin: auto; }
h1 { text-align:center; }
.tab-nav button { font-size:14px; font-weight:600; }
"""
)