docs/CONTRACTS.md

LifeOS internal contracts

Interfaces that feature modules, the server, and the frontend code against. Keep these stable; change only with a coordinated update.

memory.py (short-term structured store)

load() -> dict            # full memory dict (first call: blank, or demo persona if LIFEOS_DEMO=1)
save(mem: dict) -> None
empty_data() -> dict      # blank store schema (new-user default)
first_run_data() -> dict  # empty_data(), or seed_data() when LIFEOS_DEMO=1
reset_to_empty() -> dict  # wipe to blank slate (real-user reset)
reset_to_seed() -> dict   # load demo persona (used by demo mode)
log_meal(dish: str, ingredients: list[str], when: str | None = None) -> dict
log_workout(workout_type: str, duration_min: int, when: str | None = None) -> dict
set_subscriptions(subs: list[dict]) -> dict   # [{name, cost, last_used}]
recent_meals(days=7, mem=None) -> list[dict]          # [{date, dish, ingredients[]}]
workouts_in_window(days=14, mem=None) -> list[dict]   # [{date, type, duration_min}] sorted by date
add_event(ev: dict) -> dict                   # fills id (uuid4 hex[:8]) if missing; returns full mem
delete_event(event_id: str) -> dict
set_workout_schedule(days: list[str], time: str) -> dict   # days normalized to 3-letter lowercase
set_monthly_payments(payments: list[dict]) -> dict          # [{name, amount, due_day}]
upsert_goal(goal: dict) -> dict               # update by id; insert (new id) when id missing/unknown
delete_goal(goal_id: str) -> dict
set_section(section: str, items: list) -> dict  # wholesale replace; sections: "meals" | "workouts" | "calendar" | "subscriptions" | "monthly_payments"; raises ValueError on unknown section
set_profile(fields: dict) -> dict             # merges into user_profile; "income_monthly" routes to finances; recomputes name = first_name + " " + last_name
events_in_window(days=7, mem=None) -> list[dict]  # calendar events today..today+days, sorted (date, start)

Schema:

{
  "meals": [{"date": "YYYY-MM-DD", "dish": str, "ingredients": [str]}],
  "workouts": [{"date": "YYYY-MM-DD", "type": str, "duration_min": int}],
  "calendar": [{"id": "8-char hex", "title": str, "date": "YYYY-MM-DD", "start": "HH:MM", "end": "HH:MM", "kind": "work|class|social|other"}],
  "workout_schedule": {"days": ["mon", "wed", "sat"], "time": "HH:MM"},
  "goals": [{"id": "8-char hex", "title": str, "target_amount": num, "saved": num, "deadline": "YYYY-MM", "notes": str}],
  "finances": {
    "income_monthly": num,
    "monthly_payments": [{"name": str, "amount": num, "due_day": int}],
    "subscriptions": [{"name": str, "cost": num, "last_used": "YYYY-MM-DD"}]
  },
  "user_profile": {
    "first_name": str, "last_name": str,
    "name": str,            // derived: first_name + " " + last_name (kept for back-compat, recomputed on load/set_profile)
    "dietary_prefs": [str], "fitness_goal": str, "budget_weekly": num,
    "address": str,         // street
    "city": str, "postal_code": str, "country": str
  }
}

load() backfills missing new keys from empty_data() (never demo data), so old memory.json files upgrade in place. Legacy profiles with only name get it split on the first space into first_name/last_name; goals without an id get one (uuid4 hex[:8]).

rag.py (long-term memory, local RAG)

remember(text: str, kind: str = "fact", meta: dict | None = None) -> dict   # kind: fact|event|preference
recall(query: str, k: int = 5, kind: str | None = None) -> list[dict]       # [{text, kind, score}]
embed(text: str, is_query: bool = False) -> np.ndarray   # normalized; tests monkeypatch this to skip the embedder
ensure_seeded() -> int            # seeds demo notes (called only when LIFEOS_DEMO=1)
reset_to_seed() -> int
reset_to_empty() -> int           # delete all notes (real-user reset)
list_notes() -> list[dict]                  # [{id, text, kind}] — no vectors
update_note(note_id: str, text: str) -> bool  # re-embeds; preserves kind/meta; False if id unknown
delete_note(note_id: str) -> bool

