Add wellness micro-habit recommender domain

README.md

@@ -31,6 +31,28 @@ If you plan to run the app in a conda or virtual environment, make sure to set u
 
 4. Once running, the app will be accessible at `localhost:8501`.
 
+### Wellness Micro-Habit Domain Demo
+
+This repository also includes a wellness micro-habit domain adaptation that recommends small everyday habits from a local catalog. It can use AO packages when installed, and it includes a deterministic fallback so reviewers can run it without private packages or paid APIs.
+
+Run the wellness demo:
+
+```bash
+streamlit run wellness_recommender.py
+```
+
+Run the fallback CLI:
+
+```bash
+python wellness_recommender.py
+```
+
+Run the wellness domain tests:
+
+```bash
+python -m unittest tests/test_wellness_domain.py
+```
+
 
 ### Docker Installation
 
@@ -55,10 +77,11 @@ You're done! Access the app at `localhost:8501` in your browser.
 
 The recommender system works by loading a set of random video links. Once the user hits the Run button, a video will be shown, and the system will suggest whether it recommends the video or not. The user can then provide feedback using "pain" or "pleasure" signals to guide the recommendation process. Based on this feedback, the system adjusts its responses and suggests another video. This cycle continues, allowing for more accurate and personalized recommendations over time.
 
+The wellness micro-habit demo follows the same continuous-feedback pattern with a different domain. It encodes each habit into the same eight-bit AO-compatible input shape using focus area, time bucket, social context, and the user's current goal. User feedback updates fallback rankings immediately and trains an AO Agent when optional AO packages are available.
+
 
 ## Contributing
 
 Fork the repository, make your changes, and submit a pull request for review. 
 
 
-

arch__WellnessRecommender.py

@@ -0,0 +1,15 @@
+# -*- coding: utf-8 -*-
+"""AO architecture for the wellness micro-habit recommender domain."""
+
+import ao_arch as ar
+
+
+description = "Wellness Micro-Habit Recommender System"
+
+# focus area, time bucket, social flag, current wellness goal
+arch_i = [3, 2, 1, 2]
+arch_z = [10]
+arch_c = []
+connector_function = "full_conn"
+
+arch = ar.Arch(arch_i, arch_z, arch_c, connector_function, description)

tests/test_wellness_domain.py

@@ -0,0 +1,69 @@
+import unittest
+
+from wellness_domain import (
+    WELLNESS_HABITS,
+    apply_feedback,
+    encode_wellness_habit,
+    recommend_wellness_habits,
+)
+
+
+class WellnessDomainTests(unittest.TestCase):
+    def test_encoding_matches_ao_input_shape(self):
+        habit = WELLNESS_HABITS[0]
+
+        encoded = encode_wellness_habit(habit, goal="calm")
+
+        self.assertEqual(len(encoded), 8)
+        self.assertTrue(all(bit in (0, 1) for bit in encoded))
+
+    def test_sleep_goal_prioritizes_sleep_or_reflection(self):
+        ranked = recommend_wellness_habits(
+            goal="better-sleep",
+            minutes_available=20,
+            limit=3,
+        )
+
+        self.assertTrue(
+            any(habit.focus in {"sleep", "reflection", "stress"} for habit, _ in ranked)
+        )
+
+    def test_deep_work_goal_prioritizes_focus(self):
+        ranked = recommend_wellness_habits(goal="deep-work", minutes_available=30, limit=3)
+
+        self.assertTrue(any(habit.focus == "focus" for habit, _ in ranked))
+
+    def test_social_filter_penalizes_social_habits(self):
+        ranked = recommend_wellness_habits(
+            goal="reset",
+            minutes_available=20,
+            social_ok=False,
+            limit=5,
+        )
+
+        self.assertTrue(all(not habit.social for habit, _ in ranked[:3]))
+
+    def test_feedback_changes_ranking(self):
+        baseline = recommend_wellness_habits(goal="calm", minutes_available=15, limit=1)[0][0]
+        target = "Long-form weekly reset"
+        feedback = {}
+        for _ in range(10):
+            feedback = apply_feedback(feedback, target, liked=True)
+
+        updated = recommend_wellness_habits(
+            goal="calm",
+            minutes_available=15,
+            feedback=feedback,
+            limit=1,
+        )[0][0]
+
+        self.assertNotEqual(baseline.name, updated.name)
+        self.assertEqual(updated.name, target)
+
+    def test_unknown_goal_is_rejected(self):
+        with self.assertRaises(ValueError):
+            recommend_wellness_habits(goal="unknown")
+
+
+if __name__ == "__main__":
+    unittest.main()

wellness_domain.py

