"""Custom Gradio Blocks UI (Off-Brand).
Two surfaces:
- Schedule: paste/inspect a thread (+images, +existing calendar), watch the model
"think" live, then review/edit the ActionPlan, download .ics, optional GCal.
- Activity: a live dashboard of what the agent and model are doing — pipeline
stepper, metric tiles, event timeline, activity chart, and per-run traces.
"""
from __future__ import annotations
import base64
import html
import json
import os
import time
from datetime import timedelta
from pathlib import Path
from urllib.parse import urlencode
import gradio as gr
import pandas as pd
from dateutil import parser as dtparser
from calendar_out.ics import write_ics
from server import events as bus
from server import impact, memory
from server.imageutil import paths_to_data_uris
from server.schema import ActionPlan, Event
CSS = (Path(__file__).parent.parent / "static" / "app.css").read_text(encoding="utf-8")
# Brand logo (robot-in-a-chat-bubble holding a calendar), inlined as a data URI
# so it renders identically under every serve mode (mounted uvicorn / launch).
_LOGO_URI = "data:image/png;base64," + base64.b64encode(
(Path(__file__).parent.parent / "static" / "logo.png").read_bytes()
).decode("ascii")
# Client-side carousel: auto-advance + arrows + dots + pause-on-hover. A
# MutationObserver re-initialises when Gradio swaps the gr.HTML after an analyze.
# Passed to mount_gradio_app/launch (Gradio 6 ignores js set on Blocks when mounted).
CAROUSEL_JS = """
() => {
function init() {
document.querySelectorAll('.carousel:not([data-cz])').forEach(function (cz) {
cz.setAttribute('data-cz', '1');
var slides = [].slice.call(cz.querySelectorAll('.cz-slide'));
var dots = [].slice.call(cz.querySelectorAll('.cz-dot'));
if (slides.length <= 1) return;
var idx = 0;
slides.forEach(function (s, j) { if (s.classList.contains('is-active')) idx = j; });
var ms = parseInt(cz.getAttribute('data-interval') || '4500', 10);
var timer = null;
function show(i) {
idx = (i + slides.length) % slides.length;
slides.forEach(function (s, j) { s.classList.toggle('is-active', j === idx); });
dots.forEach(function (d, j) { d.classList.toggle('is-active', j === idx); });
}
function start() {
if (window.matchMedia && window.matchMedia('(prefers-reduced-motion: reduce)').matches) return;
stop(); timer = setInterval(function () { show(idx + 1); }, ms);
}
function stop() { if (timer) { clearInterval(timer); timer = null; } }
cz.querySelectorAll('.cz-arrow').forEach(function (a) {
a.addEventListener('click', function () { show(idx + parseInt(a.getAttribute('data-dir'), 10)); start(); });
});
dots.forEach(function (d) {
d.addEventListener('click', function () { show(parseInt(d.getAttribute('data-go'), 10)); start(); });
});
cz.addEventListener('mouseenter', stop);
cz.addEventListener('mouseleave', start);
start();
});
wireNav();
wireAnchors();
wireFaq();
wireCounter();
wireGcal();
wireFooter();
wirePasteRun();
wireModeTheme();
wirePipeClock();
}
// Theme the tool card for the device's chosen workflow. The page-load part
// reads localStorage (set-once: don't clobber a live switch); the listener
// part recolors INSTANTLY at click time, so the theme never waits for the
// server round-trip (which only swaps step content/visibility).
function wireModeTheme() {
var tc = document.getElementById('tool-card');
if (!tc) return;
if (!tc.getAttribute('data-mode')) {
var m = '';
try { m = localStorage.getItem('offgrid_mode') || ''; } catch (e) {}
tc.setAttribute('data-mode', m.indexOf('Online') >= 0 ? 'online' : 'offline');
}
document.querySelectorAll('#mode-toggle input[type="radio"]:not([data-mt])').forEach(function (r) {
r.setAttribute('data-mt', '1');
r.addEventListener('change', function () {
tc.setAttribute('data-mode', (r.value || '').indexOf('Online') >= 0 ? 'online' : 'offline');
});
});
}
// Pasting IS the intent — run the analysis automatically on paste so the
// user never has to click "Find the events". The button stays as fallback.
function wirePasteRun() {
var ta = document.querySelector('#rv-textbox textarea');
if (!ta || ta.getAttribute('data-pasterun')) return;
ta.setAttribute('data-pasterun', '1');
ta.addEventListener('paste', function () {
setTimeout(function () {
if ((ta.value || '').trim().length < 12) return;
var btn = document.querySelector('#rv-analyze button');
if (btn) btn.click();
}, 250);
});
}
// Fold Gradio's built-in footer (Use via API · Built with Gradio · Settings)
// into the custom dark banner. appendChild MOVES the node, so Gradio's click
// handlers (API docs / Settings modals) keep working; the MutationObserver
// re-runs this if Gradio ever re-renders the footer.
function wireFooter() {
var gf = document.querySelector('.gradio-container footer:not(#site-footer)');
var host = document.querySelector('#site-footer .footer-inner');
if (gf && host && !host.contains(gf)) host.appendChild(gf);
}
// Plain in-page anchors ("See it in action", footer CTA): native #hash jumps
// are unreliable inside the Gradio SPA, so scroll explicitly like the nav does.
function wireAnchors() {
document.querySelectorAll('a[href^="#"]:not([data-tab]):not([data-az])').forEach(function (a) {
a.setAttribute('data-az', '1');
a.addEventListener('click', function (e) {
var id = (a.getAttribute('href') || '').slice(1);
if (!id) return;
var el = document.getElementById(id);
if (!el) return;
e.preventDefault();
el.scrollIntoView({ behavior: 'smooth', block: 'start' });
});
});
}
// Per-user Google Calendar: open the OAuth popup; persist + restore the
// connection. The token JSON lives ONLY in localStorage; it carries a
// refresh_token (the server refreshes at push time), so once connected the
// visitor stays connected across visits — never asked again. mark() re-runs
// on every init()/re-render, restoring the ✓ state on page load.
function wireGcal() {
function tokenInfo() {
try {
var raw = localStorage.getItem('gcal_token');
if (!raw) return null;
var t = JSON.parse(raw);
if (t.refresh_token) return t; // server can refresh
if (t.expiry && new Date(t.expiry) > new Date()) return t; // still valid
localStorage.removeItem('gcal_token'); // stale + unrefreshable
return null;
} catch (e) { return null; }
}
// Verification state lives on window: wireGcal re-runs with a fresh
// closure on every MutationObserver tick, and mark() must not downgrade
// a server-verified connection on re-render.
// __gcalVerify: '' (unchecked) | 'ok' (Google answered) | 'bad' (token dead)
// __gcalCheck: once-per-page-load guard for the /oauth2/check fetch
// Write textContent ONLY when it actually changes: an identical-value
// write still replaces the text node, which is a childList mutation —
// and the body MutationObserver re-runs init() (→ mark()) on every
// childList mutation, so an unconditional write here is an infinite
// init↔mark feedback loop that pegs the main thread ("This page is
// slowing down Firefox").
function setTxt(el, v) { if (el.textContent !== v) el.textContent = v; }
function mark() {
var v = window.__gcalVerify || '';
var ok = !!tokenInfo() && v !== 'bad';
var label = ok ? (v === 'ok' ? ' ✓ connected · verified' : ' ✓ connected') : '';
document.querySelectorAll('.cal-provider[data-provider="google"]').forEach(function (p) {
p.classList.toggle('is-connected', ok);
});
document.querySelectorAll('.gcal-state').forEach(function (s) { setTxt(s, label); });
document.querySelectorAll('.gcal-badge').forEach(function (b) {
setTxt(b, ok ? 'Google:' + label : 'Google: not connected');
b.classList.toggle('is-on', ok);
});
}
// Round-trip the stored token to Google once per page load: the local
// shape check can't see a revoked/garbage token. Only a DEFINITIVE "no"
// clears the token; transient failures (env unset, network, launch-mode
// Space where /oauth2/check isn't served) keep the optimistic state.
function verify() {
var t = null;
try { t = localStorage.getItem('gcal_token'); } catch (e) {}
if (!t || window.__gcalCheck) return;
window.__gcalCheck = true;
fetch('/oauth2/check', { method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ token: t }) })
.then(function (r) { if (r.status !== 200) throw 0; return r.json(); })
.then(function (d) {
if (d.ok) {
window.__gcalVerify = 'ok';
if (d.token) { try { localStorage.setItem('gcal_token', d.token); } catch (e) {} }
} else if (!d.transient) {
window.__gcalVerify = 'bad';
try { localStorage.removeItem('gcal_token'); } catch (e) {}
}
mark();
})
.catch(function () { /* indeterminate: keep shape-check state */ });
}
mark();
verify();
document.querySelectorAll('.gcal-connect:not([data-g])').forEach(function (a) {
a.setAttribute('data-g', '1');
a.addEventListener('click', function (e) {
e.preventDefault();
window.open('/oauth2/start', 'gcal_oauth', 'width=520,height=680');
});
});
document.querySelectorAll('.gcal-disconnect:not([data-g])').forEach(function (a) {
a.setAttribute('data-g', '1');
a.addEventListener('click', function (e) {
e.preventDefault();
try { localStorage.removeItem('gcal_token'); } catch (_) {}
window.__gcalVerify = '';
mark();
});
});
if (!window.__gcalMsg) {
window.__gcalMsg = true;
window.addEventListener('message', function (e) {
if (e.origin !== location.origin || !e.data || !e.data.gcal_token) return;
try { localStorage.setItem('gcal_token', e.data.gcal_token); } catch (_) {}
// freshly minted token: reset and re-verify it
window.__gcalCheck = false;
window.__gcalVerify = '';
mark();
verify();
});
}
}
// Live character counter for the paste box.
function wireCounter() {
var ta = document.querySelector('#rv-textbox textarea');
var out = document.getElementById('rv-counter');
if (!ta || !out || ta.getAttribute('data-counter')) return;
ta.setAttribute('data-counter', '1');
function upd() { out.textContent = (ta.value ? ta.value.length : 0) + ' / 12000'; }
ta.addEventListener('input', upd);
upd();
}
// Make the grouped nav banner drive the (hidden) Gradio tabs.
function wireNav() {
function activate(name) {
var all = [].slice.call(document.querySelectorAll('button[role="tab"], .tab-nav button'));
// Prefer the real interactive buttons over Gradio's hidden measuring clones.
var vis = all.filter(function (b) { return !b.closest('.visually-hidden'); });
for (var i = 0; i < vis.length; i++) {
if ((vis[i].textContent || '').trim() === name) { vis[i].click(); return true; }
}
// Mobile fallback: a narrow viewport pushes the later tabs (Memory / Feed /
// Submission, index >= 3) into Gradio's collapsed, .visually-hidden overflow
// — absent from `vis`, so those nav links went dead while Home/Activity
// (which stay visible) worked. Click every matching button in the FULL set:
// the real (hidden) tab button switches the tab; the clones are no-ops.
var hit = false;
for (var j = 0; j < all.length; j++) {
if ((all[j].textContent || '').trim() === name) { all[j].click(); hit = true; }
}
return hit;
}
document.querySelectorAll('#site-nav [data-tab]:not([data-nav])').forEach(function (a) {
a.setAttribute('data-nav', '1');
a.addEventListener('click', function (e) {
e.preventDefault();
activate(a.getAttribute('data-tab'));
var anchor = a.getAttribute('data-anchor');
if (anchor) {
setTimeout(function () {
var el = document.querySelector(anchor);
if (el) el.scrollIntoView({ behavior: 'smooth', block: 'start' });
}, 160);
} else {
setTimeout(function () { window.scrollTo({ top: 0, behavior: 'smooth' }); }, 60);
}
if (document.activeElement && document.activeElement.blur) document.activeElement.blur();
});
});
}
// FAQ: tab switching + client-side search filter.
// While a search query is active, both panels are shown so cross-tab
// matches are visible; clearing the search restores the active tab.
function wireFaq() {
document.querySelectorAll('.lp-faq-section:not([data-faq])').forEach(function (sec) {
sec.setAttribute('data-faq', '1');
var tabs = [].slice.call(sec.querySelectorAll('.lp-faq-tab'));
var lists = [].slice.call(sec.querySelectorAll('.lp-faq-list'));
var empty = sec.querySelector('.lp-faq-empty');
var search = sec.querySelector('#faq-search');
var query = '';
function activeTab() {
for (var i = 0; i < tabs.length; i++) {
if (tabs[i].classList.contains('is-active')) return tabs[i].getAttribute('data-tab');
}
return tabs[0] && tabs[0].getAttribute('data-tab');
}
function render() {
var active = activeTab();
var anyVisible = false;
lists.forEach(function (l) {
var show = query ? true : (l.getAttribute('data-panel') === active);
l.classList.toggle('is-hidden', !show);
if (show) {
l.querySelectorAll('.lp-faq-item').forEach(function (it) {
var match = !query || (it.textContent || '').toLowerCase().indexOf(query) !== -1;
it.style.display = match ? '' : 'none';
if (match) anyVisible = true;
if (!match && it.open) it.open = false;
});
}
});
if (empty) empty.classList.toggle('is-hidden', !query || anyVisible);
}
tabs.forEach(function (t) {
t.addEventListener('click', function () {
tabs.forEach(function (o) { o.classList.toggle('is-active', o === t); });
render();
});
});
if (search) {
search.addEventListener('input', function (e) {
query = ((e.target && e.target.value) || '').trim().toLowerCase();
render();
});
}
render();
});
}
// Live elapsed clock for the pipeline card. ONE global interval + a
// window-level start timestamp, because Gradio replaces the trace DOM on
// every generator yield — per-element timers would die each re-render.
function wirePipeClock() {
var running = document.querySelector('.pipe-card[data-state="running"]');
if (running) {
if (!window.__pipeStart) window.__pipeStart = Date.now();
if (!window.__pipeTick) {
window.__pipeTick = setInterval(function () {
var s = ((Date.now() - (window.__pipeStart || Date.now())) / 1000).toFixed(1) + 's';
document.querySelectorAll('.pipe-card[data-state="running"] .pipe-clock')
.forEach(function (c) {
// nodeValue mutates the EXISTING text node (characterData) —
// the body MutationObserver below is childList-only, so the
// 100ms tick never re-fires init().
if (c.firstChild) { c.firstChild.nodeValue = s; } else { c.textContent = s; }
});
}, 100);
}
} else {
// done/error/idle: stop ticking and leave the server-rendered Speed
// value alone; reset so the next run starts from zero.
window.__pipeStart = null;
if (window.__pipeTick) { clearInterval(window.__pipeTick); window.__pipeTick = null; }
}
}
init();
// Coalesce re-init: Gradio replaces whole subtrees on every streamed yield,
// and a raw MutationObserver(init) would run the (expensive) full re-wire
// once per mutation batch — and turn any unguarded DOM write inside init()
// into a tight feedback loop. One trailing 60ms timer caps re-wiring at
// ~16/s no matter what mutates; newly rendered elements get wired on the
// next pass, which is imperceptible.
var reinitPending = false;
new MutationObserver(function () {
if (reinitPending) return;
reinitPending = true;
setTimeout(function () { reinitPending = false; init(); }, 60);
}).observe(document.body, { childList: true, subtree: true });
}
"""
THEME = gr.themes.Base(
primary_hue=gr.themes.colors.purple,
secondary_hue=gr.themes.colors.cyan,
neutral_hue=gr.themes.colors.slate,
)
STAGE_COLORS = {
"ingest": "#22d3ee",
"vision": "#a78bfa",
"model": "#f59e0b",
"decision": "#34d399",
"conflict": "#f87171",
"calendar": "#8b5cf6",
}
# A realistic class group-chat so a first-time visitor sees value in one tap,
# even in stub mode (adapted from a seed in training/make_dataset.py).
SAMPLE_THREAD = (
"Room parent (3rd grade chat): Reminder — school picture day is this "
"Thursday at 9am, kids wear the green class shirt!\n"
"Coach Dana: Also heads up, soccer practice moves to Tuesday 5pm this week.\n"
"Me: thanks! adding both so I don't forget"
)
def _load_sample() -> str:
return SAMPLE_THREAD
# --------------------------------------------------------------------------- #
# Schedule tab helpers
# --------------------------------------------------------------------------- #
def _events_to_rows(events: list[Event]) -> list[list]:
return [
[e.title, e.start, e.end or "", e.location or "", e.reminder_minutes or ""]
for e in events
]
def _rows_to_events(rows) -> list[Event]:
out = []
for r in rows or []:
if not r or not r[0]:
continue
out.append(
Event(
title=r[0],
start=r[1],
end=(r[2] or None),
location=(r[3] or None),
reminder_minutes=int(r[4]) if str(r[4]).strip() else None,
)
)
return out
def _fmt_when(start: str, end: str | None = None) -> str:
"""Human-friendly time range from ISO strings; falls back to the raw value."""
try:
s = dtparser.isoparse(start)
day = s.strftime("%a, %b ") + str(s.day)
t1 = s.strftime("%I:%M %p").lstrip("0")
if end:
e = dtparser.isoparse(end)
return f"{day} · {t1} – {e.strftime('%I:%M %p').lstrip('0')}"
return f"{day} · {t1}"
except Exception: # noqa: BLE001
return start or ""
def _plan_markdown(plan: ActionPlan) -> str:
"""Render the plan summary as HTML (shown in a gr.HTML). Keeps the ⚠️ marker so
``_count_conflicts`` can still tally conflicts from the rendered string."""
parts = []
if plan.reasoning:
parts.append(f'
{html.escape(plan.reasoning)}
')
if plan.conflicts:
badges = "".join(
f'⚠️ {html.escape(c.severity)} · '
f'{html.escape(c.clashes_with)}'
for c in plan.conflicts
)
parts.append(f'
{badges}
')
if plan.proposed_times:
chips = "".join(
f'{html.escape(_fmt_when(t))}'
for t in plan.proposed_times
)
parts.append(
f'
Free slots{chips}
'
)
if plan.needs_clarification:
parts.append(f'
❓ {html.escape(plan.needs_clarification)}
')
if not parts:
return '
✓ All clear — no conflicts.
'
return f'
{"".join(parts)}
'
def _event_meta(e: Event) -> str:
"""A • bullet-separated meta line (Netflix-style), location/reminder."""
bits = []
if e.location:
bits.append(f"📍 {html.escape(e.location)}")
if e.reminder_minutes:
bits.append(f"⏰ {int(e.reminder_minutes)} min before")
return f'
"
)
def _carousel_html(slides: list[str], interval: int = 9000) -> str:
"""Wrap slides in an auto-rotating carousel. First slide is active by default so
it shows even if JS is unavailable; CAROUSEL_JS wires advance + arrows + dots."""
if not slides:
return ""
track = "".join(
s.replace('class="cz-slide', 'class="cz-slide is-active', 1) if i == 0 else s
for i, s in enumerate(slides)
)
multi = len(slides) > 1
arrows = (
''
''
) if multi else ""
dots = (
'
'
+ "".join(
f''
for i in range(len(slides))
)
+ "
"
) if multi else ""
return (
f'
'
f'
{track}
{arrows}{dots}
'
)
def _img_slide(bg_class: str, kicker: str, title: str, sub: str) -> str:
"""A showcase slide with an SVG illustration background + dark scrim + white text."""
return (
f'
'
''
'
'
f'
▍{html.escape(kicker)}
'
f'
{html.escape(title)}
'
f'
{html.escape(sub)}
'
"
"
)
def _idle_carousel() -> str:
"""Showcase slides (visible on load) — 4 image-backed cards selling impact + use."""
return _carousel_html([
_img_slide("cz-bg-chat", "Never miss a school event",
"Paste the class group chat",
"Every date & time becomes a calendar event — automatically."),
_img_slide("cz-bg-flyer", "Snap it, don’t type it",
"A flyer or screenshot works too",
"It reads the date, time & place right from the image."),
_img_slide("cz-bg-cal", "No more double-booking",
"Checked against your calendar",
"Catches clashes and suggests free times before you commit."),
_img_slide("cz-bg-reply", "Reply in one tap",
"A response, written for you",
"RSVP and confirm without drafting a thing."),
_img_slide("cz-bg-carpool", "Carpool, sorted",
"Drop the carpool thread",
"Pickup & drop-off times — and whose turn to drive — land on your calendar."),
_img_slide("cz-bg-appt", "Don’t miss the appointment",
"Forward the reminder text",
"Doctor, dentist & vet visits become events — with a heads-up before."),
_img_slide("cz-bg-party", "RSVP before it slips",
"Drop the party invite",
"The party, the RSVP deadline, and a one-tap reply — all handled."),
])
def _render_event_carousel(rows) -> str:
"""Rotating hero carousel of the found events — one per slide."""
events = _rows_to_events(rows)
if not events:
return _carousel_html([
_slide("No events", "Nothing to schedule yet",
"Paste a chat with a date & time, or tap “Try a sample”.")
])
n = len(events)
slides = [
_event_slide(e, "Next up" if i == 0 else f"{i + 1} of {n}")
for i, e in enumerate(events)
]
return _carousel_html(slides)
def _provider_row(provider: str) -> str:
"""One row of the unified 'Connect your calendar' block. Google is the only
provider with real OAuth (one click, token kept in the visitor's browser);
Outlook and Apple already work with no sign-in via the per-event quick-add
links and the .ics download, so their rows just say where to find that."""
if provider == "google":
return (
'
'
'📅 Google Calendar'
'🔗 Connect'
''
'disconnect'
'one click — stays connected '
"in this browser"
'switch to ☁️ Online to '
"enable
"
)
if provider == "microsoft":
return (
'
'
'📅 Microsoft Outlook'
'one-click “Add to Outlook” links appear on '
"every event found — no sign-in needed
"
)
return (
'
'
'📅 Apple Calendar'
'use “Add to Apple Calendar” on any event or the '
".ics download — opens straight in Calendar
"
)
def _gcal_badge_html() -> str:
"""Tiny Google connection badge for the export toolbar — wireGcal()'s mark()
rewrites its text/class from the (server-verified) connection state, so the
user sees whether 'Add to Google Calendar' will work BEFORE clicking it."""
return (
'
'
'Google: not connected
'
)
def _connect_block_html() -> str:
"""The unified provider block in Step 2a (Google row mirrored on the Agent
tab). wireGcal() restores the connected state from localStorage on page
load, so a connected visitor is never asked to connect again."""
return (
'
Your Google token stays in this browser — '
"never on our server.
"
)
def _quick_add_links(e: Event) -> str:
"""Per-event one-click 'add to calendar' hyperlinks — THE single export
surface: prefilled Google / Outlook template URLs (one click + Save, no
OAuth), plus an iCal link that delivers a single-event .ics (Apple
Calendar, Outlook desktop, and most calendar apps; no prefill URL exists
for those)."""
try:
s = dtparser.isoparse(e.start)
end = dtparser.isoparse(e.end) if e.end else s + timedelta(hours=1)
except Exception: # noqa: BLE001 unparseable date -> no links
return ""
fmt = "%Y%m%dT%H%M%S"
g = "https://calendar.google.com/calendar/render?" + urlencode({
"action": "TEMPLATE", "text": e.title or "Event",
"dates": f"{s.strftime(fmt)}/{end.strftime(fmt)}",
"details": e.notes or "", "location": e.location or "",
})
o = "https://outlook.live.com/calendar/0/action/compose?" + urlencode({
"subject": e.title or "Event", "startdt": s.isoformat(),
"enddt": end.isoformat(), "body": e.notes or "", "location": e.location or "",
})
try:
from calendar_out.ics import events_to_ics
a = ("data:text/calendar;base64,"
+ base64.b64encode(events_to_ics([e])).decode("ascii"))
ical = (f' · iCal'
f' · .ics')
except Exception: # noqa: BLE001
ical = ""
return (
'
"
)
def _render_event_cards(rows, mode: str = "", notes=None) -> str:
"""The condensed event list (plain cards, no billboard). Every card carries
the quick-add hyperlinks in both modes. (``mode`` stays in the signature —
all existing wiring passes it.)
``notes`` is the per-event notes list captured at analysis time (the
``found_notes`` State). Notes aren't a Dataframe column, so rows alone
can't carry the arrival context ("Appointment at 10:30; arrive 15 min
early") — this card is the one place the user sees it. Notes are matched
to events by position and dropped if the counts diverge (a row was added
or removed in the editor), so a stale note never lands on the wrong event.
"""
events = _rows_to_events(rows)
if not events:
return (
'
No events found yet — tap '
"Try a sample, or paste a chat with a date & time.
"
)
notes = list(notes or [])
aligned = len(notes) == len(events)
cards = []
for i, e in enumerate(events):
note_txt = e.notes or (notes[i] if aligned else "") or ""
loc = (f'
"
)
n = len(events)
# 2+ events: one tap for everything — an all-events .ics in the same link
# style (Google/Outlook have no multi-event prefill URL, iCal does it all).
all_link = ""
if n > 1:
try:
from calendar_out.ics import events_to_ics
a = ("data:text/calendar;base64,"
+ base64.b64encode(events_to_ics(events)).decode("ascii"))
all_link = (f''
f"⬇ iCal — all {n} events")
except Exception: # noqa: BLE001
all_link = ""
head = (f'
{n} event{"" if n == 1 else "s"} found'
f"{all_link}
")
return f'
{head}
{"".join(cards)}
'
def _load_busy(cal_path):
# Self-hosted convenience: a locally hosted agent/install can point
# CAL_ICS_PATH at an exported calendar file so step 4 completes itself —
# an explicit upload always takes precedence. Unset on the cloud Space.
if not cal_path:
cal_path = os.environ.get("CAL_ICS_PATH", "").strip() or None
if not cal_path:
return []
try:
from calendar_out.freebusy import load_ics_busy
with open(cal_path, "rb") as f:
return load_ics_busy(f.read())
except Exception: # noqa: BLE001
return []
def _on_analyze(conversation: str, cal_path, image_paths, mem_json: str = ""):
"""Generator: run the SAME agentic workflow as the Agent tab (the MiniCPM/MCP
orchestrator) behind the homepage's existing UX — the step trace streams into
the "watch it think" panel, then the usual events / conflicts / reply fill in.
Yields a 7-tuple: (trace, table rows, plan html, reply, status, notes,
pipeline) — notes feeds the ``found_notes`` State (the 5-column table can't
carry per-event notes); the trailing pipeline HTML feeds the tracker card
that lives ABOVE the status line, outside the trace accordion. ``mem_json``
is the user's per-device (localStorage) memory."""
import base64 as _b64
from server.orchestrator import run_orchestrator
# Calendar: explicit upload wins; CAL_ICS_PATH keeps the self-hosted
# auto-calendar behavior (step 4 completes itself locally).
if not cal_path:
cal_path = os.environ.get("CAL_ICS_PATH", "").strip() or None
ics_b64 = None
if cal_path:
try:
ics_b64 = _b64.b64encode(Path(cal_path).read_bytes()).decode("ascii")
except Exception: # noqa: BLE001
ics_b64 = None
images = paths_to_data_uris(image_paths or [])
memory_block = memory.facts_to_recall(_facts_load(mem_json)) or None
# Pipeline-card meta: evidence reflects what's ACTUALLY read (cal_path is
# post-fallback here); elapsed at the final yield is the Speed chip.
t0 = time.monotonic()
evidence = _evidence_label(conversation, image_paths, cal_path)
_meta = lambda: {"elapsed": time.monotonic() - t0, "evidence": evidence} # noqa: E731
steps: list[dict] = []
final = None
for step in run_orchestrator(conversation or "", ics_b64=ics_b64,
memory_block=memory_block, images=images or None):
steps.append(step)
if step.get("kind") == "final":
final = step
yield (_render_agent_trace(steps), gr.update(), gr.update(), gr.update(),
"🤖 the agent is working…", gr.update(),
_render_pipeline_card(steps, _meta()))
if not final:
yield (_render_agent_trace(steps), gr.update(), gr.update(), gr.update(),
"The agent didn't reach a final step.", gr.update(),
_render_pipeline_card(steps, _meta()))
return
plan_dict = final.get("plan") or {}
try:
plan = ActionPlan(**plan_dict)
except Exception: # noqa: BLE001 defensive: dict drift -> empty plan
plan = ActionPlan(reasoning="", events=[], conflicts=[], proposed_times=[],
reply_draft="", needs_clarification=None)
# The per-event card links (the one export surface) are plain anchors —
# Gradio can't hook their clicks — so a capture is recorded when the run
# surfaces events, not on an export button.
if plan.events:
impact.record_capture(len(plan.events), conflicts_caught=len(plan.conflicts))
yield (
_render_agent_trace(steps),
_events_to_rows(plan.events),
_plan_markdown(plan),
plan.reply_draft,
f"Found {len(plan.events)} event(s) ({len(images)} image(s) read), "
f"{len(plan.conflicts)} conflict(s).",
[e.notes or "" for e in plan.events],
_render_pipeline_card(steps, _meta()),
)
def _count_conflicts(plan_md: str) -> int:
"""Conflicts the user just saw, read back from the rendered plan markdown
(one ⚠️ per conflict in ``_plan_markdown``). Lets the export handlers record
conflicts_caught without widening ``_on_analyze``'s output tuple."""
return (plan_md or "").count("⚠️")
def _on_start_over():
"""Clear inputs + hide results — the 'Start over' button."""
return (
"", # conversation
None, # images_in
None, # existing_cal
gr.update(visible=False), # results
"", # status
"", # screenshot_status
"", # trace_status
_render_pipeline_card([]), # pipeline tracker back to idle
)
def _screenshot_status(files) -> str:
"""Contextual hint shown when screenshot(s) are attached."""
n = len(files) if files else 0
if not n:
return ""
plural = "s" if n != 1 else ""
return f'
📎 {n} screenshot{plural} attached — it’ll be read for events.
'
def _on_share_trace(checked: bool, rows) -> str:
"""Privacy-safe trace toggle. When on, record a REDACTED trace locally
(events only — never the raw chat text) and emit an activity event. Does
not publish to the Hub; that stays in training/share_trace.py."""
if not checked:
return ""
events = _rows_to_events(rows)
trace = {
"events": [
{"title": e.title, "start": e.start, "end": e.end, "location": e.location}
for e in events
],
"n_events": len(events),
}
try:
path = Path(os.environ.get("TRACE_PATH", "/tmp/offgrid_traces.jsonl"))
with path.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(trace) + "\n")
except Exception: # noqa: BLE001 local logging is best-effort
pass
bus.emit("decision", f"privacy-safe trace recorded ({len(events)} event(s))", events=len(events))
return ('
✓ Privacy-safe trace recorded '
"(events only; raw text not stored).
")
def _prep_ics(rows):
"""Pre-generate the .ics the moment analysis finishes so the file widget is
a ready download link. Deliberately does NOT record impact — a capture is
only counted on an explicit export action (_on_make_ics / gcal push)."""
events = _rows_to_events(rows)
if not events:
return gr.update()
return write_ics(events)
# --------------------------------------------------------------------------- #
# Agent tab: MiniCPM-planned orchestrator over the Space's own MCP tools
# --------------------------------------------------------------------------- #
_STEP_ICONS = {"plan": "🧠", "tool_call": "🔧", "tool_result": "📥",
"final": "✅", "error": "⚠️"}
# The pipeline card's six stages: (key, tile label, running caption).
# Keys map onto the orchestrator's real step stream — capture/read on the plan
# step, the middle three on their MCP tool calls, reply on the final step.
_PIPE_STAGES = [
("capture", "Capture", "Reading your thread…"),
("read", "Read", "Planning the work…"),
("extract", "Extract", "Extracting events…"),
("conflicts", "Conflicts", "Checking your calendar…"),
("export", "Export", "Writing your .ics…"),
("reply", "Reply", "Drafting a reply…"),
]
_PIPE_TOOL_STAGE = {"extract_events": "extract", "check_conflicts": "conflicts",
"make_ics": "export"}
def _pipeline_stage_states(steps: list[dict]) -> tuple[dict[str, str], str]:
"""Derive per-stage states (todo/active/done/skip/error) plus the card state
(idle/running/done/error) purely from the orchestration step stream. The LLM
planner may interleave extra plan steps, "(observation)" results, and an
error step followed by the scripted fallback replay — only a TRAILING error
renders the error state."""
if not steps:
return {k: "todo" for k, *_ in _PIPE_STAGES}, "idle"
called, resulted = set(), set()
seen_plan = has_final = False
for s in steps:
kind, tool = s.get("kind"), s.get("tool", "")
if kind == "plan":
seen_plan = True
elif kind == "tool_call" and tool in _PIPE_TOOL_STAGE:
called.add(tool)
elif kind == "tool_result" and tool in _PIPE_TOOL_STAGE:
resulted.add(tool)
elif kind == "final":
has_final = True
states = {"capture": "done"}
states["read"] = "done" if (called or has_final) else (
"active" if seen_plan else "todo")
for tool, key in _PIPE_TOOL_STAGE.items():
if tool in resulted:
states[key] = "done"
elif tool in called:
states[key] = "active"
else:
# final arrived without this tool ever running: genuinely skipped
# (no calendar uploaded / no events to export)
states[key] = "skip" if has_final else "todo"
if has_final:
states["reply"] = "done"
elif "make_ics" in resulted or "extract_events" in resulted:
states["reply"] = "active"
else:
states["reply"] = "todo"
card = "done" if has_final else "running"
if steps[-1].get("kind") == "error":
card = "error"
for key, *_ in _PIPE_STAGES:
if states.get(key) not in ("done", "skip"):
states[key] = "error"
break
return states, card
def _evidence_label(conversation: str, image_paths, cal_path) -> str:
"""What the run actually read — the Evidence chip in the pipeline summary."""
bits = []
n_chars = len((conversation or "").strip())
if n_chars:
bits.append(f"Pasted thread ({n_chars} chars)")
n_img = len(image_paths or [])
if n_img:
bits.append(f"{n_img} screenshot{'s' if n_img != 1 else ''}")
if cal_path:
try:
bits.append(f"+ {Path(cal_path).name}")
except Exception: # noqa: BLE001
bits.append("+ calendar.ics")
label = " · ".join(bits) or "No input"
return label[:63] + "…" if len(label) > 64 else label
_PIPE_BADGE = {"done": "✓", "skip": "–", "error": "!"}
def _render_pipeline_card(steps: list[dict], meta: dict | None = None) -> str:
"""The elevated stepper card above the step trace: live stage tiles, a
sweeping shimmer + ticking clock while running (wirePipeClock drives the
100ms tick client-side off data-state), and a permanent summary-chip row
once the run completes."""
states, card_state = _pipeline_stage_states(steps)
title = {"done": "Processing complete", "error": "Processing failed"}.get(
card_state, "Processing Pipeline")
elapsed = (meta or {}).get("elapsed")
clock = f"{elapsed:.1f}s" if elapsed is not None else "—"
tiles = []
for i, (key, name, _cap) in enumerate(_PIPE_STAGES):
st = states.get(key, "todo")
badge = _PIPE_BADGE.get(st, str(i + 1))
tiles.append(
f'
'
f'{badge}'
f'{name}
'
)
track = '›'.join(tiles)
caption = ""
if card_state == "running":
cap_txt = next((c for k, _n, c in _PIPE_STAGES
if states.get(k) == "active"), "Working…")
caption = f'
{html.escape(cap_txt)}
'
summary = ""
if card_state == "done" and meta is not None:
final = next((s for s in steps if s.get("kind") == "final"), {})
plan = final.get("plan") or {}
confident = not plan.get("needs_clarification")
conf_cls, conf_txt = (("is-high", "HIGH") if confident
else ("is-review", "NEEDS REVIEW"))
counts = (f"{len(plan.get('events') or [])} event(s) · "
f"{len(plan.get('conflicts') or [])} conflict(s)")
evidence = html.escape(meta.get("evidence") or "—")
summary = (
'
"
)
def _render_agent_trace(steps: list[dict]) -> str:
"""The agent's visible work: one card per orchestration step. The pipeline
card is a SEPARATE component now (above the status line, outside the
accordion) — see the ``pipeline`` / ``ag_pipeline`` gr.HTMLs, fed by
`_render_pipeline_card` from the same generators."""
if not steps:
return '
No run yet — paste a thread and run the agent.
'
cards = []
for s in steps:
icon = _STEP_ICONS.get(s.get("kind"), "•")
kind = s.get("kind", "")
if kind == "tool_call":
body = (f'{html.escape(str(s.get("tool", "")))}'
f' {html.escape(json.dumps(s.get("args", {}), default=str)[:160])}')
elif kind == "tool_result":
body = (f'{html.escape(str(s.get("tool", "")))} → '
f'{html.escape(json.dumps(s.get("result", ""), default=str)[:200])}')
elif kind == "final":
summ = s.get("summary", {})
body = (f'done — {summ.get("events", 0)} event(s), '
f'{summ.get("conflicts", 0)} conflict(s), .ics ready')
else:
body = html.escape(str(s.get("text", "")))
cards.append(f'
{icon}'
f'{body}
')
return f'
{"".join(cards)}
'
def _agent_plan_rows(plan: dict) -> list[list]:
return [
[e.get("title"), e.get("start"), e.get("end") or "",
e.get("location") or "", e.get("reminder_minutes") or ""]
for e in (plan.get("events") or [])
]
def _on_run_agent(thread: str, cal_path, image_paths, mem_json: str = "",
mode: str = ""):
"""Generator: stream the orchestrator's steps into the trace panel, then
surface the FULL homepage result surface — plan summary, carousel, cards,
editable table, reply, and the agent-made .ics.
Yields: (trace, table_rows, plan_md, cards, carousel, reply, status,
conflicts, results_visible, ics_file, pipeline)
"""
import base64 as _b64
import tempfile
from server.orchestrator import run_orchestrator
_idle = (gr.update(),) * 9
if not (thread or "").strip() and not image_paths:
yield (_render_agent_trace([]),
*(gr.update(),) * 5,
"Paste a thread or attach a screenshot first (or use the sample).",
gr.update(), gr.update(), gr.update(),
_render_pipeline_card([]))
return
# Calendar: explicit upload wins; CAL_ICS_PATH keeps parity with the
# homepage's self-hosted auto-calendar behavior.
if not cal_path:
cal_path = os.environ.get("CAL_ICS_PATH", "").strip() or None
ics_b64 = None
if cal_path:
try:
ics_b64 = _b64.b64encode(Path(cal_path).read_bytes()).decode("ascii")
except Exception: # noqa: BLE001
ics_b64 = None
images = paths_to_data_uris(image_paths or [])
memory_block = memory.facts_to_recall(_facts_load(mem_json)) or None
# Same pipeline-card meta as the homepage run (full parity in the trace).
t0 = time.monotonic()
evidence = _evidence_label(thread, image_paths, cal_path)
_meta = lambda: {"elapsed": time.monotonic() - t0, "evidence": evidence} # noqa: E731
steps: list[dict] = []
final = None
for step in run_orchestrator(thread or "", ics_b64=ics_b64,
memory_block=memory_block, images=images or None):
steps.append(step)
if step.get("kind") == "final":
final = step
yield (_render_agent_trace(steps), *(gr.update(),) * 5, "🤖 working…",
gr.update(), gr.update(), gr.update(),
_render_pipeline_card(steps, _meta()))
if not final:
yield (_render_agent_trace(steps), *(gr.update(),) * 5,
"The agent didn't reach a final step.",
gr.update(), gr.update(), gr.update(),
_render_pipeline_card(steps, _meta()))
return
plan_dict = final.get("plan") or {}
rows = _agent_plan_rows(plan_dict)
try:
plan_md = _plan_markdown(ActionPlan(**plan_dict))
except Exception: # noqa: BLE001 defensive: dict drift -> skip summary
plan_md = ""
ics_path = None
if final.get("ics_base64"):
fd, ics_path = tempfile.mkstemp(suffix=".ics", prefix="agent_events_")
os.close(fd)
Path(ics_path).write_bytes(_b64.b64decode(final["ics_base64"]))
summ = final.get("summary", {})
# Same capture semantics as the homepage run (links can't be hooked).
if rows:
impact.record_capture(len(rows),
conflicts_caught=int(summ.get("conflicts", 0) or 0))
yield (
_render_agent_trace(steps),
rows,
plan_md,
_render_event_cards(rows, mode),
_render_event_carousel(rows),
plan_dict.get("reply_draft") or "",
f"Agent finished: {summ.get('events', 0)} event(s) "
f"({len(images)} image(s) read), {summ.get('conflicts', 0)} conflict(s).",
int(summ.get("conflicts", 0) or 0),
gr.update(visible=True),
(gr.update(value=ics_path) if ics_path else gr.update()),
_render_pipeline_card(steps, _meta()),
)
def _on_make_ics(rows, captured_conflicts: int = 0):
events = _rows_to_events(rows)
if not events:
return None, "No events to export."
path = write_ics(events)
# Exporting is the user accepting these events — record it for "This week".
impact.record_capture(len(events), conflicts_caught=int(captured_conflicts or 0))
return path, f"Wrote {len(events)} event(s) to .ics."
def _on_download_trace(redact: bool = True):
"""Export the most recent agent run as a downloadable JSON trace (Sharing is
Caring). Stays local — no Hub token in the Space; publish with
training/share_trace.py."""
from server.trace import export_run, write_trace # lazy, mirrors other handlers
trace = export_run(redact=redact)
if not trace["steps"]:
return None, "No agent run to export yet — analyze a thread first."
path = write_trace(trace)
return path, f"Trace ready ({trace['summary']['steps']} step(s), redacted={redact})."
def _on_publish_trace() -> str:
"""One-click 'Sharing is Caring': publish the latest run's REDACTED trace to a
PUBLIC Hugging Face dataset so anyone can learn from it. Requires a Space write
token (HF_WRITE_TOKEN); degrades gracefully with a clear message otherwise.
Redaction is forced — raw text / titles never reach the public repo."""
token = (os.environ.get("HF_WRITE_TOKEN") or "").strip()
repo = os.environ.get("TRACE_DATASET", "ParetoOptimal/offgridschedula-traces")
if not token:
return ("⚠️ Publishing isn't configured on this Space. Set an **`HF_WRITE_TOKEN`** "
"secret (a dataset-scoped *write* token) to enable one-click publish — "
"or use **⬇ Download trace** + `python training/share_trace.py`.")
from server.trace import export_run # lazy
trace = export_run(redact=True) # force redaction for a PUBLIC dataset
if not trace["steps"]:
return "Nothing to publish yet — run an analysis first, then publish its trace."
import io
from datetime import datetime
from huggingface_hub import HfApi
url = f"https://huggingface.co/datasets/{repo}"
fname = f"traces/trace-{datetime.now().strftime('%Y%m%d-%H%M%S')}.json"
try:
api = HfApi(token=token)
api.create_repo(repo_id=repo, repo_type="dataset", private=False, exist_ok=True)
api.upload_file(
path_or_fileobj=io.BytesIO(json.dumps(trace, indent=2).encode("utf-8")),
path_in_repo=fname,
repo_id=repo,
repo_type="dataset",
commit_message="Add agent trace (redacted) — Sharing is Caring",
)
except Exception as e: # noqa: BLE001 surface a friendly message, never crash the UI
return f"Publish failed: {type(e).__name__}: {e}"
bus.emit("decision", f"published redacted trace ({trace['summary']['steps']} step(s)) to the Hub")
return f"✅ Shared your **redacted** trace publicly → [{repo}]({url}/blob/main/{fname})"
def _on_push_gcal(rows, token_json, captured_conflicts: int = 0):
"""Push to the VISITOR'S own Google Calendar using their per-session OAuth
token (from the Connect flow). The token lives in the browser; it's passed
here only to perform the push and is never stored server-side."""
if not (token_json or "").strip():
return "Connect Google Calendar first (Step 2a → **Connect your calendar**)."
events = _rows_to_events(rows)
if not events:
return "No events to push."
try:
from calendar_out.gcal import push_events_with_token
links = push_events_with_token(token_json, events)
# Only record once the push actually succeeded.
impact.record_capture(len(events), conflicts_caught=int(captured_conflicts or 0))
return "Added to your Google Calendar:\n" + "\n".join(links)
except Exception as e: # noqa: BLE001
return f"Google Calendar push failed: {e}"
# --------------------------------------------------------------------------- #
# Activity dashboard renderers
# --------------------------------------------------------------------------- #
def _stepper_html() -> str:
cur = bus.current_stage()
cells = []
for s in bus.STAGES:
active = " active" if s == cur else ""
color = STAGE_COLORS.get(s, "#888")
cells.append(f'
'
def _timeline_html(n: int = 40) -> str:
rows = bus.recent(n)
if not rows:
return '
No activity yet — analyze a thread or POST to /ingest.
'
items = []
for e in rows:
color = STAGE_COLORS.get(e["stage"], "#888")
meta_bits = []
for k in ("latency_ms", "events", "conflicts", "images", "tokens"):
v = e.get(k)
if v:
meta_bits.append(f"{k}={v}")
meta = " · ".join(meta_bits)
err = " err" if e.get("level") == "error" else ""
items.append(
f'
"
def _impact_html() -> str:
"""Durable weekly impact (events captured, conflicts caught, time saved).
Reuses the same .tiles/.tile markup as _tiles_html so the CSS styles it for
free — but reads from the persisted impact log, not the ephemeral bus."""
w = impact.this_week()
mins = w["minutes_saved"]
saved = f"{mins // 60}h {mins % 60}m" if mins >= 60 else f"{mins}m"
defs = [
("Events captured", w["events_captured"]),
("Conflicts caught", w["conflicts_caught"]),
("Time saved", saved),
]
cells = "".join(
f'
{v}
{k}
'
for k, v in defs
)
return f'
{cells}
'
def _chart_df() -> pd.DataFrame:
return pd.DataFrame(bus.stage_counts())
def _traces_html(n: int = 6) -> str:
runs = bus.recent_runs(n)
if not runs:
return '
No agent runs yet.
'
blocks = []
for rid, evs in runs:
lines = []
for e in evs:
color = STAGE_COLORS.get(e["stage"], "#888")
lat = f' ({e["latency_ms"]} ms)' if e.get("latency_ms") else ""
lines.append(
f'
{e["stage"]} {e["message"]}{lat}
'
)
summary = f"run #{rid} — {len(evs)} step(s)"
blocks.append(
f'{summary}{"".join(lines)}'
)
return "".join(blocks)
def _refresh():
return (
_stepper_html(),
_tiles_html(),
_timeline_html(),
_chart_df(),
_traces_html(),
_impact_html(),
)
# --------------------------------------------------------------------------- #
# Landing-page sections (static marketing HTML; the tool itself stays Gradio)
# --------------------------------------------------------------------------- #
def _nav_html() -> str:
"""Single grouped nav. Page links carry data-tab (the injected JS clicks the
matching Gradio tab); landing links also carry data-anchor to scroll within
the Schedule page. The default Gradio tab strip is hidden via CSS.
(The fine-tuned-model "Powered by" line lives under the hero badges, not here.)"""
return (
'
Meetings, events, activities — buried in a noisy group chat or a '
'screenshot. Paste it in and get calendar-ready events, conflicts caught against your '
'calendar, and a reply written for you.
'
'
No app · no account · runs locally — your messages never touch a cloud AI.
'
'
'
)
def _hero_visual_html() -> str:
"""A design play on the core function: a chat bubble morphing into a
calendar event chip. Pure HTML/CSS (no JS) so it renders on every path."""
return (
'
'
'
'
'
Class group chat
'
'
Picture day is Thursday at 9am — wear the green shirt! 📸
'
'
'
'
↓
'
'
'
''
'
'
'
📸 Picture Day
'
'
Thu · 9:00 AM
'
'
Lincoln Elementary · wear green
'
'
'
'
'
)
def _hero_examples_html() -> str:
"""A 2×3 grid of compact 'chat → event' example cards for the hero.
The event lines are COMPUTED against today's date, following the site's
date-resolution rules: weekdays resolve to the next occurrence on or after
today; AM/PM is preserved exactly and inferred meridiems are flagged
'assumed'; stated times are never dropped; time-only messages inherit
today; deadlines are separate entries; month-day dates already past roll
to next year. (Rendered at app startup — refreshes on each restart.)"""
from datetime import datetime as _dt
today = _dt.now()
def next_weekday(dow: int) -> _dt: # Mon=0; next occurrence ON OR AFTER today
return today + timedelta(days=(dow - today.weekday()) % 7)
def day_label(d: _dt) -> str:
return f"{d.strftime('%a')}, {d.strftime('%b')} {d.day}"
def line(title: str, when: str, time_s: str = "", assumed: bool = False) -> str:
parts = [title, when] + ([time_s] if time_s else [])
flag = ' assumed' if assumed else ""
return html.escape(" · ".join(parts)) + flag
def month_day(month: int, day: int) -> _dt: # rule 6: roll past dates to next year
d = _dt(today.year, month, day)
return d if d.date() >= today.date() else _dt(today.year + 1, month, day)
bake = month_day(6, 12)
ex = [
("School chat", "Picture day Thursday 9am",
[line("📸 Picture Day", day_label(next_weekday(3)), "9:00 AM")]),
("Coach", "Practice moved to Tue 5pm",
[line("⚽ Practice", day_label(next_weekday(1)), "5:00 PM")]),
("Carpool", "I’ll grab them at 3:15",
[line("🚗 Pickup", "Today", "3:15 PM", assumed=True)]),
("Doctor", "Appt confirmed Mon 10:30",
[line("🩺 Checkup", day_label(next_weekday(0)), "10:30 AM", assumed=True)]),
("Party", "RSVP by Fri — party Sat 2pm",
[line("📝 RSVP due", day_label(next_weekday(4))),
line("🎉 Party", day_label(next_weekday(5)), "2:00 PM")]),
("Flyer", "Bake sale — June 12, 8am · drop off cookies by Thu 6pm",
[line("🍪 Drop off cookies", day_label(next_weekday(3)), "6:00 PM"),
line("🧁 Bake Sale", day_label(bake), "8:00 AM")]),
]
cards = "".join(
f'
{html.escape(frm)}
'
f'
{html.escape(chat)}
'
+ "".join(f'
{ev}
' for ev in evs)
+ "
"
for frm, chat, evs in ex
)
return f'
{cards}
'
def _hero_badges_html() -> str:
"""Live trust badges under the hero copy (read from env)."""
f = _compliance_facts()
local = "🔒 100% local" if f["local"] else "☁️ remote"
return (
'
"
)
# --- hackathon compliance: live facts (read from env) + showcase HTML ----------
# Model: gemma-cal E4B — our calendar-native fine-tune of google/gemma-4-E4B-it
# (~4B effective params, <= 32B), published as build-small-hackathon/gemma-4-cal-gguf.
_SPACE_URL = "https://huggingface.co/spaces/build-small-hackathon/OffGridSchedula"
_BLOB = _SPACE_URL + "/blob/main/"
_MODEL_URL = "https://huggingface.co/build-small-hackathon/gemma-4-cal-gguf"
def _compliance_facts() -> dict:
"""Read the live model/runtime config so the badges reflect reality."""
repo = os.environ.get("MODEL_REPO", "build-small-hackathon/gemma-4-cal-gguf")
fname = os.environ.get("MODEL_FILE", "gemma-cal-e4b-Q4_K_M.gguf")
base = os.environ.get("INFERENCE_BASE_URL", "")
local = (not base) or ("127.0.0.1" in base) or ("localhost" in base)
rl = f"{repo}/{fname}".lower()
if "e4b" in rl:
label, params = "gemma-cal E4B (fine-tuned)", "~4B"
elif "e2b" in rl:
label, params = "Gemma 4 E2B", "~2B"
elif "31b" in rl or "gemma-cal-q4" in rl:
label, params = "gemma-cal 31B (fine-tuned)", "31B"
else:
label, params = repo.split("/")[-1], "≤32B"
return {"repo": repo, "local": local, "label": label, "params": params}
def _submission_html() -> str:
"""A live compliance scorecard — maps every hackathon rule to evidence."""
f = _compliance_facts()
loc = "100% local — no cloud AI APIs" if f["local"] else "remote inference seam"
def row(ok: bool, title: str, ev: str) -> str:
cls, mark = ("sub-ok", "✓") if ok else ("sub-warn", "⚠")
return (
f'
{mark}'
f'
{title}
'
f'
{ev}
'
)
hard = (
row(True, "Small model ≤ 32B",
f'{html.escape(f["label"])} · {html.escape(f["params"])} params (≤ 32B) — '
"our own fine-tune of Gemma 4 E4B, published on the Hub. Laptop-class by design: "
"~4B effective params, about 5 GB at 4-bit.")
+ row(True, "Gradio app · Hugging Face Space",
"app.py is a Gradio Blocks app, shipped as a Hugging Face Space — a "
"Docker image that runs llama.cpp locally. You're reading this scorecard inside it.")
+ row(True, "Show, don’t tell",
"So we didn't tell — we showed. A demo video of the real product running (no slides, "
"just the thing doing the thing) ships with the submission, alongside the social post.")
)
track = (
'
Track A — Backyard AI. A specific real person: a busy parent '
"whose kid’s school & activity events are buried in a noisy class group chat. Honest fit — "
"short pasted chats and screenshots are exactly what a small, local model handles well, and the "
"output (a reviewable .ics) is genuinely usable from a phone browser.
"
)
bonus = (
row(True, "Off the Grid (local-first)", f"{loc} — inference runs via llama.cpp in the Space.")
+ row(True, "Well-Tuned",
"A QLoRA fine-tune of Gemma 4 E4B, eval-gated: every retrain has to clear a "
"60-example scorecard before it can ship — eight of them didn't. The one that did is "
"published on the Hub.")
+ row(True, "Off-Brand (custom UI)",
"This site — custom landing, grouped nav, dark/light hero, and bespoke CSS/JS, far past "
"the default Gradio look.")
+ row(True, "Llama Champion",
"Runs through the llama.cpp runtime (official ggml-org/llama.cpp image).")
+ row(True, "Sharing is Caring",
"One-click ☁ Publish my trace to the Hub (Activity tab) pushes a redacted "
"run to a public dataset — counts and stage names, never your chat text — so anyone can "
"learn from it. (Token-free route: ⬇ Download trace and publish it yourself.)")
+ row(True, "Agentic (MCP tool server + agent)",
"Both sides of MCP: the Space exposes "
"extract_events / make_ics / check_conflicts "
"as Model Context Protocol tools (schemas auto-generated) — and the "
"Agent tabconsumes them: an OpenBMB MiniCPM planner "
"(local llama.cpp) makes a single, visible extract_events call over MCP, "
"while conflict-checking and .ics rendering are finalized deterministically. "
"No cloud AI APIs anywhere in the loop.")
+ row(True, "OpenBMB MiniCPM (sponsor)",
"The Agent tab's planner is MiniCPM "
"(openbmb/MiniCPM5-1B-GGUF, the ≤4B tiny variant; the 8B "
"MiniCPM4.1-8B is config-selectable) served by the same local "
"llama.cpp — core to the agent experience, under the 32B cap.")
+ row(True, "Field Notes",
"By the numbers: 8 fine-tunes rejected at the eval gate, "
"7 bugs in agent code that had never run, a planner trimmed "
"207 s → 75 s — and still zero cloud calls.")
)
return (
''
'
Hackathon submission
'
'
Built-small, by the rules.
'
'
Hard constraints
' + hard + "
"
'
Track
' + track + "
"
'
Bonus quests
' + bonus + "
"
""
)
def _how_html() -> str:
steps = [
("01", "Paste it in", "Drop in the group chat — or a screenshot of a flyer, invite, or text."),
("02", "Review everything", "See the events, conflicts against your calendar, and a ready-to-send reply — before anything is saved."),
("03", "Add to your calendar", "Download a local .ics for any calendar, or push straight to Google or Apple Calendar."),
]
cards = "".join(
f'
{n}
'
f'
{t}
{d}
'
for n, t, d in steps
)
return (
''
'
How it works
'
'
From buried message to booked — in three taps.
'
f'
{cards}
'
''
)
def _uses_html() -> str:
uses = [
("🗓", "Meetings", "“Sync moved to Thursday 2pm” → on your calendar, no retyping."),
("🎒", "School & activities", "Picture day, the practice that moved, the field trip slip."),
("🚗", "Carpool & pickups", "Pickup times and whose turn to drive — sorted."),
("🩺", "Appointments", "Doctor, dentist & vet reminders, with a heads-up before."),
("🎉", "Parties & RSVPs", "The party, the RSVP deadline, and a one-tap reply."),
("📸", "Screenshots & flyers", "Snap an invite — it reads the date, time & place from the image."),
]
cards = "".join(
f'
{i}
'
f'
{t}
{d}
'
for i, t, d in uses
)
return (
''
'
Use cases
'
'
Anywhere a date hides in a conversation.
'
f'
{cards}
'
''
)
def _faq_html() -> str:
"""FAQ with two tabs (General / For developers), search filter, and native
accordion rows. Wired by the FAQ block in CAROUSEL_JS."""
general: list[tuple[str, str]] = [
(
"What problem does this actually solve?",
"
School pickups, practice changes, RSVPs, doctor reminders — "
"the dates that hide in noisy group chats and flyers. You paste "
"the relevant chat (or a screenshot); the agent extracts the "
"events, flags conflicts against your calendar, and drafts a "
"reply. You go back to managing one calendar instead of "
"scanning four chats.
",
),
(
"Is anything sent to a cloud AI service?",
"
No. Inference runs locally inside the Space via "
"llama.cpp — no OpenAI / Anthropic / Google AI "
"calls. The only outbound network call you might make is the "
"optional push to Google Calendar, and only if you click "
"it and authorise it yourself.
",
),
(
"Do I need an app or an account?",
"
No. It works from any phone or laptop browser. There's "
"nothing to install and nothing to sign up for. The output is "
"a .ics file you own — import it into Apple "
"Calendar, Google Calendar, Outlook, or anything else.
",
),
(
"Does it read my messages automatically?",
"
No. It reads only the text you choose to paste in (and any "
"screenshots you attach). It has no access to your inbox, your "
"contacts, or your phone's Messages app. Nothing is read "
"without your action.
",
),
(
"What inputs work best?",
"
"
"
Short pasted chats — the last few relevant messages, "
"not a six-month history.
"
"
Screenshots of flyers, invites, or texts where the "
"date lives in the image.
"
"
Your current calendar as .ics "
"(optional) — used only for deterministic conflict detection, "
"never sent anywhere.
"
"
",
),
(
"How accurate is it, really?",
"
Small fine-tuned models are good at narrow tasks and worse "
"at edge cases. Honestly:
"
"
"
"
Reliable for explicit times — \"Tuesday at 3pm at "
"school\".
"
"
Decent for nearby relative times — \"tomorrow\", "
"\"next Monday\".
"
"
Worse for ambiguous wording — \"maybe later\". That's "
"why the agent always shows you the events for review before "
"anything is saved.
"
"
",
),
]
dev: list[tuple[str, str]] = [
(
"Can other AI agents call this Space as a tool?",
"
Yes. The Space exposes a Model Context Protocol (MCP) "
"server, so any MCP-aware agent (Claude Desktop, Cursor, "
"etc.) can call three tools against it:
"
"
"
"
extract_events(thread, images) — the headline "
"tool: read a chat or screenshot, return events + conflicts + "
"reply.
"
"
make_ics(events) — render events as a "
"base64-encoded .ics.
"
"
check_conflicts(events, ics_base64) — "
"deterministic free/busy check against an uploaded calendar.
"
"
"
"
Schemas are auto-generated from the typed wrappers in "
f'server/mcp_tools.py.
',
),
(
"How do I connect an MCP-aware agent to this Space?",
"
Point the agent's MCP config at the Space's SSE endpoint:
In Claude Desktop, add an entry under "
"mcpServers in "
"claude_desktop_config.json with that URL. In "
"Cursor, add it under Settings → MCP. The three tools "
"appear automatically; the agent picks them up by docstring.
",
),
(
"Where do I find the fine-tuned model?",
"
Published openly on the Hugging Face Hub as "
f'build-small-hackathon/gemma-4-cal-gguf'
" — a QLoRA fine-tune of Gemma 4 converted to GGUF so "
"llama.cpp can run it on CPU or GPU. The Space pulls the model "
"weights at startup; nothing custom is bundled into this repo.
",
),
(
"Can I run this off the Space, locally?",
"
Yes — this Space is the repo. Its "
f'Files tab is a public git '
"repository: git clone "
f"{_SPACE_URL}, install requirements.txt (or "
"build the Dockerfile), set MODEL_REPO + "
"MODEL_FILE to point at any GGUF on the Hub, and "
"run python app.py. The same Gradio UI + FastAPI "
"/agent endpoint + MCP server come up on "
"localhost:7860.
",
),
]
def _rows(items: list[tuple[str, str]]) -> str:
# Inline SVG: outer circle + horizontal line (always shown) + vertical
# line (hidden when item is open) -> visual "+" toggles to "-".
ico = (
''
'"
)
return "".join(
f''
f'{q}{ico}'
f'
{a}
'
for q, a in items
)
search_svg = (
''
)
return (
''
'
'
'
Frequently asked questions
'
'"
"
"
'
'
''
''
"
"
f'
{_rows(general)}
'
f'
{_rows(dev)}
'
'
No questions match — try '
"different words.
"
""
)
def _privacy_html() -> str:
points = [
("Runs locally", "Inference happens in the Space via llama.cpp — no cloud AI APIs."),
("Nothing auto-read", "It reads only what you choose to paste. No inbox, no message access."),
("No app, no account", "Works from your phone's browser. Output is a local .ics you own."),
]
items = "".join(
f'
'
)
def _tool_header_html() -> str:
"""Header for the elevated tool card: eyebrow + title + the "Powered by" pill
(moved here from the hero — it identifies the model where you actually use it)."""
return (
''
'
"
)
def _flow_step(n: "int | str", title: str, sub: str = "") -> str:
"""Numbered workflow step header: round chip + title (+ muted sub note),
with a dashed tail (CSS) tying the steps together down the card."""
sub_html = f'{sub}' if sub else ""
return (
f'
{n}'
f'{title}{sub_html}
'
)
def _mode_note_html(online: bool) -> str:
"""Guidance line + named workflow chip inside the mode band."""
if online:
chip = "ONLINE WORKFLOW"
txt = ("☁️ Online — Google Calendar enabled: connect it in step 2a, "
"or just continue. Everything else still runs locally.")
else:
chip = "OFF-GRID WORKFLOW"
txt = ("🔌 Offline — everything stays on this device: upload your "
"calendar in step 2a, get an .ics back.")
return (f'
{chip}'
f"{txt}
")
def _footer_html() -> str:
return (
''
)
# --------------------------------------------------------------------------- #
# App
# --------------------------------------------------------------------------- #
def build_demo() -> gr.Blocks:
# Theme/css set on the Blocks so the Off-Brand look applies no matter who
# launches it (our uvicorn locally, or HF's gradio runtime on a Spaces deploy).
with gr.Blocks(theme=THEME, css=CSS, title="OffGridSchedula") as demo:
gr.HTML('', elem_id="status-banner-host")
gr.HTML(_nav_html())
# Per-user memory lives in the visitor's browser (localStorage); this hidden
# box mirrors it so handlers can read/edit it and the agent can use it.
# Loaded from localStorage on page load; persisted back on every change.
mem_box = gr.Textbox(visible=False, elem_id="mem-box")
with gr.Tab("Schedule"):
captured_conflicts = gr.State(0)
# --- landing hero: dark band — copy + example-card grid ---
with gr.Row(elem_id="hero"):
with gr.Column(elem_id="hero-left"):
gr.HTML(_hero_copy_html())
with gr.Row(elem_id="hero-cta"):
hero_try_btn = gr.Button(
"Try a sample", variant="primary", elem_id="hero-try", scale=0
)
gr.HTML('See it in action ↓')
gr.HTML(_hero_badges_html())
with gr.Column(elem_id="hero-right"):
gr.HTML(_hero_examples_html())
# --- elevated tool card (overlaps the hero); the agent up top ---
with gr.Column(elem_id="tool-card"):
gr.HTML(_tool_header_html())
# ── Step 1: Choose your workflow — the mode decision point ──── #
gr.HTML(_flow_step(1, "Choose your workflow"))
# Offline / Online mode toggle: ONE decision point that re-themes
# and reconfigures the whole workflow card. Lives in a full-width
# tinted "mode band"; the card carries data-mode for the CSS theme.
with gr.Group(elem_id="mode-band"):
mode = gr.Radio(
["🔌 Offline", "☁️ Online"], value="🔌 Offline",
show_label=False, elem_id="mode-toggle", container=False,
)
mode_note = gr.HTML(_mode_note_html(online=False))
# ── Step 2: Upload context — screenshot + paste, one group ──── #
gr.HTML(_flow_step(2, "Upload context"))
with gr.Row(elem_id="io-cols"):
with gr.Column(elem_classes="io-col"):
gr.HTML('
')
conversation = gr.Textbox(
label="",
placeholder=(
"Paste the school / activity chat — e.g. “Picture day is "
"Thursday at 9am, wear green”."
),
lines=6,
elem_id="rv-textbox",
)
gr.HTML(
'
Chat, SMS, or invite — English & '
'screenshots supported'
'0 / 12000
'
)
# ── Step 2a: Optional — calendar (first option) · trace · personalize ── #
gr.HTML(_flow_step("2a", "Optional", "(calendar · privacy trace · personalize)"))
# First option: the calendar, as a dropdown (caret) holding BOTH
# paths — connect a cloud provider (one unified block, all three
# providers) or upload an .ics. Always rendered; offline mode
# hides only the Google connect CTA via CSS (data-mode).
with gr.Accordion(
"🔗 Connect your calendar (+ optional .ics for conflict checks)",
open=False,
) as cal_acc:
gr.HTML(_connect_block_html())
existing_cal = gr.File(
label="Your calendar (.ics)", file_types=[".ics"], type="filepath",
)
with gr.Group(elem_classes="trace-card"):
trace_checkbox = gr.Checkbox(
label=("Publish privacy-safe trace — stores a redacted event summary "
"for the agent-trace dataset; raw text, screenshots, and links "
"are not stored."),
value=False, elem_id="trace-cb",
)
trace_status = gr.HTML()
# Onboarding — ALWAYS on the page so the agent can be personalized
# at any time (people the chat mentions, preferences, etc.). Open
# for first-time visitors; collapses (never disappears) once the
# device has memory or the user hits Skip.
with gr.Accordion(
"👋 Personalize in 10 seconds", open=True, elem_id="onboard"
) as onboard:
gr.HTML(
'
Tell the agent a few things — saved on your device '
"and used to read your chats better. You can skip this.
"
)
ob_people = gr.Textbox(
label="People (one per line — “Name = role”)", lines=2,
placeholder="Dana = soccer coach\nMr. Lee = teacher",
)
with gr.Row():
ob_reminder = gr.Number(label="Default reminder (min before)", value=30, precision=0)
ob_decline = gr.Textbox(label="Days you usually decline", placeholder="Mondays")
with gr.Row():
ob_save = gr.Button("Save to my device", variant="primary", scale=0)
ob_skip = gr.Button("Skip", elem_classes="rv-secondary", scale=0)
ob_status = gr.Markdown(elem_classes="rv-status")
# ── Step 3: Run — the three actions nested under one step,
# uniformly formatted, with "Run the agents" the highlighted
# (primary) choice. ───────────────────────────────────────── #
gr.HTML(_flow_step(3, "Run the agents"))
with gr.Row(elem_id="rv-actions"):
analyze_btn = gr.Button(
"Run the agents", variant="primary", elem_id="rv-analyze", scale=0
)
start_over_btn = gr.Button(
"Start over", elem_id="rv-startover", elem_classes="rv-secondary", scale=0
)
sample_btn = gr.Button(
"Use the sample", elem_id="rv-sample", elem_classes="rv-secondary", scale=0
)
# The pipeline tracker sits ABOVE the status line, always
# visible (idle card on landing) — the step-by-step trace
# stays behind the accordion below.
pipeline = gr.HTML(_render_pipeline_card([]), elem_id="rv-pipeline")
status = gr.Markdown(elem_classes="rv-status")
with gr.Accordion("🤖 Watch the agent work (live steps)", open=False):
thinking = gr.HTML(_render_agent_trace([]))
# --- results: hidden until the first analysis (progressive disclosure)
with gr.Column(elem_id="rv-results", visible=False) as results:
# The rotating carousel is superseded by the condensed
# results card below; the component stays (hidden, still
# rendered into) so every chain that targets it keeps
# working unchanged.
hero_carousel = gr.HTML("", elem_id="rv-hero", visible=False)
# Per-event notes from the last analysis. Notes aren't a
# Dataframe column, so without this State the arrival
# context would be lost when events round-trip through
# the editable table.
found_notes = gr.State([])
# ONE condensed results card: the found events and the
# export actions together, instead of separate stacked
# sections.
with gr.Group(elem_id="rv-resultcard"):
plan_md = gr.HTML()
cards = gr.HTML()
with gr.Group(elem_id="rv-export"):
# The per-event "Add to: Google · Outlook · iCal"
# links on each card are now the ONE export
# surface. This button bar duplicated it (and the
# OAuth push 404s while the Space is private), so
# everything here is HIDDEN — not deleted: the
# mode wiring, wireGcal, and tests reference these
# components, and the OAuth push returns once the
# Space is public.
with gr.Row(visible=False):
ics_btn = gr.Button("⬇ Download .ics", variant="primary", scale=1)
gcal_btn = gr.Button("Add to Google Calendar", scale=1)
apple_btn = gr.Button(" Add to Apple Calendar", scale=1)
gcal_link_html = gr.HTML("", visible=False)
# hidden per-session token, hydrated from localStorage / the OAuth popup
gcal_token = gr.Textbox(visible=False, elem_id="gcal-token")
ics_file = gr.File(label="events.ics", elem_id="ics-file",
visible=False)
gcal_status = gr.Markdown(visible=False)
gr.HTML(
'
📱 On iPhone? The share-sheet '
"Shortcut adds events in two taps — "
f'set it up.
'
)
with gr.Accordion("✏️ Edit events", open=False):
table = gr.Dataframe(
headers=["title", "start", "end", "location", "reminder_min"],
datatype=["str", "str", "str", "str", "number"],
type="array",
interactive=True,
label="",
)
with gr.Group(elem_classes="rv-reply"):
reply = gr.Textbox(label="Suggested reply", lines=3, elem_id="rv-reply-box")
copy_btn = gr.Button("Copy reply", size="sm", elem_classes="rv-copy")
# --- marketing sections (now below the working tool) ---
gr.HTML(_how_html())
gr.HTML(_uses_html())
gr.HTML(_privacy_html())
# --- wiring (existing handler signatures unchanged) ---
analyze_inputs = [conversation, existing_cal, images_in, mem_box]
analyze_outputs = [thinking, table, plan_md, reply, status, found_notes,
pipeline]
_reveal = lambda: gr.update(visible=True) # noqa: E731
# Client-side helpers: scroll the page (sample CTA lives way up in the
# hero) and turn the generated .ics into a real browser download.
_scroll_tool_js = (
"() => { var el = document.getElementById('tool-card') || "
"document.getElementById('tool'); if (el) el.scrollIntoView("
"{behavior: 'smooth', block: 'start'}); }"
)
_scroll_results_js = (
"() => { var el = document.getElementById('rv-results'); "
"if (el) el.scrollIntoView({behavior: 'smooth', block: 'start'}); }"
)
_auto_download_js = (
"(f) => { try { if (f && f.url) { var a = document.createElement('a'); "
"a.href = f.url; a.download = 'events.ics'; document.body.appendChild(a); "
"a.click(); a.remove(); } } catch (e) {} }"
)
# Opening (not downloading) the .ics lets iOS/macOS hand it straight
# to Apple Calendar ("Add All"); other platforms fall back to a download.
_open_ics_js = (
"(f) => { try { if (f && f.url) window.open(f.url, '_blank'); } catch (e) {} }"
)
analyze_btn.click(_on_analyze, analyze_inputs, analyze_outputs).then(
_count_conflicts, plan_md, captured_conflicts
).then(_render_event_cards, [table, mode, found_notes], cards).then(
_render_event_carousel, table, hero_carousel
).then(_reveal, None, results).then(
_on_share_trace, [trace_checkbox, table], trace_status
).then(_prep_ics, table, ics_file)
# Uploading a screenshot IS the intent too — analyze as soon as the
# upload lands (the .upload event only fires with files, never on clear).
images_in.upload(_on_analyze, analyze_inputs, analyze_outputs).then(
_count_conflicts, plan_md, captured_conflicts
).then(_render_event_cards, [table, mode, found_notes], cards).then(
_render_event_carousel, table, hero_carousel
).then(_reveal, None, results).then(
None, None, None, js=_scroll_results_js
).then(
_on_share_trace, [trace_checkbox, table], trace_status
).then(_prep_ics, table, ics_file)
# One tap: fill the box with the sample, then run the same analysis.
# The hero CTA and the in-tool "try a sample" share this flow.
for _btn in (sample_btn, hero_try_btn):
_btn.click(_load_sample, None, conversation).then(
None, None, None, js=_scroll_tool_js
).then(
_on_analyze, analyze_inputs, analyze_outputs
).then(_count_conflicts, plan_md, captured_conflicts).then(
_render_event_cards, [table, mode, found_notes], cards
).then(_render_event_carousel, table, hero_carousel).then(
_reveal, None, results
).then(
None, None, None, js=_scroll_results_js
).then(
_on_share_trace, [trace_checkbox, table], trace_status
).then(_prep_ics, table, ics_file)
# Start over: clear inputs + hide results.
start_over_btn.click(
_on_start_over, None,
[conversation, images_in, existing_cal, results, status,
screenshot_status, trace_status, pipeline],
)
# (Offline/Online mode wiring lives at the end of build_demo — it also
# drives components on other tabs, which don't exist yet at this point.)
# Contextual hint when a screenshot is attached.
images_in.change(_screenshot_status, images_in, screenshot_status)
# Keep cards + carousel + the pre-generated .ics in sync when the
# user edits the table.
table.change(_render_event_cards, [table, mode, found_notes], cards).then(
_render_event_carousel, table, hero_carousel
).then(_prep_ics, table, ics_file)
# Copy the reply to the clipboard (client-side, no round-trip).
copy_btn.click(None, [reply], None, js="(t) => { navigator.clipboard.writeText(t || ''); }")
# Download .ics: generate, then immediately save it as events.ics —
# no second click on the File component needed.
ics_btn.click(
_on_make_ics, [table, captured_conflicts], [ics_file, status]
).then(None, ics_file, None, js=_auto_download_js)
# Apple Calendar: same .ics, opened so the OS offers "Add to Calendar".
apple_btn.click(
_on_make_ics, [table, captured_conflicts], [ics_file, status]
).then(None, ics_file, None, js=_open_ics_js)
# Pull the visitor's per-session token straight from localStorage at
# click time (set by the OAuth popup), so the push uses THEIR account.
gcal_btn.click(
_on_push_gcal, [table, gcal_token, captured_conflicts], gcal_status,
js="(rows, tok, cc) => [rows, "
"(function(){try{return localStorage.getItem('gcal_token')||'';}catch(e){return '';}})(), "
"cc]",
)
# --- frequently-asked questions (native accordion) ---
gr.HTML(_faq_html())
# --- landing footer ---
gr.HTML(_footer_html())
with gr.Tab("Agent", elem_classes="page-wrap"):
ag_conflicts = gr.State(0)
gr.Markdown(
"### 🤖 Agent mode — the whole homepage workflow, run by an agent\n"
"A small **OpenBMB MiniCPM** planner (served by the same local "
"llama.cpp — no cloud AI) drives this Space's own **MCP tools** "
"(`extract_events` → `check_conflicts` → `make_ics`) as a "
"multi-step agent, with your device memory as context. Every "
"step it takes is shown below; the results carry the full "
"review-and-export surface from the Schedule page. On the free "
"preview the same playbook runs scripted, so the demo always works."
)
with gr.Row():
ag_images = gr.Files(label="Screenshots (optional)",
file_types=["image"], type="filepath")
ag_thread = gr.Textbox(
label="Thread", lines=5,
placeholder="Paste a chat for the agent to work end-to-end…",
)
ag_cal = gr.File(label="Calendar (.ics, optional — for conflict checks)",
file_types=[".ics"], type="filepath")
with gr.Row():
ag_run = gr.Button("Run the agent", variant="primary", scale=0)
ag_sample = gr.Button("use the sample", elem_classes="rv-linkbtn",
size="sm", scale=0)
# Same placement as the homepage: tracker above the status line,
# step cards below.
ag_pipeline = gr.HTML(_render_pipeline_card([]))
ag_status = gr.Markdown(elem_classes="rv-status")
gr.Markdown("#### The agent's steps")
ag_trace = gr.HTML(_render_agent_trace([]))
# --- results: the homepage's full surface, agent edition ---
with gr.Column(visible=False) as ag_results:
ag_carousel = gr.HTML("")
ag_plan_md = gr.HTML()
ag_cards = gr.HTML()
with gr.Accordion("✏️ Edit events", open=False):
ag_table = gr.Dataframe(
headers=["title", "start", "end", "location", "reminder_min"],
datatype=["str", "str", "str", "str", "number"],
type="array", interactive=True, label="",
)
with gr.Group(elem_classes="rv-reply"):
ag_reply = gr.Textbox(label="Suggested reply", lines=2)
ag_copy = gr.Button("Copy reply", size="sm", elem_classes="rv-copy")
with gr.Group(elem_classes="ag-export"):
# Hidden like the homepage bar: the per-event card links
# are the one export surface (components kept for the
# mode wiring + the OAuth push's return).
with gr.Row(visible=False):
ag_ics_btn = gr.Button("⬇ Download .ics", variant="primary", scale=1)
ag_apple_btn = gr.Button(" Add to Apple Calendar", scale=1)
ag_gcal_btn = gr.Button("Add to Google Calendar", scale=1,
visible=False)
ag_gcal_link = gr.HTML(_provider_row("google"), visible=False)
ag_ics = gr.File(label="events.ics (made by the agent)",
visible=False)
ag_gcal_status = gr.Markdown(visible=False)
ag_sample.click(_load_sample, None, ag_thread)
ag_run.click(
_on_run_agent, [ag_thread, ag_cal, ag_images, mem_box, mode],
[ag_trace, ag_table, ag_plan_md, ag_cards, ag_carousel,
ag_reply, ag_status, ag_conflicts, ag_results, ag_ics,
ag_pipeline],
)
# Same export semantics as the homepage (impact recorded on export).
ag_copy.click(None, [ag_reply], None,
js="(t) => { navigator.clipboard.writeText(t || ''); }")
ag_ics_btn.click(
_on_make_ics, [ag_table, ag_conflicts], [ag_ics, ag_status]
).then(None, ag_ics, None, js=_auto_download_js)
ag_apple_btn.click(
_on_make_ics, [ag_table, ag_conflicts], [ag_ics, ag_status]
).then(None, ag_ics, None, js=_open_ics_js)
ag_gcal_btn.click(
_on_push_gcal, [ag_table, gcal_token, ag_conflicts], ag_gcal_status,
js="(rows, tok, cc) => [rows, "
"(function(){try{return localStorage.getItem('gcal_token')||'';}catch(e){return '';}})(), "
"cc]",
)
# Edits keep cards + carousel + the pre-made .ics in sync (parity).
ag_table.change(_render_event_cards, [ag_table, mode], ag_cards).then(
_render_event_carousel, ag_table, ag_carousel
).then(_prep_ics, ag_table, ag_ics)
with gr.Tab("Activity", elem_classes="page-wrap"):
gr.Markdown("### This week")
gr.Markdown(
"What this saved you — events captured, conflicts caught, and "
"estimated time saved. Persists across restarts."
)
impact_panel = gr.HTML()
gr.Markdown("---\nLive view of what the agent and model are doing (refreshes every 2s).")
stepper = gr.HTML()
tiles = gr.HTML()
with gr.Row(elem_classes="act-chart-row"):
timeline = gr.HTML()
chart = gr.BarPlot(
x="stage", y="count", title="Activity by stage", height=280,
# rotate x labels so the stage names (ingest/vision/model/…)
# don't crowd into garbled overlap on a narrow phone chart
x_label_angle=-45,
)
gr.Markdown("### Run traces")
traces = gr.HTML()
gr.Markdown(
"**Sharing is Caring.** Publish a redacted run to a public Hugging Face "
"dataset so others can learn from it — one click (needs an `HF_WRITE_TOKEN` "
"secret on the Space). Or **⬇ Download** it (stays on your device) and publish "
"with `python training/share_trace.py`."
)
with gr.Row():
redact_chk = gr.Checkbox(label="Redact personal data", value=True)
trace_dl_btn = gr.Button("⬇ Download trace (JSON)")
trace_pub_btn = gr.Button("☁ Publish my trace to the Hub", variant="primary")
trace_file = gr.File(label="trace.json")
trace_status = gr.Markdown()
trace_dl_btn.click(
_on_download_trace, redact_chk, [trace_file, trace_status]
)
trace_pub_btn.click(_on_publish_trace, None, trace_status)
outputs = [stepper, tiles, timeline, chart, traces, impact_panel]
timer = gr.Timer(2.0)
timer.tick(_refresh, None, outputs)
demo.load(_refresh, None, outputs)
with gr.Tab("Memory", elem_classes="page-wrap"):
gr.Markdown(
"**What the agent knows about you.** Stored on *your device* "
"(your browser's localStorage) — never on our servers — and injected "
"into every extraction to personalize it. Active from your first "
"message; grows as you add facts, import contacts, or use the app."
)
mem_table = gr.Dataframe(
headers=["id", "kind", "text", "weight"],
datatype=["number", "str", "str", "number"],
interactive=False,
label="Memory (on this device)",
)
with gr.Row():
mem_text = gr.Textbox(label="Add a fact", scale=3,
placeholder="e.g. Dana is the soccer coach")
mem_kind = gr.Dropdown(
["note", "contact", "preference", "location"], value="note", label="Kind"
)
mem_add = gr.Button("Remember", variant="primary")
with gr.Row():
mem_fid = gr.Number(label="Forget id", precision=0)
mem_forget = gr.Button("Forget")
# Demo seed: a curated set of people/preferences/locations that
# map to the sample conversations, so visitors can see memory
# personalizing a run without typing facts first.
mem_samples = gr.Button("✨ Load sample memories",
elem_classes="rv-secondary", scale=0)
with gr.Accordion("📇 Import contacts / calendar (.vcf · CSV · .ics)", open=False):
mem_file = gr.File(label="Contacts (.vcf/CSV) or calendar (.ics)",
file_types=[".vcf", ".csv", ".ics"], type="filepath")
mem_import = gr.Button("Import to my memory")
gr.Markdown("Parsed **on your device's behalf, locally** — no cloud, no account.")
# Online workflow only — hidden while the site is in Offline mode.
with gr.Accordion("🗓 Import from Google Calendar (optional · cloud)",
open=False, visible=False) as gcal_import_acc:
gcal_ok = gr.Checkbox(label="Allow reading my Google Calendar to seed memory", value=False)
gcal_import = gr.Button("Import from Google Calendar")
gr.Markdown("_Opt-in cloud step. Everything else stays off-grid._")
mem_status = gr.Markdown()
# Every action renders the table DIRECTLY (third output) — the
# mem_box.change chain below still persists to localStorage, but
# the visible refresh no longer depends on that browser-side hop.
_mem_out = [mem_box, mem_status, mem_table]
mem_add.click(_with_rows(_mem_remember), [mem_text, mem_kind, mem_box], _mem_out)
mem_forget.click(_with_rows(_mem_forget), [mem_fid, mem_box], _mem_out)
mem_samples.click(_with_rows(_mem_load_samples), [mem_box], _mem_out)
mem_import.click(_with_rows(_import_file), [mem_file, mem_box], _mem_out)
gcal_import.click(_with_rows(_import_gcal), [gcal_ok, mem_box], _mem_out)
# Single source of truth: whenever the device memory changes, re-render
# the table AND persist back to the browser's localStorage.
mem_box.change(
lambda m: _facts_rows(_facts_load(m)), mem_box, mem_table
).then(
None, mem_box, None,
js="(m) => { try { localStorage.setItem('offgrid_memory', m || ''); } catch (e) {} }",
)
with gr.Tab("Feed", elem_classes="page-wrap"):
gr.Markdown(
"Messages pushed by the Mac collector arrive at `/ingest`. "
"**Privacy:** message contents only render when the deployment "
"sets `EXPOSE_FEED=1` — public builds keep them sealed."
)
feed = gr.JSON(label="Recent ingested messages", value=_read_feed())
gr.Button("Refresh").click(lambda: _read_feed(), None, feed)
with gr.Tab("About"):
gr.Markdown(
"### Who it's for\n"
"A busy parent whose kid's school and activity events are buried in a "
"noisy class group chat — picture day Thursday, the practice that moved "
"to Tuesday, the birthday-party RSVP. They read it once, mean to add it "
"later, and miss it. With this, they paste the chat (or a screenshot of "
"a flyer or invite) from their phone's browser and get back: the events, "
"a conflict check against their calendar, and a ready-to-send reply — all "
"surfaced for review before anything is saved. Output is a local .ics they "
"can add to any calendar, with optional Google Calendar push.\n\n"
"### Private by design\n"
"No app to install and no account. It reads nothing automatically — the "
"parent pastes only what they choose. Inference runs in the Space via "
"llama.cpp (no cloud AI APIs), and works out of the box with no GPU "
"(see Accuracy upgrade below).\n\n"
"### Accuracy upgrade\n"
f"**Model:** `{os.environ.get('MODEL_REPO', 'unset')}` / "
f"`{os.environ.get('MODEL_FILE', 'unset')}` · "
f"**mmproj:** `{os.environ.get('MMPROJ_FILE', 'unset')}`\n\n"
"The free **cpu-basic** preview runs a small local model so the app works "
"with no GPU. With a **GPU** enabled, the Space serves **gemma-cal E4B** — "
"our calendar-native fine-tune (~5 GB, runs on a 16 GB T4) with the "
"`mmproj` vision projector for screenshots and flyers. Every published "
"version is eval-gated (see docs/eval-roadmap.md). All inference stays in "
"the Space — no cloud AI APIs."
)
with gr.Tab("Submission"):
gr.HTML(_submission_html())
# --- MCP tool surface (hidden) ----------------------------------- #
# Each .click() with api_name=... registers a Gradio API endpoint, and
# with mcp_server=True on the launch/mount that endpoint is auto-exposed
# as a Model Context Protocol tool. Type hints + docstrings on the
# functions in server.mcp_tools become the MCP schema, so any MCP-aware
# agent (Claude Desktop, Cursor, etc.) can call this Space as a tool.
from server.mcp_tools import check_conflicts as _mcp_check_conflicts
from server.mcp_tools import extract_events as _mcp_extract_events
from server.mcp_tools import make_ics as _mcp_make_ics
# extract_events stays text-only via a hidden component: a gr.JSON
# `images` input renders an MCP schema of {"type": {}} (gradio can't type
# it), which small planners (MiniCPM) fill with `{}` -> rejected. The UI's
# vision path is separate (run_agent); the scripted path calls
# mcp_tools.extract_events(thread, images, memory) directly.
with gr.Row(visible=False):
_mcp_thread = gr.Textbox(visible=False)
_mcp_out_plan = gr.JSON(visible=False)
_mcp_btn_extract = gr.Button(visible=False)
_mcp_btn_extract.click(
fn=_mcp_extract_events,
inputs=[_mcp_thread],
outputs=_mcp_out_plan,
api_name="extract_events",
)
# make_ics / check_conflicts take list[dict] `events`. As a gr.JSON
# component that also renders {"type": {}}; registered via gr.api the MCP
# schema is typed from the function hints ("type": "array"), so the
# planner passes the real events list instead of a bare dict.
gr.api(_mcp_make_ics, api_name="make_ics")
gr.api(_mcp_check_conflicts, api_name="check_conflicts")
# --- Offline/Online: ONE decision point that reconfigures the site ----
# Each mode is a separate guided workflow: Offline = .ics in/out, no
# Google anywhere (the Step 2a connect block hides its Google CTA via
# CSS on #tool-card[data-mode]); Online = one-click Google connect in
# Step 2a, Google push in the export bar, Google import in Memory.
# The choice persists on the device and is restored on page load.
def _on_mode(m: str, rows, ag_rows, notes):
"""ONE atomic update per flip (visibility + both card re-renders) —
chained .then steps caused visible intermediate states."""
online = "Online" in (m or "")
return (
gr.update(), # cal_acc — always available (holds both paths)
gr.update(), # gcal_btn — always shown (export trio is fixed)
gr.update(), # gcal_link_html — hidden placeholder
gr.update(visible=online), # gcal_import_acc (Memory tab)
# ag_gcal_btn / ag_gcal_link stay hidden in BOTH modes while the
# per-event card links are the single export surface (the OAuth
# push returns when the Space is public).
gr.update(), # ag_gcal_btn (Agent tab)
gr.update(), # ag_gcal_link (Agent tab)
_mode_note_html(online), # mode_note
# cards (quick-add follows mode); pass notes through so the
# arrival-context callout survives a mode flip
_render_event_cards(rows, m, notes),
_render_event_cards(ag_rows, m), # ag_cards (no notes State on that tab)
)
# show_progress="hidden": no loading overlays flashing across the
# outputs (that flash was the "glitchy" feel); the theme itself is
# already recolored client-side at click time by wireModeTheme.
mode.change(
_on_mode, [mode, table, ag_table, found_notes],
[cal_acc, gcal_btn, gcal_link_html,
gcal_import_acc, ag_gcal_btn, ag_gcal_link, mode_note, cards, ag_cards],
show_progress="hidden",
).then(
None, mode, None,
js="(m) => { try { localStorage.setItem('offgrid_mode', m || ''); } catch (e) {} "
"var tc = document.getElementById('tool-card'); if (tc) tc.setAttribute("
"'data-mode', (m || '').indexOf('Online') >= 0 ? 'online' : 'offline'); }",
)
# Restore the device's last choice on load. Setting the radio fires
# .change (only when the stored value differs from the default), which
# re-applies every visibility update above.
demo.load(None, None, mode,
js="() => { try { return localStorage.getItem('offgrid_mode') || '🔌 Offline'; } "
"catch (e) { return '🔌 Offline'; } }")
# --- onboarding actions ---
# Save keeps the panel open so the confirmation is visible; Skip just
# collapses it — the panel stays on the page either way, so the agent
# can be personalized again at any time.
ob_save.click(_on_onboard, [ob_people, ob_reminder, ob_decline, mem_box],
[mem_box, onboard, ob_status])
ob_skip.click(lambda: gr.update(open=False), None, onboard)
# --- load per-user memory from the browser on page load ---
# Reading localStorage into mem_box triggers mem_box.change → the Memory
# table renders + memory re-persists. Returning visitors (device already
# has facts) get the onboarding collapsed — still there, one tap to open.
demo.load(None, None, mem_box,
js="() => { try { return localStorage.getItem('offgrid_memory') || ''; } "
"catch (e) { return ''; } }").then(
lambda m: gr.update(open=not _facts_load(m)), mem_box, onboard
)
return demo
# --------------------------------------------------------------------------- #
# Memory tab handlers
# --------------------------------------------------------------------------- #
def _memory_rows() -> list[list]:
return [[f["id"], f["kind"], f["text"], f.get("weight", 1)] for f in memory.list_facts()]
def _on_remember(text: str, kind: str):
if not (text or "").strip():
return _memory_rows(), "Type a fact first."
fact = memory.remember(text, kind or "note")
return _memory_rows(), f"Remembered: {fact['text']}"
def _on_forget(fact_id):
if fact_id is None or str(fact_id).strip() == "":
return _memory_rows(), "Enter an id to forget."
ok = memory.forget(int(fact_id))
return _memory_rows(), ("Forgotten." if ok else "No fact with that id.")
# --- client-owned (per-user, browser localStorage) memory --------------------
def _facts_load(s) -> list[dict]:
try:
data = json.loads(s) if s else []
except Exception: # noqa: BLE001
return []
# Tolerate the server memory-file shape ({"facts": [...]}) — silently
# treating it as empty would wipe those facts on the next save.
if isinstance(data, dict):
data = data.get("facts") or []
return [f for f in data if isinstance(f, dict) and f.get("text")] if isinstance(data, list) else []
def _facts_dump(facts) -> str:
return json.dumps(facts or [])
def _facts_rows(facts) -> list[list]:
return [[f.get("id"), f.get("kind", "note"), f.get("text", ""), f.get("weight", 1)]
for f in (facts or [])]
def _with_rows(handler):
"""Memory handlers return (mem_json, status); the click ALSO renders the
table directly, so the visible refresh never depends on the browser-side
mem_box.change hop (the data persisted fine, but the table could stay
stale until the next page load)."""
def run(*args):
mem_json, status = handler(*args)
return mem_json, status, _facts_rows(_facts_load(mem_json))
return run
def _mem_remember(text: str, kind: str, mem_json: str):
facts = _facts_load(mem_json)
if not (text or "").strip():
return _facts_dump(facts), "Type a fact first."
facts = memory.merge_facts(facts, [(text, kind or "note")], kind or "note")
return _facts_dump(facts), f"Remembered: {text.strip()}"
# Showcase memories for the demo: each maps to one of the sample conversations
# (Weekend Crew BBQ invite, the school-pickup juggle, the doctor-appointment
# confirmation) so a visitor can SEE memory personalizing the extraction —
# known people in attendees, the right reminder lead, a familiar location.
_SAMPLE_FACTS = [
("Alex hosts the Weekend Crew — get-togethers are usually in Alex's backyard", "contact"),
("Maya is in the Weekend Crew — always brings her famous potato salad, no arguments", "contact"),
("Jordan is in the Weekend Crew — will drink anything cold", "contact"),
("Party drinks: kombucha for Maya, lemonade or sparkling water for Alex", "preference"),
("RSVP to invites before the deadline — don't leave the host hanging", "preference"),
("Jenna is my partner — we trade school pickup when meetings collide", "contact"),
("School pickup is at 3:30 on weekdays", "note"),
("Primary Care of Manhattan (my doctor) is at 112A West 72nd Street, Upper West Side", "location"),
("Medical appointments: arrive 15 minutes early and remind me 60 minutes before", "preference"),
]
def _mem_load_samples(mem_json: str):
"""Seed the device memory with the showcase facts. merge_facts dedupes by
normalized text (reloading just bumps weights), so this is idempotent."""
facts = memory.merge_facts(_facts_load(mem_json), _SAMPLE_FACTS)
return _facts_dump(facts), (
f"✨ Loaded {len(_SAMPLE_FACTS)} sample memories — paste the BBQ invite, "
"a pickup chat, or an appointment text and watch them personalize the run."
)
def _mem_forget(fact_id, mem_json: str):
facts = _facts_load(mem_json)
if fact_id is None or str(fact_id).strip() == "":
return _facts_dump(facts), "Enter an id to forget."
try:
fid = int(fact_id)
except Exception: # noqa: BLE001
return _facts_dump(facts), "Invalid id."
new = [f for f in facts if int(f.get("id", -1)) != fid]
ok = len(new) != len(facts)
return _facts_dump(new), ("Forgotten." if ok else "No fact with that id.")
def _on_onboard(people: str, reminder, decline: str, mem_json: str):
"""First-run onboarding → write structured facts to the user's device memory."""
facts = _facts_load(mem_json)
new: list[tuple] = []
for line in (people or "").splitlines():
line = line.strip()
if not line:
continue
sep = "=" if "=" in line else ("-" if " - " in line else (":" if ":" in line else None))
if sep and sep in line:
name, role = line.split(sep, 1)
name, role = name.strip(), role.strip()
if name and role:
new.append((f"{name} is the {role}", "contact"))
continue
new.append((f"{line} is someone you make plans with", "contact"))
if reminder:
try:
new.append((f"Default reminder: {int(reminder)} minutes before events", "preference"))
except Exception: # noqa: BLE001
pass
if (decline or "").strip():
new.append((f"You usually decline events on: {decline.strip()}", "preference"))
facts = memory.merge_facts(facts, new)
msg = f"Saved {len(new)} fact(s) to your device." if new else "Nothing to save."
# Stay open so the confirmation is visible; the panel never leaves the page.
return _facts_dump(facts), gr.update(open=True), msg
def _parse_memory_file(path: str) -> tuple[list[str], list[tuple]]:
"""Local, off-grid parse of a contacts/calendar file → (contact names, pref facts)."""
names: list[str] = []
prefs: list[tuple] = []
try:
raw = Path(path).read_text(encoding="utf-8", errors="ignore")
except Exception: # noqa: BLE001
return names, prefs
low = path.lower()
if low.endswith(".vcf"):
for line in raw.splitlines():
if line.upper().startswith("FN:"):
nm = line[3:].strip()
if nm:
names.append(nm)
elif low.endswith(".csv"):
import csv as _csv
rdr = _csv.reader(raw.splitlines())
rows = list(rdr)
if rows:
header = [h.strip().lower() for h in rows[0]]
idx = next((i for i, h in enumerate(header)
if h in ("name", "full name", "fullname", "first name", "display name")), 0)
for r in rows[1:]:
if len(r) > idx and r[idx].strip():
names.append(r[idx].strip())
elif low.endswith(".ics"):
for line in raw.splitlines():
u = line.upper()
if u.startswith("SUMMARY:"):
s = line.split(":", 1)[1].strip()
if s:
prefs.append((f"You have an event like: {s}", "note"))
elif "ATTENDEE" in u and "CN=" in u:
try:
cn = line.split("CN=", 1)[1].split(":", 1)[0].split(";")[0].strip()
if cn:
names.append(cn)
except Exception: # noqa: BLE001
pass
# de-dup, cap
seen, uniq = set(), []
for n in names:
k = n.lower()
if k not in seen and len(n) <= 60:
seen.add(k)
uniq.append(n)
return uniq[:50], prefs[:20]
def _import_file(file_path, mem_json: str):
facts = _facts_load(mem_json)
if not file_path:
return _facts_dump(facts), "Choose a .vcf, .csv, or .ics file first."
names, prefs = _parse_memory_file(file_path)
new = [(f"{n} is a contact you make plans with", "contact") for n in names] + prefs
facts = memory.merge_facts(facts, new)
return (_facts_dump(facts),
f"Imported {len(names)} contact(s) and {len(prefs)} note(s) — saved to your device.")
def _import_gcal(enabled: bool, mem_json: str):
"""Opt-in Google Calendar import → contacts/locations as facts. Cloud opt-in;
degrades gracefully when Google libs/creds aren't configured on the Space."""
facts = _facts_load(mem_json)
if not enabled:
return _facts_dump(facts), "Tick the box to allow Google Calendar import."
try:
from calendar_out.gcal import read_recent_facts # lazy: google libs optional
names, locs = read_recent_facts()
except Exception as e: # noqa: BLE001 no creds / libs / offline -> graceful
return (_facts_dump(facts),
f"Google Calendar not connected on this Space ({type(e).__name__}). "
"Local-only memory is unaffected.")
new = [(f"{n} is a contact you make plans with", "contact") for n in names] \
+ [(f"You often have events at: {loc}", "location") for loc in locs]
facts = memory.merge_facts(facts, new)
return (_facts_dump(facts),
f"Imported {len(names)} contact(s) and {len(locs)} location(s) from Google Calendar.")
_FEED_PATH = os.environ.get("FEED_PATH", "/tmp/ingest_feed.json")
def _read_feed():
"""The feed holds RAW private iMessages (text, senders, images). Never
serve it from a public UI: gated behind EXPOSE_FEED=1, which only a
trusted/local deployment should set."""
if os.environ.get("EXPOSE_FEED") != "1":
return {"feed": "private",
"hint": "Set EXPOSE_FEED=1 on a trusted/local deployment to view "
"ingested messages here. Public builds never expose them."}
try:
return json.loads(Path(_FEED_PATH).read_text())
except Exception: # noqa: BLE001
return []
'
'OffGridSchedula'
'