Every note has an id (8-char hex, backfilled on load for old stores).

engine.py (reasoning)

build_prompt(domain: str, mem: dict, user_input: str, domains: list[str] | None = None) -> list[dict]  # [system, user]; domains narrows the memory slice to just those domains + profile (None = default slice)
slice_for_domains(mem: dict, domains: list[str]) -> dict  # merged _slice_for_domain outputs; "kitchen" aliases to "food"; user_profile always included
generate_stream(messages, max_tokens=1024, temperature=0.4, domain="chat", extra_context="") -> Iterator[str]  # yields CUMULATIVE, reasoning-stripped text; extra_context (e.g. web results) is appended to the last user message; on model-load failure yields one friendly message instead of raising
run_domain(domain: str, user_input: str = "", max_tokens=1024) -> Iterator[str]
describe_food_image(path: str) -> str   # vision model: deduped bulleted food items from a photo (raises ModelUnavailable if VLM can't load); downscales to config.VLM_MAX_IMAGE_SIDE first (~4x faster on the CPU path)
status() -> dict                        # {state: idle|loading|ready|error, backend, error}
warmup() -> None

Domains: "food" | "health" | "money" | "chat" | "goal" | "meal_photo" | "payment_impact". goal is a Socratic financial-goal coach (one question per turn, plan after ~3-4 exchanges); meal_photo analyzes the food items a vision model identified in a photo (or text read from a receipt); payment_impact is a short note (streamed after saving monthly payments) on how those payments affect each savings goal's timeline. Memory slices: health gets calendar (next 7 days) + workout_schedule; money/goal/payment_impact get monthly_payments + goals + income; chat gets everything. Streams yield the full text-so-far each step (not deltas). Prompts handle a nameless (new) user gracefully. Tests inject a fake engine._llm and rag.embed rather than loading real models.

This Nemotron GGUF reasons in plain prose and ignores /no_think. The prompt ends with a recency nudge telling it to jump straight to an ==ANSWER== delimiter; _clean_response strips everything before the answer (delimiter or, as a fallback, the first markdown block), hiding reasoning during streaming so the UI shows its "thinking…" state until the real answer begins.

Feature modules (Phase B deliverables)

features/food.py

extract_deals(path: str) -> list[dict]   # [{item: str, price_text: str}] from PDF (pdfplumber) or image (pytesseract)
extract_deals_from_text(text: str) -> list[dict]   # same, from pasted text
shortlist(deals, recent_meals, profile) -> list[dict]  # top 6 recipe candidates, deterministic
build_food_input(deals, mem) -> str      # the user_input string handed to engine.run_domain("food", ...)
RECIPES: list[dict]  # {name, main: str, ingredients: [str], tags: [str], protein_g: int, cost_tier: 1|2|3}

features/health.py

weekly_pattern(workouts: list[dict]) -> dict  # {by_type: {...}, total_min, days_since_rest, last_7: [{date,type,duration_min}|None x7], consecutive_training_days}
build_health_input(mem: dict) -> str
plan_workout_slots(schedule: dict, events: list[dict], today=None) -> list[dict]
# slots over next 7 days on preferred days; slot = time..time+1h; an event clashes
# when [start,end) overlaps. [{date, day, time, free: bool, clashes: [{id,title,start,end}]}]

features/websearch.py

search_web(query: str, max_results: int = 5) -> list[dict]  # [{title, snippet, url}]; [] on ANY failure (offline-safe)
format_results(results: list[dict]) -> str   # readable block for prompt injection ("" for [])

features/money.py