@@ -0,0 +1,151 @@
+"""Wellness micro-habit recommendation domain for the AO recommender demo."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Iterable
+
+
+FOCUS_BITS = {
+    "stress": [0, 0, 0],
+    "sleep": [0, 0, 1],
+    "focus": [0, 1, 0],
+    "energy": [0, 1, 1],
+    "connection": [1, 0, 0],
+    "reflection": [1, 0, 1],
+}
+
+TIME_BITS = {
+    "short": [0, 0],
+    "medium": [0, 1],
+    "long": [1, 1],
+}
+
+GOAL_BITS = {
+    "calm": [0, 0],
+    "better-sleep": [0, 1],
+    "deep-work": [1, 0],
+    "reset": [1, 1],
+}
+
+
+@dataclass(frozen=True)
+class WellnessHabit:
+    """A small wellness habit that can be recommended."""
+
+    name: str
+    focus: str
+    time_bucket: str
+    social: bool
+    energy_level: str
+    minutes: int
+    contexts: tuple[str, ...]
+    description: str
+
+
+WELLNESS_HABITS: tuple[WellnessHabit, ...] = (
+    WellnessHabit("Two-minute box breathing", "stress", "short", False, "low", 2, ("calm", "reset"), "Use a simple breathing cadence before a meeting or after context switching."),
+    WellnessHabit("Write tomorrow's first task", "focus", "short", False, "low", 5, ("deep-work", "better-sleep"), "Close the day by choosing one concrete next action for tomorrow."),
+    WellnessHabit("Screen-off wind-down block", "sleep", "medium", False, "low", 20, ("better-sleep", "calm"), "Create a low-stimulation buffer before bedtime."),
+    WellnessHabit("Sunlight and water reset", "energy", "short", False, "medium", 10, ("reset", "deep-work"), "Step outside, hydrate, and give your attention a clean restart."),
+    WellnessHabit("Focus sprint with phone away", "focus", "medium", False, "medium", 25, ("deep-work", "reset"), "Work on one task with notifications out of reach."),
+    WellnessHabit("Send one appreciation note", "connection", "short", True, "low", 5, ("calm", "reset"), "Strengthen a relationship with a quick specific message."),
+    WellnessHabit("Midday walk without audio", "reflection", "medium", False, "medium", 15, ("calm", "reset"), "Let your mind settle while moving at an easy pace."),
+    WellnessHabit("Tidy one visible surface", "energy", "short", False, "medium", 8, ("reset", "deep-work"), "Remove one small source of visual friction from your workspace."),
+    WellnessHabit("Guided body scan", "stress", "medium", False, "low", 12, ("calm", "better-sleep"), "Notice tension and relax one area at a time."),
+    WellnessHabit("Plan a low-effort social check-in", "connection", "medium", True, "medium", 15, ("calm", "reset"), "Pick one person and one easy way to reconnect this week."),
+    WellnessHabit("Evening reflection note", "reflection", "short", False, "low", 7, ("better-sleep", "calm"), "Write what worked today and what can be lighter tomorrow."),
+    WellnessHabit("Long-form weekly reset", "reflection", "long", False, "medium", 35, ("reset", "deep-work"), "Review commitments, remove stale tasks, and choose the next week's focus."),
+)
+
+
+def encode_wellness_habit(habit: WellnessHabit, goal: str = "calm") -> list[int]:
+    """Encode a wellness habit plus user goal into the AO eight-bit input shape."""
+
+    if habit.focus not in FOCUS_BITS:
+        raise ValueError(f"Unknown focus: {habit.focus}")
+    if habit.time_bucket not in TIME_BITS:
+        raise ValueError(f"Unknown time bucket: {habit.time_bucket}")
+    if goal not in GOAL_BITS:
+        raise ValueError(f"Unknown goal: {goal}")
+
+    social_bit = [1 if habit.social else 0]
+    return FOCUS_BITS[habit.focus] + TIME_BITS[habit.time_bucket] + social_bit + GOAL_BITS[goal]
+
+
+def score_wellness_habit(
+    habit: WellnessHabit,
+    goal: str,
+    minutes_available: int,
+    preferred_energy: str = "low",
+    social_ok: bool = True,
+    feedback: dict[str, int] | None = None,
+) -> int:
+    """Score a habit using deterministic context preferences plus feedback."""
+
+    score = 0
+    if goal in habit.contexts:
+        score += 35
+    if habit.minutes <= minutes_available:
+        score += 20
+    else:
+        score -= (habit.minutes - minutes_available) // 5 * 5
+    if habit.energy_level == preferred_energy:
+        score += 14
+    if not social_ok and habit.social:
+        score -= 18
+    if goal == "better-sleep" and habit.focus in {"sleep", "reflection", "stress"}:
+        score += 12
+    if goal == "deep-work" and habit.focus in {"focus", "energy"}:
+        score += 12
+    if goal == "calm" and habit.energy_level == "low":
+        score += 8
+    if feedback:
+        score += feedback.get(habit.name, 0) * 12
+    return score
+
+
+def recommend_wellness_habits(
+    goal: str = "calm",
+    minutes_available: int = 15,
+    preferred_energy: str = "low",
+    social_ok: bool = True,
+    feedback: dict[str, int] | None = None,
+    habits: Iterable[WellnessHabit] = WELLNESS_HABITS,
+    limit: int = 5,
+) -> list[tuple[WellnessHabit, int]]:
+    """Return wellness recommendations sorted from strongest to weakest match."""
+
+    if goal not in GOAL_BITS:
+        raise ValueError(f"Unknown goal: {goal}")
+    if preferred_energy not in {"low", "medium"}:
+        raise ValueError(f"Unknown preferred energy: {preferred_energy}")
+
+    ranked = [
+        (
+            habit,
+            score_wellness_habit(
+                habit,
+                goal=goal,
+                minutes_available=minutes_available,
+                preferred_energy=preferred_energy,
+                social_ok=social_ok,
+                feedback=feedback,
+            ),
+        )
+        for habit in habits
+    ]
+    ranked.sort(key=lambda item: (item[1], -item[0].minutes, item[0].name), reverse=True)
+    return ranked[:limit]
+
+
+def apply_feedback(
+    feedback: dict[str, int] | None,
+    habit_name: str,
+    liked: bool,
+) -> dict[str, int]:
+    """Return updated feedback weights for a wellness habit."""
+
+    updated = dict(feedback or {})
+    updated[habit_name] = updated.get(habit_name, 0) + (1 if liked else -1)
+    return updated