parse_transactions(csv_text: str) -> list[dict]   # [{date: "YYYY-MM-DD", merchant: str, amount: float}] tolerant headers
detect_recurring(txns: list[dict]) -> list[dict]  # [{name, cost, occurrences, last_charged, cadence_days}] — >=2 charges, median interval 25-35d, amount stable ±15%
build_money_input(subs: list[dict], mem: dict) -> str

All feature logic must be pure/deterministic and unit-tested in tests/ (plain asserts runnable via python tests/test_<name>.py, no pytest dep, no model calls).

Server endpoints (app.py — gr.Server)

Streaming @app.api endpoints (SSE generators, cumulative text):

name	args (in order, all strings)	behavior
food_recommend	deals_json	3 recipe picks from flyer deals
health_recommend	—	tomorrow's workout recommendation
money_review	subs_json	subscription CANCEL/KEEP/WATCH review
chat	message, history_json, use_web, refs_json	general chat; use_web="1" injects DuckDuckGo results as extra context ("" = off); refs_json = JSON array of domain refs like ["kitchen","health"] ("kitchen" normalizes to "food"; allowed: kitchen/food, health, money) — when present the prompt contains ONLY those domains' memory slices + profile; "" = full context (backward-compatible). NOTE: the wire protocol requires all 4 values in data — send "" for refs off
goal_chat	message, history_json, goal_json	Socratic goal coaching; goal_json is one goal object (or ""); history format same as chat: [{role, content}]
local_deals	city	web-searches " grocery store weekly flyer deals", extracts deals, streams a food-domain recommendation (falls back to a search summary; offline-safe message if no results). city="" uses profile city
meal_analyze	ocr_text, meal_label	meal_photo-domain analysis of OCR text; non-empty meal_label is logged via log_meal first
payment_impact	—	streams a short note on how the saved monthly payments affect each savings goal's timeline; one line if no goals exist. Called by the UI right after set_payments

Non-streaming @app.api (each returns the FULL memory dict as a JSON string, except get_memory/health_plan as noted):

name	args	notes
log_workout	workout_type, duration	existing
get_memory	—	full memory JSON
set_profile	profile_json	merge fields into user_profile; accepts first_name, last_name, address, city, postal_code, country, income_monthly, dietary_prefs, fitness_goal, budget_weekly; income_monthly routes to finances; name recomputed from first/last
upsert_goal	goal_json	one goal object {id?, title, target_amount, saved, deadline, notes}; updates by id, inserts (new id) when missing; returns full mem JSON
delete_goal	goal_id	returns full mem JSON
edit_memory	section, items_json	wholesale-replace a section (meals, workouts, calendar, subscriptions, monthly_payments); returns full mem JSON, or {"error": "..."} on unknown section
list_notes	—	JSON array [{id, text, kind}] of long-term notes (no vectors)
update_note	note_id, text	re-embeds; returns {"ok": true}
delete_note	note_id	returns {"ok": true}
add_event	event_json	{title, date, start, end, kind}; id auto-assigned
delete_event	event_id
set_schedule	days_json, time	days_json = JSON array like ["mon","wed"], time = "HH:MM"
set_payments	payments_json	JSON array [{name, amount, due_day}]
health_plan	—	deterministic JSON string: {"schedule": {...}, "next_7_days_events": [...], "recommendation_slots": [{date, day, time, free, clashes}]}
reset_demo	—	load demo persona; returns {"error": ...} unless LIFEOS_DEMO=1
reset_account	—	wipe all data to a blank slate (real-user reset); returns full mem JSON

FastAPI routes: POST /upload/flyer, POST /upload/flyer_text, POST /upload/transactions, POST /upload/meal_photo (multipart image → {"text": items, "source": "vision"|"ocr"} — vision model identifies food items, OCR fallback for receipts — or {"error": "..."} if both fail), GET /status ({state, backend, error, demo}), static frontend at /.

Frontend calls REST directly (no CDN client), protocol unchanged: POST /gradio_api/call/<name> with {"data": [...]} → {event_id}; then GET /gradio_api/call/<name>/<event_id> is an SSE stream of data: lines (JSON arrays of outputs).