Deploy Habit Journal backend S0-S10 to Hugging Face Space.

EOF

Co-authored-by: Cursor <cursoragent@cursor.com>

.gitignore

@@ -0,0 +1,6 @@
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.venv/
+.env
+data/

Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.11-slim
+WORKDIR /app
+RUN mkdir -p /data/loop_logger
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+COPY . .
+ENV PYTHONUNBUFFERED=1
+CMD uvicorn app.main:app --host 0.0.0.0 --port ${PORT:-7860}

README.md

@@ -0,0 +1,37 @@
+---
+title: Habit Journal
+emoji: 📓
+colorFrom: gray
+colorTo: blue
+sdk: docker
+app_port: 7860
+pinned: false
+---
+
+# Habit Journal API
+
+Personal habit journal API designed for a Hugging Face Docker Space. The
+backend uses signed cookie sessions and durable JSONL files under `DATA_ROOT`.
+
+## Run locally
+
+Set `APP_PASSWORD` and `APP_SECRET_KEY`, install `requirements.txt`, then run:
+
+```text
+uvicorn app.main:app --host 0.0.0.0 --port 7860
+```
+
+`ENV=dev` enables the OpenAPI endpoints for local development.
+
+## Environment
+
+- `APP_PASSWORD`: initial password used only when `config.json` is absent
+- `APP_SECRET_KEY`: required session-signing key
+- `DATA_ROOT`: storage directory (default `/data/loop_logger`)
+- `ENV`: runtime environment (default `prod`)
+- `APP_NAME`: displayed application name (default `Habit Journal`)
+- `SESSION_MAX_AGE_SEC`: signed-session lifetime (default `604800`)
+- `LOGIN_RATE_LIMIT`: login attempts per window (default `10`)
+- `LOGIN_RATE_WINDOW_SEC`: login window in seconds (default `900`)
+
+Additional coach-related variables are documented in `docs/BACKEND.md`.

app/__init__.py

@@ -0,0 +1,4 @@
+"""Habit Journal backend package.
+
+The package exposes a small FastAPI application backed by durable local files.
+"""

app/backup_replies.py

@@ -0,0 +1,283 @@
+"""Hardcoded coach backup rules ordered by priority.
+
+Selects the highest-priority matching rule; fills SERVER_PICK when allowed.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+RULES: list[dict[str, Any]] = [
+    {
+        "id": "crisis",
+        "priority": 100,
+        "keywords": ["suicide", "kill myself", "self-harm", "self harm", "end it"],
+        "tags": ["crisis"],
+        "template": (
+            "LOOP: crisis\n"
+            "FEELING: overwhelm\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: Contact local emergency / crisis resources now\n"
+            "NEXT_BRICK: S\n"
+            "DO_NOT: Isolate or escalate alone\n"
+            "LINE: Get help before anything else\n"
+            "NOTE: This is not therapy; seek real support\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+    {
+        "id": "court",
+        "priority": 90,
+        "keywords": ["court", "argue with myself", "verdict"],
+        "tags": ["court"],
+        "template": (
+            "LOOP: court\n"
+            "FEELING: judgment\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: {{server_pick}}\n"
+            "NEXT_BRICK: E\n"
+            "DO_NOT: Reopen the case tonight\n"
+            "LINE: Close court; one brick only\n"
+            "NOTE: Silence after one boundary line\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+    {
+        "id": "rerun",
+        "priority": 80,
+        "keywords": ["rerun", "replay", "looping the scene"],
+        "tags": ["rerun"],
+        "template": (
+            "LOOP: rerun\n"
+            "FEELING: stuck\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: {{server_pick}}\n"
+            "NEXT_BRICK: S\n"
+            "DO_NOT: Replay the scene again\n"
+            "LINE: Interrupt and move body once\n"
+            "NOTE: Water or short walk\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+    {
+        "id": "rage_movie",
+        "priority": 70,
+        "keywords": ["rage", "movie", "fantasy fight"],
+        "tags": ["rage", "movie"],
+        "template": (
+            "LOOP: rage_movie\n"
+            "FEELING: heat\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: {{server_pick}}\n"
+            "NEXT_BRICK: B\n"
+            "DO_NOT: Feed the movie\n"
+            "LINE: Name it and stop the reel\n"
+            "NOTE: One hygiene or environment act\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+    {
+        "id": "daydream_fc",
+        "priority": 60,
+        "keywords": ["daydream", "fantasy", "fc"],
+        "tags": ["daydream", "fc"],
+        "template": (
+            "LOOP: daydream\n"
+            "FEELING: drift\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: {{server_pick}}\n"
+            "NEXT_BRICK: A\n"
+            "DO_NOT: Skip today's brick\n"
+            "LINE: Brick first, then optional rest\n"
+            "NOTE: Daydream only after brick\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+    {
+        "id": "urge_corn",
+        "priority": 55,
+        "keywords": ["urge", "corn", "delay"],
+        "tags": ["urge", "corn"],
+        "template": (
+            "LOOP: urge\n"
+            "FEELING: pull\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: {{server_pick}}\n"
+            "NEXT_BRICK: C\n"
+            "DO_NOT: Stack sessions\n"
+            "LINE: Delay, then one brick\n"
+            "NOTE: Ceiling is one session\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+    {
+        "id": "bully",
+        "priority": 50,
+        "keywords": ["bully", "harsh voice", "attack myself"],
+        "tags": ["bully"],
+        "template": (
+            "LOOP: bully\n"
+            "FEELING: harsh\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: {{server_pick}}\n"
+            "NEXT_BRICK: E\n"
+            "DO_NOT: Argue with the voice\n"
+            "LINE: One boundary line then silence\n"
+            "NOTE: No pep talk\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+    {
+        "id": "shame",
+        "priority": 40,
+        "keywords": ["shame", "embarrassed", "humiliated"],
+        "tags": ["shame"],
+        "template": (
+            "LOOP: shame\n"
+            "FEELING: shame\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: {{server_pick}}\n"
+            "NEXT_BRICK: B\n"
+            "DO_NOT: Hide and ruminate\n"
+            "LINE: Small body reset, then brick\n"
+            "NOTE: Keep it practical\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+    {
+        "id": "spain_admin",
+        "priority": 35,
+        "keywords": ["admin", "paperwork", "travel", "deadline"],
+        "tags": ["admin", "travel"],
+        "template": (
+            "LOOP: admin\n"
+            "FEELING: pressure\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: {{server_pick}}\n"
+            "NEXT_BRICK: A\n"
+            "DO_NOT: Cancel committed process from fear\n"
+            "LINE: One admin checklist item\n"
+            "NOTE: Fear is not a cancel signal\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+    {
+        "id": "build_earn",
+        "priority": 30,
+        "keywords": ["build", "earn", "ship", "work"],
+        "tags": ["build", "earn"],
+        "template": (
+            "LOOP: build\n"
+            "FEELING: drive\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: {{server_pick}}\n"
+            "NEXT_BRICK: C\n"
+            "DO_NOT: Open five fronts\n"
+            "LINE: Ship one small unit\n"
+            "NOTE: One earn/build act\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+    {
+        "id": "home",
+        "priority": 20,
+        "keywords": ["home", "mess", "environment"],
+        "tags": ["home"],
+        "template": (
+            "LOOP: home\n"
+            "FEELING: clutter\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: {{server_pick}}\n"
+            "NEXT_BRICK: B\n"
+            "DO_NOT: Redesign everything\n"
+            "LINE: One hygiene or tidy act\n"
+            "NOTE: Environment first\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+    {
+        "id": "loneliness",
+        "priority": 10,
+        "keywords": ["lonely", "alone", "isolation"],
+        "tags": ["lonely"],
+        "template": (
+            "LOOP: loneliness\n"
+            "FEELING: lonely\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: {{server_pick}}\n"
+            "NEXT_BRICK: D\n"
+            "DO_NOT: Spiral into absence stories\n"
+            "LINE: One logistics line, then brick\n"
+            "NOTE: Keep contact practical\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+    {
+        "id": "default",
+        "priority": 0,
+        "keywords": [],
+        "tags": [],
+        "template": (
+            "LOOP: default\n"
+            "FEELING: mixed\n"
+            "INTENSITY_GUESS: n/a\n"
+            "REMEDY: {{server_pick}}\n"
+            "NEXT_BRICK: A\n"
+            "DO_NOT: Overthink the next hour\n"
+            "LINE: Do the next small brick\n"
+            "NOTE: Checklist over story\n"
+            "DATA_THIN: {data_thin}"
+        ),
+    },
+]
+
+
+DEFAULT_PICK = "Do one listed brick: water, admin item, or tidy act"
+
+
+def select_backup_rule(
+    text: str,
+    tags: list[str],
+) -> dict[str, Any]:
+    """Return the highest-priority rule matching keywords or tags."""
+
+    haystack = (text or "").lower()
+    tag_set = {t.strip().lower() for t in tags if t and t.strip()}
+    ordered = sorted(RULES, key=lambda rule: rule["priority"], reverse=True)
+    for rule in ordered:
+        keywords = rule.get("keywords") or []
+        rule_tags = {t.lower() for t in (rule.get("tags") or [])}
+        if any(keyword in haystack for keyword in keywords):
+            return rule
+        if tag_set & rule_tags:
+            return rule
+    return ordered[-1]
+
+
+def render_backup(
+    rule: dict[str, Any],
+    *,
+    server_picks: list[dict[str, Any]],
+    data_thin: bool,
+) -> str:
+    """Fill the selected rule template with pick and DATA_THIN."""
+
+    pick = server_picks[0]["remedy_key"] if server_picks else DEFAULT_PICK
+    if rule["id"] == "crisis":
+        text = rule["template"]
+    else:
+        text = rule["template"].replace("{{server_pick}}", pick)
+    return text.format(data_thin=str(data_thin).lower())
+
+
+def backup_reply(
+    text: str,
+    tags: list[str],
+    server_picks: list[dict[str, Any]],
+    *,
+    data_thin: bool,
+) -> tuple[str, str]:
+    """Select and render a backup reply; return (text, rule_id)."""
+
+    rule = select_backup_rule(text, tags)
+    return render_backup(rule, server_picks=server_picks, data_thin=data_thin), rule["id"]

app/coach_service.py

@@ -0,0 +1,478 @@
+"""Coach pipeline: evidence, OpenRouter (optional), parse, and backup.
+
+Always returns HTTP-ready text with source=model|backup|model_unparsed_fallback.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import time
+from datetime import date, datetime, timezone
+from typing import Any
+from uuid import uuid4
+
+import httpx
+
+from app.backup_replies import backup_reply
+from app.config import APP_VERSION, Settings
+from app.evidence import build_evidence, strip_sensitive
+from app.models import CoachMeta, Entry
+from app.store_daily import DailyStore
+from app.store_entries import EntryStore
+from app.store_traces import TraceStore
+
+DEFAULT_POLICY = """Bricks: A admin/critical; B environment/hygiene; C build/earn; D logistics; E boundary then silence; S stop spiral (water/shower/sleep)."""
+
+OUTPUT_FORMAT = """Output format exactly (first line LOOP:):
+LOOP: <tag>
+FEELING: <word>
+INTENSITY_GUESS: <1-10|n/a>
+REMEDY: <one concrete action>
+NEXT_BRICK: <A|B|C|D|E|S|specific>
+DO_NOT: <one thing>
+LINE: <≤12 words>
+NOTE: <optional ≤20 words>
+DATA_THIN: <true|false>"""
+
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def parse_template(raw: str) -> dict[str, str] | None:
+    """Parse line-based coach template; fail if LOOP: is missing."""
+
+    text = raw.strip()
+    if text.startswith("```"):
+        text = re.sub(r"^```(?:\w+)?\s*", "", text)
+        text = re.sub(r"\s*```$", "", text)
+    fields: dict[str, str] = {}
+    for line in text.splitlines():
+        if ":" not in line:
+            continue
+        key, value = line.split(":", 1)
+        key = key.strip().upper()
+        value = value.strip()
+        if key:
+            fields[key] = value
+    if "LOOP" not in fields:
+        return None
+    return fields
+
+
+def format_parsed(parsed: dict[str, str]) -> str:
+    """Rebuild canonical template text from parsed fields."""
+
+    order = [
+        "LOOP",
+        "FEELING",
+        "INTENSITY_GUESS",
+        "REMEDY",
+        "NEXT_BRICK",
+        "DO_NOT",
+        "LINE",
+        "NOTE",
+        "DATA_THIN",
+    ]
+    lines = []
+    for key in order:
+        if key in parsed:
+            lines.append(f"{key}: {parsed[key]}")
+    return "\n".join(lines)
+
+
+def truncate_entry(entry: Entry) -> dict[str, Any]:
+    """Compact entry fields for coach history."""
+
+    happened = entry.happened[:280]
+    return {
+        "id": entry.id,
+        "ts": entry.ts.isoformat(),
+        "tags": entry.tags,
+        "emotions": entry.emotions,
+        "intensity": entry.intensity,
+        "remedy": entry.remedy,
+        "result": entry.result.value,
+        "happened": happened,
+    }
+
+
+class CoachService:
+    """Build evidence-backed coach replies with model or backup fallback."""
+
+    def __init__(
+        self,
+        settings: Settings,
+        entry_store: EntryStore,
+        daily_store: DailyStore,
+        trace_store: TraceStore,
+    ) -> None:
+        self.settings = settings
+        self.entry_store = entry_store
+        self.daily_store = daily_store
+        self.trace_store = trace_store
+
+    def _all_entry_dicts(self) -> list[dict[str, Any]]:
+        return [e.model_dump(mode="json") for e in self.entry_store._load()]
+
+    def _today_daily(self) -> dict[str, Any] | None:
+        row = self.daily_store.get(date.today())
+        return row.model_dump(mode="json") if row else None
+
+    def _daily_dicts(self) -> list[dict[str, Any]]:
+        return [r.model_dump(mode="json") for r in self.daily_store._load()]
+
+    def build_messages(
+        self,
+        *,
+        brief: str,
+        evidence_block: str,
+        picks_text: str,
+        current: dict[str, Any],
+        today_daily: dict[str, Any] | None,
+        history: list[dict[str, Any]],
+        data_thin_flag: bool,
+    ) -> list[dict[str, str]]:
+        """Assemble short system/user messages for the free model."""
+
+        system = "\n\n".join(
+            [
+                "Role: brief checklist coach. No pity. No pep talk.",
+                "NEVER invent numbers — only EVIDENCE / SERVER_PICKS.",
+                "NEVER cancel committed admin/travel process; fear ≠ cancel.",
+                "If self-harm language → crisis redirect, no deep exploration.",
+                OUTPUT_FORMAT,
+                DEFAULT_POLICY,
+                strip_sensitive(brief),
+            ]
+        )
+        user_payload = {
+            "current": current,
+            "today_daily": today_daily,
+            "recent_entries": history,
+            "EVIDENCE": evidence_block,
+            "SERVER_PICKS": picks_text,
+            "DATA_THIN": data_thin_flag,
+            "instruction": (
+                "Choose REMEDY consistent with SERVER_PICKS when possible. "
+                "Begin with LOOP:"
+            ),
+        }
+        user = strip_sensitive(json.dumps(user_payload, ensure_ascii=False, indent=2))
+        return [
+            {"role": "system", "content": system},
+            {"role": "user", "content": user},
+        ]
+
+    def call_openrouter(
+        self,
+        messages: list[dict[str, str]],
+    ) -> tuple[str | None, int | None, str | None, int]:
+        """Call OpenRouter; return content, http_status, error, latency_ms."""
+
+        if not self.settings.openrouter_api_key:
+            return None, None, "NO_API_KEY", 0
+        started = time.perf_counter()
+        url = f"{self.settings.openrouter_base_url.rstrip('/')}/chat/completions"
+        headers = {
+            "Authorization": f"Bearer {self.settings.openrouter_api_key}",
+            "Content-Type": "application/json",
+            "X-Title": "Habit Journal",
+        }
+        body = {
+            "model": self.settings.openrouter_model,
+            "messages": messages,
+            "temperature": self.settings.coach_temperature,
+            "max_tokens": self.settings.coach_max_tokens,
+        }
+        try:
+            with httpx.Client(timeout=self.settings.coach_timeout_sec) as client:
+                response = client.post(url, headers=headers, json=body)
+            latency_ms = int((time.perf_counter() - started) * 1000)
+            if response.status_code >= 400:
+                return None, response.status_code, f"http_{response.status_code}", latency_ms
+            payload = response.json()
+            content = (
+                payload.get("choices", [{}])[0]
+                .get("message", {})
+                .get("content")
+            )
+            if not content or not str(content).strip():
+                return None, response.status_code, "EMPTY_MODEL", latency_ms
+            return str(content), response.status_code, None, latency_ms
+        except httpx.TimeoutException:
+            latency_ms = int((time.perf_counter() - started) * 1000)
+            return None, None, "TIMEOUT", latency_ms
+        except Exception as exc:  # noqa: BLE001 — fail-open to backup
+            latency_ms = int((time.perf_counter() - started) * 1000)
+            return None, None, f"error:{type(exc).__name__}", latency_ms
+
+    def coach(
+        self,
+        *,
+        text: str | None = None,
+        entry_id: str | None = None,
+        include_history: int | None = None,
+        persist: bool = False,
+        force_backup: bool = False,
+        request_meta: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        """Run the full coach pipeline and always return a text reply."""
+
+        if not text and not entry_id:
+            raise ValueError("require text or entry_id")
+
+        entry: Entry | None = None
+        if entry_id:
+            entry = self.entry_store.get(entry_id)
+            if entry is None:
+                raise KeyError(entry_id)
+
+        current_text = text or ""
+        current_tags: list[str] = []
+        current_emotions: list[str] = []
+        if entry:
+            current_tags = list(entry.tags)
+            current_emotions = list(entry.emotions)
+            if not current_text:
+                current_text = f"{entry.activity}. {entry.happened}"
+        current = {
+            "text": current_text,
+            "entry_id": entry.id if entry else None,
+            "tags": current_tags,
+            "emotions": current_emotions,
+            "intensity": entry.intensity if entry else None,
+            "remedy": entry.remedy if entry else None,
+            "result": entry.result.value if entry else None,
+        }
+
+        history_k = (
+            self.settings.coach_history_k
+            if include_history is None
+            else max(0, include_history)
+        )
+        all_entries = self.entry_store._load()
+        all_entries.sort(key=lambda item: item.ts, reverse=True)
+        history = [truncate_entry(item) for item in all_entries[:history_k]]
+        today_daily = self._today_daily()
+        evidence = build_evidence(
+            self._all_entry_dicts(),
+            self._daily_dicts(),
+            current_tags,
+            min_n=self.settings.min_stats_n,
+            shrink_k=self.settings.stats_shrink_k,
+            match_alpha=self.settings.match_alpha,
+        )
+        picks = evidence["server_picks"]
+        brief_meta = self.trace_store.brief_meta(
+            include_full=self.settings.debug_include_full_brief
+        )
+        brief = self.trace_store.read_brief()
+        messages = self.build_messages(
+            brief=brief,
+            evidence_block=evidence["evidence_block"],
+            picks_text=evidence["server_picks_text"],
+            current=current,
+            today_daily=today_daily,
+            history=history,
+            data_thin_flag=evidence["DATA_THIN"],
+        )
+
+        trace_id = str(uuid4())
+        flags: list[str] = []
+        if evidence["DATA_THIN"]:
+            flags.append("DATA_THIN")
+
+        raw = None
+        http_status = None
+        error = None
+        latency_ms = 0
+        source = "backup"
+        parsed = None
+        backup_rule_id = None
+        model_id = self.settings.openrouter_model
+
+        if force_backup or not self.settings.openrouter_api_key:
+            if not self.settings.openrouter_api_key:
+                flags.append("NO_API_KEY")
+            final_text, backup_rule_id = backup_reply(
+                current_text,
+                current_tags + current_emotions,
+                picks,
+                data_thin=evidence["DATA_THIN"],
+            )
+            source = "backup"
+            parsed = parse_template(final_text)
+        else:
+            raw, http_status, error, latency_ms = self.call_openrouter(messages)
+            if error == "TIMEOUT":
+                flags.append("TIMEOUT")
+            elif error == "EMPTY_MODEL":
+                flags.append("EMPTY_MODEL")
+            if raw:
+                parsed = parse_template(raw)
+                if parsed:
+                    source = "model"
+                    final_text = format_parsed(parsed)
+                    remedy = (parsed.get("REMEDY") or "").strip().lower()
+                    pick_keys = {p["remedy_key"].lower() for p in picks}
+                    if picks and remedy and not any(
+                        key in remedy or remedy in key for key in pick_keys
+                    ):
+                        flags.append("OUT_OF_EVIDENCE")
+                else:
+                    flags.append("PARSE_FAIL")
+                    source = "model_unparsed_fallback"
+                    final_text, backup_rule_id = backup_reply(
+                        current_text,
+                        current_tags + current_emotions,
+                        picks,
+                        data_thin=evidence["DATA_THIN"],
+                    )
+                    parsed = parse_template(final_text)
+            else:
+                final_text, backup_rule_id = backup_reply(
+                    current_text,
+                    current_tags + current_emotions,
+                    picks,
+                    data_thin=evidence["DATA_THIN"],
+                )
+                source = "backup"
+                parsed = parse_template(final_text)
+
+        now = utc_now()
+        trace = {
+            "trace_id": trace_id,
+            "ts": now.isoformat(),
+            "request": request_meta
+            or {
+                "text": text,
+                "entry_id": entry_id,
+                "include_history": history_k,
+                "persist": persist,
+                "force_backup": force_backup,
+            },
+            "current": current,
+            "history_truncated": history,
+            "evidence": {
+                "n_scored": evidence["n_scored"],
+                "DATA_THIN": evidence["DATA_THIN"],
+                "evidence_block": evidence["evidence_block"],
+                "by_remedy": evidence["by_remedy"],
+                "daily": evidence["daily"],
+            },
+            "server_picks": picks,
+            "brief_sha256": brief_meta["brief_sha256"],
+            "brief_excerpt": brief_meta["brief_excerpt"],
+            "brief_full": brief_meta["brief_full"],
+            "system_prompt": messages[0]["content"],
+            "user_prompt": messages[1]["content"],
+            "model_id": model_id if source != "backup" or raw else model_id,
+            "temperature": self.settings.coach_temperature,
+            "max_tokens": self.settings.coach_max_tokens,
+            "latency_ms": latency_ms,
+            "http_status": http_status,
+            "error": error,
+            "raw_model_response": raw,
+            "parsed": parsed,
+            "source": source,
+            "backup_rule_id": backup_rule_id,
+            "flags": flags,
+            "app_version": APP_VERSION,
+            "final_text": final_text,
+        }
+        self.trace_store.append_trace(trace)
+
+        if persist and entry is not None:
+            self.entry_store.set_coach(
+                entry.id,
+                CoachMeta(
+                    text=final_text,
+                    source=source,
+                    model=model_id if source == "model" else None,
+                    ts=now,
+                    trace_id=trace_id,
+                ),
+            )
+
+        return {
+            "text": final_text,
+            "source": source,
+            "model": model_id if source == "model" else None,
+            "trace_id": trace_id,
+            "flags": flags,
+            "server_picks": [
+                {
+                    "remedy_key": p["remedy_key"],
+                    "pick": p["pick"],
+                    "n": p["n"],
+                    "p_helped": p["p_helped"],
+                }
+                for p in picks
+            ],
+            "parsed": parsed,
+        }
+
+    def prompt_preview(
+        self,
+        *,
+        text: str | None = None,
+        entry_id: str | None = None,
+        include_history: int | None = None,
+    ) -> dict[str, Any]:
+        """Build evidence and messages without calling the model."""
+
+        if not text and not entry_id:
+            raise ValueError("require text or entry_id")
+
+        entry: Entry | None = None
+        if entry_id:
+            entry = self.entry_store.get(entry_id)
+            if entry is None:
+                raise KeyError(entry_id)
+
+        current_text = text or ""
+        current_tags: list[str] = list(entry.tags) if entry else []
+        current_emotions: list[str] = list(entry.emotions) if entry else []
+        if entry and not current_text:
+            current_text = f"{entry.activity}. {entry.happened}"
+        current = {
+            "text": current_text,
+            "entry_id": entry.id if entry else None,
+            "tags": current_tags,
+            "emotions": current_emotions,
+        }
+        history_k = (
+            self.settings.coach_history_k
+            if include_history is None
+            else max(0, include_history)
+        )
+        all_entries = self.entry_store._load()
+        all_entries.sort(key=lambda item: item.ts, reverse=True)
+        history = [truncate_entry(item) for item in all_entries[:history_k]]
+        evidence = build_evidence(
+            self._all_entry_dicts(),
+            self._daily_dicts(),
+            current_tags,
+            min_n=self.settings.min_stats_n,
+            shrink_k=self.settings.stats_shrink_k,
+            match_alpha=self.settings.match_alpha,
+        )
+        brief = self.trace_store.read_brief()
+        messages = self.build_messages(
+            brief=brief,
+            evidence_block=evidence["evidence_block"],
+            picks_text=evidence["server_picks_text"],
+            current=current,
+            today_daily=self._today_daily(),
+            history=history,
+            data_thin_flag=evidence["DATA_THIN"],
+        )
+        return {
+            "evidence_block": evidence["evidence_block"],
+            "server_picks": evidence["server_picks"][:5],
+            "system_prompt": messages[0]["content"],
+            "user_prompt": messages[1]["content"],
+            "DATA_THIN": evidence["DATA_THIN"],
+        }

app/config.py

@@ -0,0 +1,58 @@
+"""Load immutable runtime settings from environment variables.
+
+Settings are created once by the application factory and passed to services.
+"""
+
+from functools import lru_cache
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+APP_VERSION = "2.0.0"
+
+
+class Settings(BaseSettings):
+    """Environment-backed application configuration."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+    app_password: str | None = None
+    app_secret_key: str = Field(min_length=1)
+    data_root: str = "/data/loop_logger"
+    env: str = "prod"
+    app_name: str = "Habit Journal"
+
+    openrouter_api_key: str | None = None
+    openrouter_model: str = "openrouter/free"
+    openrouter_base_url: str = "https://openrouter.ai/api/v1"
+    coach_timeout_sec: float = 25
+    coach_temperature: float = 0.3
+    coach_max_tokens: int = 400
+    coach_history_k: int = 10
+    coach_trace_limit: int = 200
+
+    min_stats_n: int = 5
+    stats_shrink_k: float = 3
+    match_alpha: float = 0.5
+    session_max_age_sec: int = 604800
+    debug_include_full_brief: bool = False
+    login_rate_limit: int = 10
+    login_rate_window_sec: int = 900
+    max_body_bytes: int = 65536
+
+    @property
+    def is_prod(self) -> bool:
+        """Return whether production-only security behavior is enabled."""
+
+        return self.env.lower() == "prod"
+
+
+@lru_cache
+def get_settings() -> Settings:
+    """Read and cache environment settings."""
+
+    return Settings()

app/deps.py

@@ -0,0 +1,46 @@
+"""Provide FastAPI dependencies for protected routes.
+
+Session validity is checked against expiry and durable session version.
+"""
+
+import time
+from typing import Any
+
+from fastapi import HTTPException, Request, status
+
+from app.models import StoredConfig
+
+
+def api_http_error(status_code: int, code: str, message: str) -> HTTPException:
+    """Create an HTTP exception understood by the envelope handler."""
+
+    return HTTPException(
+        status_code=status_code,
+        detail={"code": code, "message": message},
+    )
+
+
+def require_login(request: Request) -> StoredConfig:
+    """Require a current signed session and return durable config."""
+
+    config: StoredConfig | None = request.app.state.config_store.load()
+    if config is None:
+        raise api_http_error(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            "setup_required",
+            "Application setup is required",
+        )
+
+    session: dict[str, Any] = request.session
+    authenticated = session.get("auth") is True
+    version_matches = session.get("sv") == config.session_version
+    expiry = session.get("exp")
+    unexpired = isinstance(expiry, (int, float)) and expiry >= time.time()
+    if not (authenticated and version_matches and unexpired):
+        request.session.clear()
+        raise api_http_error(
+            status.HTTP_401_UNAUTHORIZED,
+            "unauthorized",
+            "Login required",
+        )
+    return config

app/evidence.py

@@ -0,0 +1,145 @@
+"""Build coach evidence blocks and SERVER_PICKS from server math.
+
+Formats precomputed probabilities for prompts; never invents numbers.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from app.stats_math import (
+    FORMULAS,
+    by_emotion,
+    by_remedy,
+    by_tag,
+    daily_rates,
+    data_thin,
+    scored_entries,
+    server_picks,
+)
+
+
+def strip_sensitive(text: str) -> str:
+    """Best-effort strip of emails and long digit runs from prompt text."""
+
+    import re
+
+    cleaned = re.sub(
+        r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b",
+        "[email]",
+        text,
+    )
+    cleaned = re.sub(r"\b\d{8,}\b", "[digits]", cleaned)
+    return cleaned
+
+
+def format_evidence_block(
+    entries: list[dict[str, Any]],
+    daily_rows: list[dict[str, Any]],
+    *,
+    min_n: int,
+    shrink_k: float,
+) -> str:
+    """Render the EVIDENCE markdown block for coach prompts."""
+
+    scored = scored_entries(entries)
+    remedies = by_remedy(entries, min_n=min_n, shrink_k=shrink_k)
+    tags = by_tag(entries)[:8]
+    emotions = by_emotion(entries)[:8]
+    rates = daily_rates(daily_rows)
+    lines = [
+        "EVIDENCE (server-computed; do not invent numbers)",
+        f"n_scored: {len(scored)}",
+        f"min_n: {min_n}",
+        "TOP_REMEDIES: remedy | n | p_worked | p_helped | rank",
+    ]
+    if remedies:
+        for row in remedies[:10]:
+            lines.append(
+                f"- {row['key']} | {row['n']} | {row['p_worked']:.3f} | "
+                f"{row['p_helped']:.3f} | {row['rank']:.3f}"
+            )
+    else:
+        lines.append("- (none above min_n)")
+    lines.append("WORST_TAGS: tag | n | p_fail")
+    if tags:
+        for row in tags:
+            lines.append(f"- {row['key']} | {row['n']} | {row['p_failed']:.3f}")
+    else:
+        lines.append("- (none)")
+    lines.append("EMOTION_HITS: emotion | n | p_helped")
+    if emotions:
+        for row in emotions:
+            lines.append(f"- {row['key']} | {row['n']} | {row['p_helped']:.3f}")
+    else:
+        lines.append("- (none)")
+    lines.append(
+        "DAILY_RATES: "
+        f"p_brick_done={rates['p_brick_done']:.3f} "
+        f"p_corn_ok={rates['p_corn_ok']:.3f} "
+        f"p_no_fc={rates['p_no_fc']:.3f} "
+        f"p_rerun_clean={rates['p_rerun_clean']:.3f} "
+        f"p_court_closed={rates['p_court_closed']:.3f} "
+        f"avg_points={rates['avg_points']:.3f}"
+    )
+    lines.append(
+        "FORMULAS: "
+        f"p_helped={FORMULAS['p_helped']}; rank={FORMULAS['rank']}"
+    )
+    lines.append(f"DATA_THIN: {str(data_thin(len(scored))).lower()}")
+    return "\n".join(lines)
+
+
+def format_server_picks(picks: list[dict[str, Any]]) -> str:
+    """Render numbered SERVER_PICKS lines for prompts and debug paste."""
+
+    if not picks:
+        return "SERVER_PICKS: (none)"
+    lines = ["SERVER_PICKS:"]
+    for index, pick in enumerate(picks, start=1):
+        lines.append(
+            f"{index}) {pick['remedy_key']} pick={pick['pick']:.3f} "
+            f"n={pick['n']} p_helped={pick['p_helped']:.3f}"
+        )
+    return "\n".join(lines)
+
+
+def build_evidence(
+    entries: list[dict[str, Any]],
+    daily_rows: list[dict[str, Any]],
+    current_tags: list[str],
+    *,
+    min_n: int,
+    shrink_k: float,
+    match_alpha: float,
+) -> dict[str, Any]:
+    """Return structured evidence plus formatted blocks and picks."""
+
+    scored = scored_entries(entries)
+    picks = server_picks(
+        entries,
+        current_tags,
+        min_n=min_n,
+        shrink_k=shrink_k,
+        match_alpha=match_alpha,
+    )
+    block = format_evidence_block(
+        entries,
+        daily_rows,
+        min_n=min_n,
+        shrink_k=shrink_k,
+    )
+    picks_text = format_server_picks(picks)
+    return {
+        "n_scored": len(scored),
+        "min_n": min_n,
+        "DATA_THIN": data_thin(len(scored)),
+        "by_remedy": by_remedy(entries, min_n=min_n, shrink_k=shrink_k),
+        "by_tag": by_tag(entries),
+        "by_emotion": by_emotion(entries),
+        "daily": daily_rates(daily_rows),
+        "server_picks": picks,
+        "evidence_block": block,
+        "server_picks_text": picks_text,
+        "formulas": FORMULAS,
+    }

app/fsutil.py

@@ -0,0 +1,110 @@
+"""Provide locked JSON/JSONL persistence and atomic file replacement.
+
+Mutations use a sibling lock file so append and full rewrites cannot overlap.
+"""
+
+import json
+import os
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any, Iterator
+
+
+@contextmanager
+def file_lock(path: Path) -> Iterator[None]:
+    """Hold an exclusive process lock for a data path."""
+
+    lock_path = path.with_name(f"{path.name}.lock")
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    with lock_path.open("a+b") as handle:
+        if os.name == "nt":
+            import msvcrt
+
+            if handle.tell() == 0:
+                handle.write(b"\0")
+                handle.flush()
+            handle.seek(0)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
+        else:
+            import fcntl
+
+            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+        try:
+            yield
+        finally:
+            if os.name == "nt":
+                handle.seek(0)
+                msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
+            else:
+                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+
+
+def atomic_write_text(path: Path, text: str) -> None:
+    """Replace a file atomically after flushing its contents to disk."""
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(f"{path.name}.tmp")
+    with temporary.open("w", encoding="utf-8", newline="\n") as handle:
+        handle.write(text)
+        handle.flush()
+        os.fsync(handle.fileno())
+    os.replace(temporary, path)
+
+
+def read_json(path: Path) -> dict[str, Any] | None:
+    """Read a JSON object, returning None when the file is absent."""
+
+    if not path.exists():
+        return None
+    with path.open("r", encoding="utf-8") as handle:
+        value = json.load(handle)
+    if not isinstance(value, dict):
+        raise ValueError(f"{path.name} must contain a JSON object")
+    return value
+
+
+def write_json(path: Path, value: dict[str, Any]) -> None:
+    """Write a JSON object under a lock using atomic replacement."""
+
+    payload = json.dumps(value, ensure_ascii=False, indent=2) + "\n"
+    with file_lock(path):
+        atomic_write_text(path, payload)
+
+
+def read_jsonl(path: Path) -> list[dict[str, Any]]:
+    """Read non-empty JSONL records in file order."""
+
+    if not path.exists():
+        return []
+    records: list[dict[str, Any]] = []
+    with path.open("r", encoding="utf-8") as handle:
+        for line in handle:
+            if line.strip():
+                value = json.loads(line)
+                if not isinstance(value, dict):
+                    raise ValueError(f"{path.name} contains a non-object record")
+                records.append(value)
+    return records
+
+
+def append_jsonl(path: Path, value: dict[str, Any]) -> None:
+    """Append one durable JSONL record while holding the file lock."""
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    line = json.dumps(value, ensure_ascii=False, separators=(",", ":")) + "\n"
+    with file_lock(path):
+        with path.open("a", encoding="utf-8", newline="\n") as handle:
+            handle.write(line)
+            handle.flush()
+            os.fsync(handle.fileno())
+
+
+def rewrite_jsonl(path: Path, values: list[dict[str, Any]]) -> None:
+    """Atomically replace a JSONL file while holding its lock."""
+
+    text = "".join(
+        json.dumps(value, ensure_ascii=False, separators=(",", ":")) + "\n"
+        for value in values
+    )
+    with file_lock(path):
+        atomic_write_text(path, text)

app/main.py

@@ -0,0 +1,177 @@
+"""Create and wire the Habit Journal FastAPI application.
+
+The factory initializes durable paths, middleware, errors, and API routers.
+"""
+
+import logging
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from pathlib import Path
+from typing import Any
+
+from fastapi import FastAPI, Request
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+from fastapi.staticfiles import StaticFiles
+from starlette.exceptions import HTTPException as StarletteHTTPException
+from starlette.middleware.sessions import SessionMiddleware
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+from app.config import get_settings
+from app.coach_service import CoachService
+from app.models import err
+from app.paths import Paths
+from app.routers import auth, coach, daily, debug, entries, export, health, stats
+from app.routers import settings as settings_router
+from app.store_config import ConfigStore
+from app.store_daily import DailyStore
+from app.store_entries import EntryStore
+from app.store_traces import TraceStore
+
+logger = logging.getLogger(__name__)
+
+
+class BodyTooLarge(Exception):
+    """Signal that an HTTP request exceeded the configured byte limit."""
+
+
+class BodySizeMiddleware:
+    """Reject HTTP request bodies larger than the configured limit."""
+
+    def __init__(self, app: ASGIApp, max_bytes: int) -> None:
+        self.app = app
+        self.max_bytes = max_bytes
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        headers = dict(scope.get("headers", []))
+        raw_length = headers.get(b"content-length")
+        if raw_length:
+            try:
+                if int(raw_length) > self.max_bytes:
+                    await self._reject(scope, receive, send)
+                    return
+            except ValueError:
+                pass
+
+        size = 0
+
+        async def limited_receive() -> Message:
+            nonlocal size
+            message = await receive()
+            if message["type"] == "http.request":
+                size += len(message.get("body", b""))
+                if size > self.max_bytes:
+                    raise BodyTooLarge
+            return message
+
+        try:
+            await self.app(scope, limited_receive, send)
+        except BodyTooLarge:
+            await self._reject(scope, receive, send)
+
+    async def _reject(self, scope: Scope, receive: Receive, send: Send) -> None:
+        response = err("validation_error", "Request body too large", 413)
+        await response(scope, receive, send)
+
+
+def create_app() -> FastAPI:
+    """Build an application from environment settings."""
+
+    settings = get_settings()
+    paths = Paths(settings.data_root)
+    paths.ensure()
+    config_store = ConfigStore(paths)
+    config_store.bootstrap(settings.app_password)
+    entry_store = EntryStore(paths)
+    daily_store = DailyStore(paths)
+    trace_store = TraceStore(paths, trace_limit=settings.coach_trace_limit)
+    trace_store.ensure_brief()
+    coach_service = CoachService(settings, entry_store, daily_store, trace_store)
+
+    @asynccontextmanager
+    async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
+        paths.ensure()
+        trace_store.ensure_brief()
+        logger.info("Habit Journal backend ready")
+        yield
+
+    hidden_docs: dict[str, str | None] = {}
+    if settings.is_prod:
+        hidden_docs = {"docs_url": None, "redoc_url": None, "openapi_url": None}
+
+    app = FastAPI(title=settings.app_name, lifespan=lifespan, **hidden_docs)
+    app.state.settings = settings
+    app.state.paths = paths
+    app.state.config_store = config_store
+    app.state.entry_store = entry_store
+    app.state.daily_store = daily_store
+    app.state.trace_store = trace_store
+    app.state.coach_service = coach_service
+
+    app.add_middleware(BodySizeMiddleware, max_bytes=settings.max_body_bytes)
+    app.add_middleware(
+        SessionMiddleware,
+        secret_key=settings.app_secret_key,
+        max_age=settings.session_max_age_sec,
+        same_site="lax",
+        https_only=settings.is_prod,
+    )
+
+    @app.exception_handler(StarletteHTTPException)
+    async def http_exception_handler(
+        _request: Request,
+        exc: StarletteHTTPException,
+    ) -> JSONResponse:
+        detail: Any = exc.detail
+        if isinstance(detail, dict):
+            code = detail.get("code", "internal")
+            message = detail.get("message", "Request failed")
+        else:
+            code = {
+                401: "unauthorized",
+                403: "forbidden",
+                404: "not_found",
+                422: "validation_error",
+                429: "rate_limited",
+                503: "setup_required",
+            }.get(exc.status_code, "internal")
+            message = "Request failed"
+        return err(code, message, exc.status_code)
+
+    @app.exception_handler(RequestValidationError)
+    async def validation_exception_handler(
+        _request: Request,
+        _exc: RequestValidationError,
+    ) -> JSONResponse:
+        return err("validation_error", "Invalid request", 422)
+
+    @app.exception_handler(Exception)
+    async def internal_exception_handler(
+        _request: Request,
+        exc: Exception,
+    ) -> JSONResponse:
+        logger.exception("Unhandled API error", exc_info=exc)
+        return err("internal", "Internal server error", 500)
+
+    app.include_router(health.router)
+    app.include_router(auth.router)
+    app.include_router(entries.router)
+    app.include_router(daily.router)
+    app.include_router(stats.router)
+    app.include_router(coach.router)
+    app.include_router(debug.router)
+    app.include_router(export.router)
+    app.include_router(settings_router.router)
+
+    static_dir = Path("static")
+    if static_dir.is_dir():
+        app.mount("/", StaticFiles(directory=static_dir, html=True), name="static")
+
+    return app
+
+
+app = create_app()

app/models.py

@@ -0,0 +1,382 @@
+"""Define API request, response, and persisted configuration models.
+
+Envelope helpers keep every endpoint response shape consistent.
+"""
+
+from datetime import date, datetime
+from enum import Enum
+from typing import Any, Literal
+
+from fastapi.encoders import jsonable_encoder
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+ErrorCode = Literal[
+    "unauthorized",
+    "forbidden",
+    "validation_error",
+    "not_found",
+    "setup_required",
+    "rate_limited",
+    "internal",
+]
+
+
+class ApiError(BaseModel):
+    """Structured API failure details."""
+
+    code: ErrorCode
+    message: str
+
+
+class ApiEnvelope(BaseModel):
+    """Common success or failure response wrapper."""
+
+    ok: bool
+    data: Any | None
+    error: ApiError | None
+
+
+class StoredConfig(BaseModel):
+    """Durable password and session invalidation state."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    password_hash: str
+    session_version: int = Field(ge=1)
+    created_at: datetime
+    updated_at: datetime
+
+
+class LoginRequest(BaseModel):
+    """Password login payload."""
+
+    password: str = Field(min_length=1)
+
+
+class ChangePasswordRequest(BaseModel):
+    """Authenticated password rotation payload."""
+
+    old_password: str = Field(min_length=1)
+    new_password: str = Field(min_length=1)
+
+
+class AuthenticatedData(BaseModel):
+    """Authentication state returned after a session mutation."""
+
+    authenticated: bool
+
+
+class MeData(AuthenticatedData):
+    """Public session status and discrete application name."""
+
+    app_name: str
+
+
+class HealthData(BaseModel):
+    """Liveness information."""
+
+    status: Literal["up"]
+    time: datetime
+
+
+class ReadyData(BaseModel):
+    """Bootstrap and storage readiness information."""
+
+    ready: bool
+    data_writable: bool
+    config_present: bool
+
+
+class Result(str, Enum):
+    """Outcome of a logged remedy; `pending` is excluded from stats."""
+
+    worked = "worked"
+    partial = "partial"
+    failed = "failed"
+    pending = "pending"
+
+
+def normalize_emotions(values: list[str]) -> list[str]:
+    """Lowercase, strip, drop blanks, and de-duplicate emotion labels."""
+
+    out: list[str] = []
+    seen: set[str] = set()
+    for value in values:
+        token = value.strip().lower()
+        if token and token not in seen:
+            seen.add(token)
+            out.append(token)
+    return out
+
+
+def normalize_tags(values: list[str]) -> list[str]:
+    """Lowercase, strip, collapse spaces to underscores, and de-duplicate."""
+
+    out: list[str] = []
+    seen: set[str] = set()
+    for value in values:
+        token = "_".join(value.strip().lower().split())
+        if token and token not in seen:
+            seen.add(token)
+            out.append(token)
+    return out
+
+
+class CoachMeta(BaseModel):
+    """Coach reply metadata persisted on an entry."""
+
+    text: str | None = None
+    source: str | None = None
+    model: str | None = None
+    ts: datetime | None = None
+    trace_id: str | None = None
+
+
+class EntryCreate(BaseModel):
+    """Validated payload for creating a log entry."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    activity: str = Field(min_length=1, max_length=500)
+    happened: str = Field(min_length=1, max_length=4000)
+    emotions: list[str] = Field(default_factory=list, max_length=12)
+    intensity: int = Field(ge=1, le=10)
+    remedy: str = Field(default="", max_length=2000)
+    result: Result
+    tags: list[str] = Field(default_factory=list, max_length=20)
+    notes: str = Field(default="", max_length=2000)
+    ts: datetime | None = None
+
+    @field_validator("activity", "happened", mode="after")
+    @classmethod
+    def _strip_required(cls, value: str) -> str:
+        stripped = value.strip()
+        if not stripped:
+            raise ValueError("must not be blank")
+        return stripped
+
+    @field_validator("remedy", "notes", mode="after")
+    @classmethod
+    def _strip_optional(cls, value: str) -> str:
+        return value.strip()
+
+    @field_validator("emotions", mode="after")
+    @classmethod
+    def _clean_emotions(cls, value: list[str]) -> list[str]:
+        return normalize_emotions(value)
+
+    @field_validator("tags", mode="after")
+    @classmethod
+    def _clean_tags(cls, value: list[str]) -> list[str]:
+        return normalize_tags(value)
+
+
+class EntryUpdate(BaseModel):
+    """Partial update payload; only provided fields are applied."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    activity: str | None = Field(default=None, min_length=1, max_length=500)
+    happened: str | None = Field(default=None, min_length=1, max_length=4000)
+    emotions: list[str] | None = Field(default=None, max_length=12)
+    intensity: int | None = Field(default=None, ge=1, le=10)
+    remedy: str | None = Field(default=None, max_length=2000)
+    result: Result | None = None
+    tags: list[str] | None = Field(default=None, max_length=20)
+    notes: str | None = Field(default=None, max_length=2000)
+    ts: datetime | None = None
+
+    @field_validator("activity", "happened", mode="after")
+    @classmethod
+    def _strip_required(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        stripped = value.strip()
+        if not stripped:
+            raise ValueError("must not be blank")
+        return stripped
+
+    @field_validator("remedy", "notes", mode="after")
+    @classmethod
+    def _strip_optional(cls, value: str | None) -> str | None:
+        return value.strip() if value is not None else None
+
+    @field_validator("emotions", mode="after")
+    @classmethod
+    def _clean_emotions(cls, value: list[str] | None) -> list[str] | None:
+        return normalize_emotions(value) if value is not None else None
+
+    @field_validator("tags", mode="after")
+    @classmethod
+    def _clean_tags(cls, value: list[str] | None) -> list[str] | None:
+        return normalize_tags(value) if value is not None else None
+
+
+class Entry(BaseModel):
+    """A persisted log entry as stored on one JSONL line."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    id: str
+    ts: datetime
+    created_at: datetime
+    updated_at: datetime
+    activity: str
+    happened: str
+    emotions: list[str] = Field(default_factory=list)
+    intensity: int = Field(ge=1, le=10)
+    remedy: str = ""
+    result: Result
+    tags: list[str] = Field(default_factory=list)
+    notes: str = ""
+    coach: CoachMeta = Field(default_factory=CoachMeta)
+
+
+class EntryListData(BaseModel):
+    """Paginated entry listing with total match count."""
+
+    items: list[Entry]
+    total: int
+    limit: int
+    offset: int
+
+
+class DeletedData(BaseModel):
+    """Confirmation payload for a hard delete."""
+
+    deleted: bool
+    id: str
+
+
+class PrimaryBrick(str, Enum):
+    """Daily primary brick label."""
+
+    A = "A"
+    B = "B"
+    C = "C"
+    D = "D"
+    E = "E"
+    S = "S"
+    none = "none"
+
+
+class Daydream(str, Enum):
+    """Daily daydream status."""
+
+    none = "none"
+    done = "done"
+    fc = "fc"
+
+
+class Rerun(str, Enum):
+    """Daily rerun status."""
+
+    clean = "clean"
+    R = "R"
+
+
+class Court(str, Enum):
+    """Daily court status."""
+
+    closed = "closed"
+    court = "court"
+
+
+class DailyUpsert(BaseModel):
+    """Client payload for upserting a daily scoreboard row."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    primary_brick: PrimaryBrick = PrimaryBrick.none
+    brick_done: bool = False
+    corn_sessions: int = Field(default=0, ge=0, le=50)
+    delay_ok: bool = True
+    daydream: Daydream = Daydream.none
+    rerun: Rerun = Rerun.clean
+    court: Court = Court.closed
+    note: str = Field(default="", max_length=2000)
+
+    @field_validator("note", mode="after")
+    @classmethod
+    def _strip_note(cls, value: str) -> str:
+        return value.strip()
+
+
+class DailyRow(BaseModel):
+    """A persisted daily scoreboard row; points are server-authoritative."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    date: date
+    primary_brick: PrimaryBrick = PrimaryBrick.none
+    brick_done: bool = False
+    corn_sessions: int = Field(default=0, ge=0, le=50)
+    delay_ok: bool = True
+    daydream: Daydream = Daydream.none
+    rerun: Rerun = Rerun.clean
+    court: Court = Court.closed
+    points: int = Field(ge=0, le=6)
+    note: str = ""
+    updated_at: datetime
+
+
+class DailyRangeData(BaseModel):
+    """Daily rows for a date range plus week band summary."""
+
+    items: list[DailyRow]
+    week_points: int
+    band: Literal["incomplete", "strong", "mixed", "escape_heavy"]
+    days_present: int
+
+
+class CoachRequest(BaseModel):
+    """Coach request: free text and/or entry id."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    text: str | None = None
+    entry_id: str | None = None
+    include_history: int | None = Field(default=None, ge=0, le=50)
+    persist: bool = False
+    force_backup: bool = False
+
+
+class CoachResponseData(BaseModel):
+    """Coach reply always includes text and source."""
+
+    text: str
+    source: Literal["model", "backup", "model_unparsed_fallback"]
+    model: str | None = None
+    trace_id: str
+    flags: list[str] = Field(default_factory=list)
+    server_picks: list[dict[str, Any]] = Field(default_factory=list)
+    parsed: dict[str, str] | None = None
+
+
+class SettingsStatusData(BaseModel):
+    """Authenticated settings status without secrets."""
+
+    app_name: str
+    app_version: str
+    coach_configured: bool
+    model: str
+    data_ok: bool
+    env: str
+
+
+def ok(data: BaseModel | dict[str, Any] | list[Any]) -> dict[str, Any]:
+    """Build a JSON-serializable success envelope."""
+
+    return jsonable_encoder(ApiEnvelope(ok=True, data=data, error=None))
+
+
+def err(code: ErrorCode, message: str, status_code: int) -> JSONResponse:
+    """Build a consistent failure response."""
+
+    payload = ApiEnvelope(
+        ok=False,
+        data=None,
+        error=ApiError(code=code, message=message),
+    )
+    return JSONResponse(status_code=status_code, content=jsonable_encoder(payload))

app/paste_bundle.py

@@ -0,0 +1,75 @@
+"""Format context-free coach debug paste bundles for external AI review.
+
+Bundles include formulas, evidence, prompts, raw output, and checklist.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+
+def render_paste_bundle(trace: dict[str, Any], *, app_name: str, shrink_k: float, match_alpha: float, min_n: int) -> str:
+    """Render the exact markdown paste bundle from a stored trace."""
+
+    evidence = trace.get("evidence") or {}
+    evidence_block = evidence.get("evidence_block") or json.dumps(evidence, indent=2)
+    picks = json.dumps(trace.get("server_picks") or [], indent=2, ensure_ascii=False)
+    current = json.dumps(trace.get("current") or {}, indent=2, ensure_ascii=False)
+    history = json.dumps(trace.get("history_truncated") or [], indent=2, ensure_ascii=False)
+    parsed = json.dumps(trace.get("parsed"), indent=2, ensure_ascii=False)
+    return f"""# Coach Trace Paste Bundle (context-free)
+app: {app_name} | version: {trace.get('app_version')} | trace_id: {trace.get('trace_id')} | ts: {trace.get('ts')}
+
+## How to help
+You have NO prior context about the user. Use ONLY this bundle.
+Do not ask for biography. Critique prompts, math, backup rules, and free-model fitness.
+Output: (1) findings (2) concrete patch list for backend prompts/rules/math weights.
+
+## Math definitions (server authoritative)
+- pending excluded from denominators
+- p_worked(r) = N(worked,r) / N(r)
+- p_helped(r) = N(worked|partial,r) / N(r)
+- rank(r) = p_helped * n/(n+k) with k={shrink_k}
+- pick(r) = rank * (1 + alpha * match) with alpha={match_alpha}
+- min_n = {min_n}
+- DATA_THIN if n_scored < 10
+
+## Server evidence
+{evidence_block}
+
+## Server picks
+{picks}
+
+## Current situation
+{current}
+
+## Recent history (truncated)
+{history}
+
+## System prompt
+{trace.get('system_prompt') or ''}
+
+## User prompt
+{trace.get('user_prompt') or ''}
+
+## Model raw response
+{trace.get('raw_model_response') or ''}
+
+## Parse / source / flags
+source: {trace.get('source')}
+backup_rule_id: {trace.get('backup_rule_id')}
+flags: {trace.get('flags')}
+parsed: {parsed}
+
+## Final text shown to user
+{trace.get('final_text') or ''}
+
+## Reviewer checklist
+1. Invented numbers?
+2. Ignored SERVER_PICKS?
+3. Format broken (free model)?
+4. Backup better?
+5. Evidence too long/short?
+6. Shrinkage k / alpha tweak?
+"""

app/paths.py

@@ -0,0 +1,44 @@
+"""Define and initialize all durable filesystem paths.
+
+Only this module decides where application data files live.
+"""
+
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class Paths:
+    """Concrete files rooted below the configured data directory."""
+
+    root: Path
+
+    def __init__(self, root: str | Path) -> None:
+        object.__setattr__(self, "root", Path(root))
+
+    @property
+    def config(self) -> Path:
+        return self.root / "config.json"
+
+    @property
+    def entries(self) -> Path:
+        return self.root / "entries.jsonl"
+
+    @property
+    def daily(self) -> Path:
+        return self.root / "daily.jsonl"
+
+    @property
+    def traces(self) -> Path:
+        return self.root / "traces.jsonl"
+
+    @property
+    def coach_brief(self) -> Path:
+        return self.root / "coach_brief.md"
+
+    def ensure(self) -> None:
+        """Create the data directory and initial empty JSONL files."""
+
+        self.root.mkdir(parents=True, exist_ok=True)
+        for path in (self.entries, self.daily, self.traces):
+            path.touch(exist_ok=True)

app/routers/__init__.py

@@ -0,0 +1 @@
+"""FastAPI route modules for the Habit Journal API."""

app/routers/auth.py

@@ -0,0 +1,160 @@
+"""Implement password login and signed-session lifecycle routes.
+
+Login attempts are rate-limited in memory by the direct client address.
+"""
+
+import time
+from collections import defaultdict, deque
+from threading import Lock
+
+from fastapi import APIRouter, Depends, Request, status
+
+from app.deps import require_login
+from app.models import (
+    ApiEnvelope,
+    AuthenticatedData,
+    ChangePasswordRequest,
+    LoginRequest,
+    MeData,
+    StoredConfig,
+    err,
+    ok,
+)
+
+router = APIRouter(prefix="/api/auth", tags=["auth"])
+_attempts: dict[str, deque[float]] = defaultdict(deque)
+_attempts_lock = Lock()
+
+
+def _client_key(request: Request) -> str:
+    return request.client.host if request.client else "unknown"
+
+
+def _rate_limited(request: Request) -> bool:
+    settings = request.app.state.settings
+    now = time.monotonic()
+    cutoff = now - settings.login_rate_window_sec
+    key = _client_key(request)
+    with _attempts_lock:
+        attempts = _attempts[key]
+        while attempts and attempts[0] <= cutoff:
+            attempts.popleft()
+        if len(attempts) >= settings.login_rate_limit:
+            return True
+        attempts.append(now)
+        return False
+
+
+def _clear_attempts(request: Request) -> None:
+    with _attempts_lock:
+        _attempts.pop(_client_key(request), None)
+
+
+def _set_session(request: Request, config: StoredConfig) -> None:
+    max_age = request.app.state.settings.session_max_age_sec
+    request.session.clear()
+    request.session.update(
+        {
+            "auth": True,
+            "sv": config.session_version,
+            "exp": int(time.time()) + max_age,
+        }
+    )
+
+
+def _session_is_current(request: Request, config: StoredConfig | None) -> bool:
+    if config is None:
+        return False
+    session = request.session
+    expiry = session.get("exp")
+    return (
+        session.get("auth") is True
+        and session.get("sv") == config.session_version
+        and isinstance(expiry, (int, float))
+        and expiry >= time.time()
+    )
+
+
+@router.post("/login", response_model=ApiEnvelope)
+def login(request: Request, body: LoginRequest) -> object:
+    """Authenticate and establish a signed cookie session."""
+
+    config = request.app.state.config_store.load()
+    if config is None:
+        return err(
+            "setup_required",
+            "Application setup is required",
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+        )
+    if _rate_limited(request):
+        return err(
+            "rate_limited",
+            "Too many login attempts",
+            status.HTTP_429_TOO_MANY_REQUESTS,
+        )
+    if not request.app.state.config_store.password_matches(body.password):
+        return err(
+            "unauthorized",
+            "Invalid credentials",
+            status.HTTP_401_UNAUTHORIZED,
+        )
+
+    _clear_attempts(request)
+    _set_session(request, config)
+    return ok(AuthenticatedData(authenticated=True))
+
+
+@router.post("/logout", response_model=ApiEnvelope)
+def logout(
+    request: Request,
+    _config: StoredConfig = Depends(require_login),
+) -> dict[str, object]:
+    """Clear the current signed session."""
+
+    request.session.clear()
+    return ok(AuthenticatedData(authenticated=False))
+
+
+@router.get("/me", response_model=ApiEnvelope)
+def me(request: Request) -> dict[str, object]:
+    """Return public app identity and current session state."""
+
+    config = request.app.state.config_store.load()
+    authenticated = _session_is_current(request, config)
+    if not authenticated:
+        request.session.clear()
+    return ok(
+        MeData(
+            authenticated=authenticated,
+            app_name=request.app.state.settings.app_name,
+        )
+    )
+
+
+@router.post("/change-password", response_model=ApiEnvelope)
+def change_password(
+    request: Request,
+    body: ChangePasswordRequest,
+    _config: StoredConfig = Depends(require_login),
+) -> object:
+    """Rotate the shared password and invalidate older sessions."""
+
+    changed = request.app.state.config_store.change_password(
+        body.old_password,
+        body.new_password,
+    )
+    if not changed:
+        return err(
+            "unauthorized",
+            "Invalid credentials",
+            status.HTTP_401_UNAUTHORIZED,
+        )
+    config = request.app.state.config_store.load()
+    if config is None:
+        return err(
+            "internal",
+            "Configuration unavailable",
+            status.HTTP_500_INTERNAL_SERVER_ERROR,
+        )
+    _set_session(request, config)
+    return ok(AuthenticatedData(authenticated=True))

app/routers/coach.py

@@ -0,0 +1,52 @@
+"""Coach endpoints: free-text coach and per-entry coach with optional persist.
+
+Always returns 200 with backup text when the model is unavailable.
+"""
+
+from fastapi import APIRouter, Depends, Request, status
+
+from app.deps import require_login
+from app.models import ApiEnvelope, CoachRequest, CoachResponseData, err, ok
+
+router = APIRouter(prefix="/api", tags=["coach"], dependencies=[Depends(require_login)])
+
+
+def _run_coach(request: Request, body: CoachRequest, *, default_persist: bool) -> object:
+    if not body.text and not body.entry_id:
+        return err(
+            "validation_error",
+            "Provide text or entry_id",
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+        )
+    persist = (
+        body.persist if "persist" in body.model_fields_set else default_persist
+    )
+    try:
+        result = request.app.state.coach_service.coach(
+            text=body.text,
+            entry_id=body.entry_id,
+            include_history=body.include_history,
+            persist=persist,
+            force_backup=body.force_backup,
+        )
+    except KeyError:
+        return err("not_found", "Entry not found", status.HTTP_404_NOT_FOUND)
+    except ValueError as exc:
+        return err("validation_error", str(exc), status.HTTP_422_UNPROCESSABLE_ENTITY)
+    return ok(CoachResponseData.model_validate(result))
+
+
+@router.post("/coach", response_model=ApiEnvelope)
+def coach(request: Request, body: CoachRequest) -> object:
+    """Run coach for free text or an entry; persist defaults to false."""
+
+    return _run_coach(request, body, default_persist=False)
+
+
+@router.post("/entries/{entry_id}/coach", response_model=ApiEnvelope)
+def entry_coach(request: Request, entry_id: str, body: CoachRequest | None = None) -> object:
+    """Run coach for one entry; persist defaults to true."""
+
+    payload = body or CoachRequest()
+    merged = payload.model_copy(update={"entry_id": entry_id})
+    return _run_coach(request, merged, default_persist=True)

app/routers/daily.py

@@ -0,0 +1,63 @@
+"""Expose authenticated daily scoreboard upsert and range queries.
+
+Points and week band are always computed on the server.
+"""
+
+from datetime import date
+
+from fastapi import APIRouter, Depends, Query, Request, status
+
+from app.deps import require_login
+from app.models import ApiEnvelope, DailyRangeData, DailyUpsert, err, ok
+
+router = APIRouter(
+    prefix="/api/daily",
+    tags=["daily"],
+    dependencies=[Depends(require_login)],
+)
+
+
+def _store(request: Request):
+    return request.app.state.daily_store
+
+
+@router.put("/{day}", response_model=ApiEnvelope)
+def upsert_daily(request: Request, day: date, body: DailyUpsert) -> dict[str, object]:
+    """Upsert one daily row and recompute points."""
+
+    return ok(_store(request).upsert(day, body))
+
+
+@router.get("/{day}", response_model=ApiEnvelope)
+def get_daily(request: Request, day: date) -> object:
+    """Fetch one daily row by date."""
+
+    row = _store(request).get(day)
+    if row is None:
+        return err("not_found", "Daily row not found", status.HTTP_404_NOT_FOUND)
+    return ok(row)
+
+
+@router.get("", response_model=ApiEnvelope)
+def list_daily(
+    request: Request,
+    start: date = Query(...),
+    end: date = Query(...),
+) -> object:
+    """List daily rows in a date range with week band summary."""
+
+    if end < start:
+        return err(
+            "validation_error",
+            "end must be on or after start",
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+        )
+    items, week_points, band, days_present = _store(request).range(start, end)
+    return ok(
+        DailyRangeData(
+            items=items,
+            week_points=week_points,
+            band=band,  # type: ignore[arg-type]
+            days_present=days_present,
+        )
+    )

app/routers/debug.py

@@ -0,0 +1,132 @@
+"""Authenticated debug routes for coach traces and prompt previews.
+
+Paste bundles are self-contained for an external AI with no user history.
+"""
+
+from fastapi import APIRouter, Depends, Query, Request, status
+from fastapi.responses import PlainTextResponse
+
+from app.deps import require_login
+from app.models import ApiEnvelope, CoachRequest, err, ok
+from app.paste_bundle import render_paste_bundle
+
+router = APIRouter(
+    prefix="/api/debug",
+    tags=["debug"],
+    dependencies=[Depends(require_login)],
+)
+
+
+def _settings(request: Request):
+    return request.app.state.settings
+
+
+def _paste(request: Request, trace: dict) -> PlainTextResponse:
+    settings = _settings(request)
+    body = render_paste_bundle(
+        trace,
+        app_name=settings.app_name,
+        shrink_k=settings.stats_shrink_k,
+        match_alpha=settings.match_alpha,
+        min_n=settings.min_stats_n,
+    )
+    return PlainTextResponse(content=body, media_type="text/markdown")
+
+
+@router.get("/traces", response_model=ApiEnvelope)
+def list_traces(
+    request: Request,
+    limit: int = Query(default=20, ge=1, le=200),
+) -> dict[str, object]:
+    """Return newest-first trace summaries."""
+
+    records = request.app.state.trace_store.list_traces(limit=limit)
+    summaries = [
+        {
+            "trace_id": r.get("trace_id"),
+            "ts": r.get("ts"),
+            "source": r.get("source"),
+            "flags": r.get("flags"),
+            "model": r.get("model_id"),
+        }
+        for r in records
+    ]
+    return ok({"items": summaries})
+
+
+@router.get("/traces/{trace_id}", response_model=ApiEnvelope)
+def get_trace(request: Request, trace_id: str) -> object:
+    """Return one full JSON trace."""
+
+    trace = request.app.state.trace_store.get_trace(trace_id)
+    if trace is None:
+        return err("not_found", "Trace not found", status.HTTP_404_NOT_FOUND)
+    return ok(trace)
+
+
+@router.get("/traces/{trace_id}/paste")
+def paste_trace(request: Request, trace_id: str) -> object:
+    """Return the context-free markdown paste bundle for one trace."""
+
+    trace = request.app.state.trace_store.get_trace(trace_id)
+    if trace is None:
+        return err("not_found", "Trace not found", status.HTTP_404_NOT_FOUND)
+    return _paste(request, trace)
+
+
+@router.get("/last/paste")
+def paste_last(request: Request) -> object:
+    """Return the newest trace paste bundle."""
+
+    trace = request.app.state.trace_store.latest_trace()
+    if trace is None:
+        return err("not_found", "No traces yet", status.HTTP_404_NOT_FOUND)
+    return _paste(request, trace)
+
+
+@router.post("/prompt-preview", response_model=ApiEnvelope)
+def prompt_preview(request: Request, body: CoachRequest) -> object:
+    """Build evidence and messages without calling the model."""
+
+    if not body.text and not body.entry_id:
+        return err(
+            "validation_error",
+            "Provide text or entry_id",
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+        )
+    try:
+        preview = request.app.state.coach_service.prompt_preview(
+            text=body.text,
+            entry_id=body.entry_id,
+            include_history=body.include_history,
+        )
+    except KeyError:
+        return err("not_found", "Entry not found", status.HTTP_404_NOT_FOUND)
+    except ValueError as exc:
+        return err("validation_error", str(exc), status.HTTP_422_UNPROCESSABLE_ENTITY)
+    return ok(preview)
+
+
+@router.post("/coach-test", response_model=ApiEnvelope)
+def coach_test(request: Request, body: CoachRequest) -> object:
+    """Run coach and force a trace; optional force_backup."""
+
+    if not body.text and not body.entry_id:
+        return err(
+            "validation_error",
+            "Provide text or entry_id",
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+        )
+    try:
+        result = request.app.state.coach_service.coach(
+            text=body.text,
+            entry_id=body.entry_id,
+            include_history=body.include_history,
+            persist=body.persist,
+            force_backup=body.force_backup,
+        )
+    except KeyError:
+        return err("not_found", "Entry not found", status.HTTP_404_NOT_FOUND)
+    except ValueError as exc:
+        return err("validation_error", str(exc), status.HTTP_422_UNPROCESSABLE_ENTITY)
+    return ok(result)

app/routers/entries.py

@@ -0,0 +1,99 @@
+"""Expose authenticated CRUD and filtered listing for log entries.
+
+All routes require a valid session and use the shared entry store.
+"""
+
+from datetime import date
+
+from fastapi import APIRouter, Depends, Query, Request, status
+
+from app.deps import require_login
+from app.models import (
+    ApiEnvelope,
+    DeletedData,
+    EntryCreate,
+    EntryListData,
+    EntryUpdate,
+    Result,
+    err,
+    normalize_emotions,
+    normalize_tags,
+    ok,
+)
+
+router = APIRouter(
+    prefix="/api/entries",
+    tags=["entries"],
+    dependencies=[Depends(require_login)],
+)
+
+
+def _store(request: Request):
+    return request.app.state.entry_store
+
+
+@router.post("", response_model=ApiEnvelope)
+def create_entry(request: Request, body: EntryCreate) -> dict[str, object]:
+    """Create a new log entry."""
+
+    return ok(_store(request).create(body))
+
+
+@router.get("", response_model=ApiEnvelope)
+def list_entries(
+    request: Request,
+    start: date | None = None,
+    end: date | None = None,
+    tag: str | None = None,
+    result: Result | None = None,
+    emotion: str | None = None,
+    q: str | None = None,
+    limit: int = Query(default=50, ge=1, le=200),
+    offset: int = Query(default=0, ge=0),
+) -> dict[str, object]:
+    """List entries filtered by date, tag, result, emotion, or text."""
+
+    tag_norm = normalize_tags([tag])[0] if tag and tag.strip() else None
+    emotion_norm = (
+        normalize_emotions([emotion])[0] if emotion and emotion.strip() else None
+    )
+    items, total = _store(request).list(
+        start=start,
+        end=end,
+        tag=tag_norm,
+        result=result,
+        emotion=emotion_norm,
+        q=q,
+        limit=limit,
+        offset=offset,
+    )
+    return ok(EntryListData(items=items, total=total, limit=limit, offset=offset))
+
+
+@router.get("/{entry_id}", response_model=ApiEnvelope)
+def get_entry(request: Request, entry_id: str) -> object:
+    """Fetch a single entry by id."""
+
+    entry = _store(request).get(entry_id)
+    if entry is None:
+        return err("not_found", "Entry not found", status.HTTP_404_NOT_FOUND)
+    return ok(entry)
+
+
+@router.patch("/{entry_id}", response_model=ApiEnvelope)
+def update_entry(request: Request, entry_id: str, body: EntryUpdate) -> object:
+    """Apply a partial update to an entry."""
+
+    entry = _store(request).update(entry_id, body)
+    if entry is None:
+        return err("not_found", "Entry not found", status.HTTP_404_NOT_FOUND)
+    return ok(entry)
+
+
+@router.delete("/{entry_id}", response_model=ApiEnvelope)
+def delete_entry(request: Request, entry_id: str) -> object:
+    """Hard-delete an entry by id."""
+
+    if not _store(request).delete(entry_id):
+        return err("not_found", "Entry not found", status.HTTP_404_NOT_FOUND)
+    return ok(DeletedData(deleted=True, id=entry_id))

app/routers/export.py

@@ -0,0 +1,75 @@
+"""Authenticated export of entries and daily rows as JSONL or CSV.
+
+Exports stream whole files from DATA_ROOT without mounting that directory.
+"""
+
+import csv
+import io
+from typing import Any
+
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import PlainTextResponse, Response
+
+from app.deps import require_login
+from app.fsutil import read_jsonl
+
+router = APIRouter(
+    prefix="/api/export",
+    tags=["export"],
+    dependencies=[Depends(require_login)],
+)
+
+CSV_COLUMNS = [
+    "id",
+    "ts",
+    "created_at",
+    "updated_at",
+    "activity",
+    "happened",
+    "emotions",
+    "intensity",
+    "remedy",
+    "result",
+    "tags",
+    "notes",
+]
+
+
+def _jsonl_response(path) -> Response:
+    if not path.exists():
+        content = ""
+    else:
+        content = path.read_text(encoding="utf-8")
+    return Response(content=content, media_type="application/x-ndjson")
+
+
+@router.get("/entries.jsonl")
+def export_entries_jsonl(request: Request) -> Response:
+    """Download entries as newline-delimited JSON."""
+
+    return _jsonl_response(request.app.state.paths.entries)
+
+
+@router.get("/daily.jsonl")
+def export_daily_jsonl(request: Request) -> Response:
+    """Download daily rows as newline-delimited JSON."""
+
+    return _jsonl_response(request.app.state.paths.daily)
+
+
+@router.get("/entries.csv")
+def export_entries_csv(request: Request) -> PlainTextResponse:
+    """Download entries as CSV with fixed headers."""
+
+    records: list[dict[str, Any]] = read_jsonl(request.app.state.paths.entries)
+    buffer = io.StringIO()
+    writer = csv.DictWriter(buffer, fieldnames=CSV_COLUMNS, extrasaction="ignore")
+    writer.writeheader()
+    for record in records:
+        row = {key: record.get(key, "") for key in CSV_COLUMNS}
+        emotions = record.get("emotions") or []
+        tags = record.get("tags") or []
+        row["emotions"] = "|".join(str(item) for item in emotions)
+        row["tags"] = "|".join(str(item) for item in tags)
+        writer.writerow(row)
+    return PlainTextResponse(content=buffer.getvalue(), media_type="text/csv")

app/routers/health.py

@@ -0,0 +1,45 @@
+"""Expose unauthenticated liveness and bootstrap-readiness endpoints.
+
+Readiness verifies both durable configuration and a writable data directory.
+"""
+
+import tempfile
+from datetime import datetime, timezone
+from pathlib import Path
+
+from fastapi import APIRouter, Request
+
+from app.models import ApiEnvelope, HealthData, ReadyData, ok
+
+router = APIRouter(prefix="/api", tags=["health"])
+
+
+def _is_writable(root: Path) -> bool:
+    try:
+        with tempfile.NamedTemporaryFile(dir=root, prefix=".ready-", delete=True):
+            return True
+    except OSError:
+        return False
+
+
+@router.get("/health", response_model=ApiEnvelope)
+def health() -> dict[str, object]:
+    """Return process liveness without requiring setup or login."""
+
+    return ok(HealthData(status="up", time=datetime.now(timezone.utc)))
+
+
+@router.get("/ready", response_model=ApiEnvelope)
+def ready(request: Request) -> dict[str, object]:
+    """Return whether storage is writable and bootstrap is complete."""
+
+    root = request.app.state.paths.root
+    writable = root.is_dir() and _is_writable(root)
+    config_present = request.app.state.config_store.load() is not None
+    return ok(
+        ReadyData(
+            ready=writable and config_present,
+            data_writable=writable,
+            config_present=config_present,
+        )
+    )

app/routers/settings.py

@@ -0,0 +1,35 @@
+"""Authenticated settings status without exposing secrets.
+
+Reports coach configuration state and durable data readiness.
+"""
+
+from fastapi import APIRouter, Depends, Request
+
+from app.config import APP_VERSION
+from app.deps import require_login
+from app.models import ApiEnvelope, SettingsStatusData, ok
+
+router = APIRouter(
+    prefix="/api/settings",
+    tags=["settings"],
+    dependencies=[Depends(require_login)],
+)
+
+
+@router.get("/status", response_model=ApiEnvelope)
+def settings_status(request: Request) -> dict[str, object]:
+    """Return non-secret application and coach status."""
+
+    settings = request.app.state.settings
+    paths = request.app.state.paths
+    data_ok = paths.root.is_dir() and paths.entries.exists() and paths.daily.exists()
+    return ok(
+        SettingsStatusData(
+            app_name=settings.app_name,
+            app_version=APP_VERSION,
+            coach_configured=bool(settings.openrouter_api_key),
+            model=settings.openrouter_model,
+            data_ok=data_ok,
+            env=settings.env,
+        )
+    )

app/routers/stats.py

@@ -0,0 +1,103 @@
+"""Expose authenticated server-side statistics endpoints.
+
+Probabilities and ranks are computed in Python, never by the model.
+"""
+
+from datetime import date, datetime, timezone
+
+from fastapi import APIRouter, Depends, Query, Request
+
+from app.deps import require_login
+from app.models import ApiEnvelope, ok
+from app.stats_math import build_stats, by_remedy
+
+router = APIRouter(
+    prefix="/api/stats",
+    tags=["stats"],
+    dependencies=[Depends(require_login)],
+)
+
+
+def _entry_dicts(request: Request) -> list[dict]:
+    return [e.model_dump(mode="json") for e in request.app.state.entry_store._load()]
+
+
+def _daily_dicts(
+    request: Request,
+    start: date | None,
+    end: date | None,
+) -> list[dict]:
+    rows = request.app.state.daily_store._load()
+    out = []
+    for row in rows:
+        if start and row.date < start:
+            continue
+        if end and row.date > end:
+            continue
+        out.append(row.model_dump(mode="json"))
+    return out
+
+
+def _filter_entries(
+    entries: list[dict],
+    start: date | None,
+    end: date | None,
+) -> list[dict]:
+    if start is None and end is None:
+        return entries
+    filtered = []
+    for entry in entries:
+        ts = entry.get("ts")
+        if isinstance(ts, str):
+            day = date.fromisoformat(ts[:10])
+        else:
+            continue
+        if start and day < start:
+            continue
+        if end and day > end:
+            continue
+        filtered.append(entry)
+    return filtered
+
+
+@router.get("", response_model=ApiEnvelope)
+def get_stats(
+    request: Request,
+    start: date | None = None,
+    end: date | None = None,
+    min_n: int | None = Query(default=None, ge=1),
+) -> dict[str, object]:
+    """Return aggregate statistics for entries and daily rows."""
+
+    settings = request.app.state.settings
+    effective_min_n = settings.min_stats_n if min_n is None else min_n
+    entries = _filter_entries(_entry_dicts(request), start, end)
+    daily = _daily_dicts(request, start, end)
+    payload = build_stats(
+        entries,
+        daily,
+        min_n=effective_min_n,
+        shrink_k=settings.stats_shrink_k,
+        generated_at=datetime.now(timezone.utc).isoformat(),
+    )
+    return ok(payload)
+
+
+@router.get("/remedies", response_model=ApiEnvelope)
+def get_remedy_leaderboard(
+    request: Request,
+    start: date | None = None,
+    end: date | None = None,
+    min_n: int | None = Query(default=None, ge=1),
+) -> dict[str, object]:
+    """Return remedies sorted by shrinkage rank."""
+
+    settings = request.app.state.settings
+    effective_min_n = settings.min_stats_n if min_n is None else min_n
+    entries = _filter_entries(_entry_dicts(request), start, end)
+    rows = by_remedy(
+        entries,
+        min_n=effective_min_n,
+        shrink_k=settings.stats_shrink_k,
+    )
+    return ok({"items": rows, "min_n": effective_min_n})

app/security_passwords.py

@@ -0,0 +1,52 @@
+"""Hash and verify the single application password.
+
+PBKDF2, random salts, and constant-time comparison are provided by stdlib.
+"""
+
+import base64
+import hashlib
+import hmac
+import os
+
+ALGORITHM = "pbkdf2_sha256"
+DEFAULT_ITERATIONS = 260_000
+MIN_ITERATIONS = 200_000
+
+
+def hash_password(password: str, iterations: int = DEFAULT_ITERATIONS) -> str:
+    """Return a salted PBKDF2 password hash in the configured wire format."""
+
+    if iterations < MIN_ITERATIONS:
+        raise ValueError(f"iterations must be at least {MIN_ITERATIONS}")
+    salt = os.urandom(16)
+    digest = hashlib.pbkdf2_hmac(
+        "sha256",
+        password.encode("utf-8"),
+        salt,
+        iterations,
+    )
+    salt_b64 = base64.b64encode(salt).decode("ascii")
+    digest_b64 = base64.b64encode(digest).decode("ascii")
+    return f"{ALGORITHM}${iterations}${salt_b64}${digest_b64}"
+
+
+def verify_password(password: str, encoded: str) -> bool:
+    """Verify a password without leaking comparison timing."""
+
+    try:
+        algorithm, raw_iterations, salt_b64, expected_b64 = encoded.split("$", 3)
+        iterations = int(raw_iterations)
+        if algorithm != ALGORITHM or iterations < MIN_ITERATIONS:
+            return False
+        salt = base64.b64decode(salt_b64, validate=True)
+        expected = base64.b64decode(expected_b64, validate=True)
+    except (ValueError, TypeError):
+        return False
+
+    actual = hashlib.pbkdf2_hmac(
+        "sha256",
+        password.encode("utf-8"),
+        salt,
+        iterations,
+    )
+    return hmac.compare_digest(actual, expected)

app/stats_math.py

@@ -0,0 +1,301 @@
+"""Pure probability and ranking math for log and daily statistics.
+
+No I/O: callers pass entry/daily dicts; this module only aggregates.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from typing import Any
+
+
+def _normalize_remedy(value: str) -> str:
+    return " ".join(value.strip().lower().split())
+
+
+def intensity_bucket(intensity: int) -> str:
+    """Map intensity 1-10 into coarse buckets."""
+
+    if intensity <= 3:
+        return "1-3"
+    if intensity <= 6:
+        return "4-6"
+    return "7-10"
+
+
+def scored_entries(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Exclude pending outcomes from probability denominators."""
+
+    return [e for e in entries if e.get("result") != "pending"]
+
+
+def outcome_histogram(entries: list[dict[str, Any]]) -> dict[str, int]:
+    """Count outcomes including pending."""
+
+    counts = {"worked": 0, "partial": 0, "failed": 0, "pending": 0}
+    for entry in entries:
+        result = entry.get("result")
+        if result in counts:
+            counts[result] += 1
+    return counts
+
+
+def by_remedy(
+    entries: list[dict[str, Any]],
+    *,
+    min_n: int,
+    shrink_k: float,
+) -> list[dict[str, Any]]:
+    """Compute per-remedy n, p_worked, p_helped, and shrinkage rank."""
+
+    scored = scored_entries(entries)
+    totals: dict[str, int] = defaultdict(int)
+    worked: dict[str, int] = defaultdict(int)
+    helped: dict[str, int] = defaultdict(int)
+    for entry in scored:
+        key = _normalize_remedy(str(entry.get("remedy") or ""))
+        if not key:
+            continue
+        totals[key] += 1
+        result = entry.get("result")
+        if result == "worked":
+            worked[key] += 1
+            helped[key] += 1
+        elif result == "partial":
+            helped[key] += 1
+
+    rows: list[dict[str, Any]] = []
+    for key, n in totals.items():
+        if n < min_n:
+            continue
+        p_worked = worked[key] / n
+        p_helped = helped[key] / n
+        rank = p_helped * (n / (n + shrink_k))
+        rows.append(
+            {
+                "key": key,
+                "n": n,
+                "p_worked": p_worked,
+                "p_helped": p_helped,
+                "rank": rank,
+            }
+        )
+    rows.sort(key=lambda row: (-row["rank"], -row["n"], row["key"]))
+    return rows
+
+
+def by_emotion(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Emotion multi-label buckets with p_helped among scored uses."""
+
+    totals: dict[str, int] = defaultdict(int)
+    helped: dict[str, int] = defaultdict(int)
+    for entry in scored_entries(entries):
+        result = entry.get("result")
+        is_helped = result in ("worked", "partial")
+        for emotion in entry.get("emotions") or []:
+            key = str(emotion).strip().lower()
+            if not key:
+                continue
+            totals[key] += 1
+            if is_helped:
+                helped[key] += 1
+    rows = [
+        {
+            "key": key,
+            "n": n,
+            "p_helped": (helped[key] / n) if n else 0.0,
+        }
+        for key, n in totals.items()
+    ]
+    rows.sort(key=lambda row: (-row["n"], row["key"]))
+    return rows
+
+
+def by_tag(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Tag buckets with failure rate among scored uses."""
+
+    totals: dict[str, int] = defaultdict(int)
+    failed: dict[str, int] = defaultdict(int)
+    for entry in scored_entries(entries):
+        is_failed = entry.get("result") == "failed"
+        for tag in entry.get("tags") or []:
+            key = str(tag).strip().lower()
+            if not key:
+                continue
+            totals[key] += 1
+            if is_failed:
+                failed[key] += 1
+    rows = [
+        {
+            "key": key,
+            "n": n,
+            "p_failed": (failed[key] / n) if n else 0.0,
+        }
+        for key, n in totals.items()
+    ]
+    rows.sort(key=lambda row: (-row["p_failed"], -row["n"], row["key"]))
+    return rows
+
+
+def by_intensity_bucket(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Intensity bucket counts and helped rates."""
+
+    totals = {"1-3": 0, "4-6": 0, "7-10": 0}
+    helped = {"1-3": 0, "4-6": 0, "7-10": 0}
+    for entry in scored_entries(entries):
+        try:
+            intensity = int(entry.get("intensity", 0))
+        except (TypeError, ValueError):
+            continue
+        if intensity < 1 or intensity > 10:
+            continue
+        bucket = intensity_bucket(intensity)
+        totals[bucket] += 1
+        if entry.get("result") in ("worked", "partial"):
+            helped[bucket] += 1
+    return [
+        {
+            "key": key,
+            "n": totals[key],
+            "p_helped": (helped[key] / totals[key]) if totals[key] else 0.0,
+        }
+        for key in ("1-3", "4-6", "7-10")
+    ]
+
+
+def corn_ok(row: dict[str, Any]) -> bool:
+    """True when corn sessions are within the delay policy."""
+
+    sessions = int(row.get("corn_sessions") or 0)
+    delay_ok = bool(row.get("delay_ok", True))
+    return sessions == 0 or (sessions <= 1 and delay_ok)
+
+
+def daily_rates(daily_rows: list[dict[str, Any]]) -> dict[str, float]:
+    """Aggregate daily scoreboard rates for a set of days."""
+
+    n = len(daily_rows)
+    if n == 0:
+        return {
+            "p_brick_done": 0.0,
+            "p_corn_ok": 0.0,
+            "p_no_fc": 0.0,
+            "p_rerun_clean": 0.0,
+            "p_court_closed": 0.0,
+            "avg_points": 0.0,
+        }
+    brick = sum(1 for row in daily_rows if row.get("brick_done"))
+    corn = sum(1 for row in daily_rows if corn_ok(row))
+    no_fc = sum(1 for row in daily_rows if row.get("daydream") != "fc")
+    rerun = sum(1 for row in daily_rows if row.get("rerun") == "clean")
+    court = sum(1 for row in daily_rows if row.get("court") == "closed")
+    avg_points = sum(float(row.get("points") or 0) for row in daily_rows) / n
+    return {
+        "p_brick_done": brick / n,
+        "p_corn_ok": corn / n,
+        "p_no_fc": no_fc / n,
+        "p_rerun_clean": rerun / n,
+        "p_court_closed": court / n,
+        "avg_points": avg_points,
+    }
+
+
+def helped_tags_by_remedy(entries: list[dict[str, Any]]) -> dict[str, set[str]]:
+    """Tags that appear on helped uses of each remedy."""
+
+    tags_by_remedy: dict[str, set[str]] = defaultdict(set)
+    for entry in scored_entries(entries):
+        if entry.get("result") not in ("worked", "partial"):
+            continue
+        key = _normalize_remedy(str(entry.get("remedy") or ""))
+        if not key:
+            continue
+        for tag in entry.get("tags") or []:
+            token = str(tag).strip().lower()
+            if token:
+                tags_by_remedy[key].add(token)
+    return tags_by_remedy
+
+
+def match_score(current_tags: set[str], remedy_tags: set[str]) -> float:
+    """Fraction of current tags that match a remedy's helped-tag set."""
+
+    if not current_tags:
+        return 0.0
+    return len(current_tags & remedy_tags) / max(len(current_tags), 1)
+
+
+def server_picks(
+    entries: list[dict[str, Any]],
+    current_tags: list[str] | set[str],
+    *,
+    min_n: int,
+    shrink_k: float,
+    match_alpha: float,
+    limit: int = 5,
+) -> list[dict[str, Any]]:
+    """Rank remedies by shrinkage + optional tag-match boost."""
+
+    tag_set = {str(tag).strip().lower() for tag in current_tags if str(tag).strip()}
+    remedy_tags = helped_tags_by_remedy(entries)
+    picks: list[dict[str, Any]] = []
+    for row in by_remedy(entries, min_n=min_n, shrink_k=shrink_k):
+        match = match_score(tag_set, remedy_tags.get(row["key"], set()))
+        pick = row["rank"] * (1.0 + match_alpha * match)
+        picks.append(
+            {
+                "remedy_key": row["key"],
+                "pick": pick,
+                "n": row["n"],
+                "p_helped": row["p_helped"],
+                "p_worked": row["p_worked"],
+                "rank": row["rank"],
+                "match": match,
+            }
+        )
+    picks.sort(key=lambda item: (-item["pick"], -item["n"], item["remedy_key"]))
+    return picks[:limit]
+
+
+def data_thin(n_scored: int) -> bool:
+    """True when scored history is too thin for strong coaching."""
+
+    return n_scored < 10
+
+
+FORMULAS = {
+    "p_worked": "N(worked,r) / N(r); pending excluded",
+    "p_helped": "N(worked|partial,r) / N(r); pending excluded",
+    "rank": "p_helped * n/(n+k)",
+    "pick": "rank * (1 + alpha * match)",
+    "match": "|T intersect T_r| / max(|T|,1)",
+    "DATA_THIN": "n_scored < 10",
+}
+
+
+def build_stats(
+    entries: list[dict[str, Any]],
+    daily_rows: list[dict[str, Any]],
+    *,
+    min_n: int,
+    shrink_k: float,
+    generated_at: str,
+) -> dict[str, Any]:
+    """Assemble the /api/stats response payload."""
+
+    scored = scored_entries(entries)
+    return {
+        "n_entries_total": len(entries),
+        "n_entries_scored": len(scored),
+        "outcomes": outcome_histogram(entries),
+        "by_remedy": by_remedy(entries, min_n=min_n, shrink_k=shrink_k),
+        "by_emotion": by_emotion(entries),
+        "by_tag": by_tag(entries),
+        "by_intensity_bucket": by_intensity_bucket(entries),
+        "daily": daily_rates(daily_rows),
+        "formulas": FORMULAS,
+        "generated_at": generated_at,
+        "min_n": min_n,
+        "shrink_k": shrink_k,
+        "DATA_THIN": data_thin(len(scored)),
+    }

app/store_config.py

@@ -0,0 +1,84 @@
+"""Persist bootstrap credentials and session-version state.
+
+Configuration updates are serialized and atomically replace `config.json`.
+"""
+
+import json
+from datetime import datetime, timezone
+
+from app.fsutil import atomic_write_text, file_lock, read_json
+from app.models import StoredConfig
+from app.paths import Paths
+from app.security_passwords import hash_password, verify_password
+
+
+def utc_now() -> datetime:
+    """Return an aware UTC timestamp."""
+
+    return datetime.now(timezone.utc)
+
+
+class ConfigStore:
+    """Read and mutate the durable single-user configuration."""
+
+    def __init__(self, paths: Paths) -> None:
+        self.paths = paths
+
+    def bootstrap(self, initial_password: str | None) -> StoredConfig | None:
+        """Create initial configuration only when a password is available."""
+
+        with file_lock(self.paths.config):
+            existing = read_json(self.paths.config)
+            if existing is not None:
+                return StoredConfig.model_validate(existing)
+            if not initial_password:
+                return None
+            now = utc_now()
+            config = StoredConfig(
+                password_hash=hash_password(initial_password),
+                session_version=1,
+                created_at=now,
+                updated_at=now,
+            )
+            self._write_unlocked(config)
+            return config
+
+    def load(self) -> StoredConfig | None:
+        """Load configuration, or None when setup has not occurred."""
+
+        value = read_json(self.paths.config)
+        return StoredConfig.model_validate(value) if value is not None else None
+
+    def password_matches(self, password: str) -> bool:
+        """Check a candidate password against current configuration."""
+
+        config = self.load()
+        return bool(config and verify_password(password, config.password_hash))
+
+    def change_password(self, old_password: str, new_password: str) -> bool:
+        """Rotate the password and invalidate all prior sessions."""
+
+        with file_lock(self.paths.config):
+            value = read_json(self.paths.config)
+            if value is None:
+                return False
+            config = StoredConfig.model_validate(value)
+            if not verify_password(old_password, config.password_hash):
+                return False
+            updated = config.model_copy(
+                update={
+                    "password_hash": hash_password(new_password),
+                    "session_version": config.session_version + 1,
+                    "updated_at": utc_now(),
+                }
+            )
+            self._write_unlocked(updated)
+            return True
+
+    def _write_unlocked(self, config: StoredConfig) -> None:
+        payload = json.dumps(
+            config.model_dump(mode="json"),
+            ensure_ascii=False,
+            indent=2,
+        )
+        atomic_write_text(self.paths.config, payload + "\n")

app/store_daily.py

@@ -0,0 +1,120 @@
+"""Persist daily scoreboard rows and compute server-authoritative points.
+
+Rows are unique by date; upserts rewrite the JSONL file under lock.
+"""
+
+from datetime import date, datetime, timezone
+
+from app.fsutil import read_jsonl, rewrite_jsonl
+from app.models import Court, DailyRow, DailyUpsert, Daydream, Rerun
+from app.paths import Paths
+
+
+def compute_daily_points(
+    *,
+    brick_done: bool,
+    corn_sessions: int,
+    delay_ok: bool,
+    daydream: Daydream | str,
+    rerun: Rerun | str,
+    court: Court | str,
+) -> int:
+    """Compute daily points from the authoritative formula (max 6)."""
+
+    points = 0
+    if brick_done:
+        points += 2
+    if corn_sessions == 0 or (corn_sessions <= 1 and delay_ok):
+        points += 1
+    daydream_value = daydream.value if isinstance(daydream, Daydream) else daydream
+    if daydream_value != Daydream.fc.value:
+        points += 1
+    rerun_value = rerun.value if isinstance(rerun, Rerun) else rerun
+    if rerun_value == Rerun.clean.value:
+        points += 1
+    court_value = court.value if isinstance(court, Court) else court
+    if court_value == Court.closed.value:
+        points += 1
+    return points
+
+
+def week_band(days_present: int, week_points: int) -> str:
+    """Classify a 7-day window from present days and point sum."""
+
+    if days_present < 7:
+        return "incomplete"
+    if week_points >= 28:
+        return "strong"
+    if week_points >= 18:
+        return "mixed"
+    return "escape_heavy"
+
+
+def _utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class DailyStore:
+    """Read and mutate the durable daily JSONL file."""
+
+    def __init__(self, paths: Paths) -> None:
+        self.paths = paths
+
+    def _load(self) -> list[DailyRow]:
+        return [DailyRow.model_validate(record) for record in read_jsonl(self.paths.daily)]
+
+    def get(self, day: date) -> DailyRow | None:
+        """Return the row for one calendar date, or None."""
+
+        for row in self._load():
+            if row.date == day:
+                return row
+        return None
+
+    def upsert(self, day: date, data: DailyUpsert) -> DailyRow:
+        """Create or replace a daily row and recompute points."""
+
+        points = compute_daily_points(
+            brick_done=data.brick_done,
+            corn_sessions=data.corn_sessions,
+            delay_ok=data.delay_ok,
+            daydream=data.daydream,
+            rerun=data.rerun,
+            court=data.court,
+        )
+        row = DailyRow(
+            date=day,
+            primary_brick=data.primary_brick,
+            brick_done=data.brick_done,
+            corn_sessions=data.corn_sessions,
+            delay_ok=data.delay_ok,
+            daydream=data.daydream,
+            rerun=data.rerun,
+            court=data.court,
+            points=points,
+            note=data.note,
+            updated_at=_utc_now(),
+        )
+        records = read_jsonl(self.paths.daily)
+        out: list[dict] = []
+        replaced = False
+        for record in records:
+            if record.get("date") == day.isoformat():
+                out.append(row.model_dump(mode="json"))
+                replaced = True
+            else:
+                out.append(record)
+        if not replaced:
+            out.append(row.model_dump(mode="json"))
+        rewrite_jsonl(self.paths.daily, out)
+        return row
+
+    def range(self, start: date, end: date) -> tuple[list[DailyRow], int, str, int]:
+        """Return rows in [start, end], week_points, band, and days_present."""
+
+        items = [row for row in self._load() if start <= row.date <= end]
+        items.sort(key=lambda row: row.date)
+        days_present = len(items)
+        week_points = sum(row.points for row in items)
+        band = week_band(days_present, week_points)
+        return items, week_points, band, days_present

app/store_entries.py

@@ -0,0 +1,146 @@
+"""Persist log entries as append-only JSONL with locked full rewrites.
+
+Creation appends one line; updates and deletes rewrite the file atomically.
+"""
+
+from datetime import date, datetime, timezone
+from uuid import uuid4
+
+from app.fsutil import append_jsonl, read_jsonl, rewrite_jsonl
+from app.models import CoachMeta, Entry, EntryCreate, EntryUpdate, Result
+from app.paths import Paths
+
+
+def _utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class EntryStore:
+    """Read and mutate the durable entries JSONL file."""
+
+    def __init__(self, paths: Paths) -> None:
+        self.paths = paths
+
+    def create(self, data: EntryCreate) -> Entry:
+        """Append a new entry and return the stored record."""
+
+        now = _utc_now()
+        entry = Entry(
+            id=str(uuid4()),
+            ts=data.ts or now,
+            created_at=now,
+            updated_at=now,
+            activity=data.activity,
+            happened=data.happened,
+            emotions=data.emotions,
+            intensity=data.intensity,
+            remedy=data.remedy,
+            result=data.result,
+            tags=data.tags,
+            notes=data.notes,
+            coach=CoachMeta(),
+        )
+        append_jsonl(self.paths.entries, entry.model_dump(mode="json"))
+        return entry
+
+    def _load(self) -> list[Entry]:
+        return [Entry.model_validate(record) for record in read_jsonl(self.paths.entries)]
+
+    def get(self, entry_id: str) -> Entry | None:
+        """Return a single entry by id, or None when absent."""
+
+        for entry in self._load():
+            if entry.id == entry_id:
+                return entry
+        return None
+
+    def list(
+        self,
+        *,
+        start: date | None = None,
+        end: date | None = None,
+        tag: str | None = None,
+        result: Result | None = None,
+        emotion: str | None = None,
+        q: str | None = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> tuple[list[Entry], int]:
+        """Return newest-first entries matching filters plus the total count."""
+
+        needle = q.lower() if q else None
+        matches: list[Entry] = []
+        for entry in self._load():
+            if start and entry.ts.date() < start:
+                continue
+            if end and entry.ts.date() > end:
+                continue
+            if tag and tag not in entry.tags:
+                continue
+            if result and entry.result != result:
+                continue
+            if emotion and emotion not in entry.emotions:
+                continue
+            if needle:
+                haystack = " ".join(
+                    [entry.activity, entry.happened, entry.remedy, entry.notes]
+                ).lower()
+                if needle not in haystack:
+                    continue
+            matches.append(entry)
+
+        matches.sort(key=lambda item: item.ts, reverse=True)
+        total = len(matches)
+        page = matches[offset : offset + limit]
+        return page, total
+
+    def update(self, entry_id: str, patch: EntryUpdate) -> Entry | None:
+        """Apply a partial update and rewrite the file, or return None."""
+
+        changes = patch.model_dump(exclude_unset=True, mode="json")
+        records = read_jsonl(self.paths.entries)
+        updated: Entry | None = None
+        out: list[dict] = []
+        for record in records:
+            if updated is None and record.get("id") == entry_id:
+                merged = {**record, **changes, "updated_at": _utc_now().isoformat()}
+                updated = Entry.model_validate(merged)
+                out.append(updated.model_dump(mode="json"))
+            else:
+                out.append(record)
+        if updated is None:
+            return None
+        rewrite_jsonl(self.paths.entries, out)
+        return updated
+
+    def delete(self, entry_id: str) -> bool:
+        """Hard-delete an entry, returning whether it existed."""
+
+        records = read_jsonl(self.paths.entries)
+        out = [record for record in records if record.get("id") != entry_id]
+        if len(out) == len(records):
+            return False
+        rewrite_jsonl(self.paths.entries, out)
+        return True
+
+    def set_coach(self, entry_id: str, coach: CoachMeta) -> Entry | None:
+        """Persist coach metadata on an entry without changing other fields."""
+
+        records = read_jsonl(self.paths.entries)
+        updated: Entry | None = None
+        out: list[dict] = []
+        for record in records:
+            if updated is None and record.get("id") == entry_id:
+                merged = {
+                    **record,
+                    "coach": coach.model_dump(mode="json"),
+                    "updated_at": _utc_now().isoformat(),
+                }
+                updated = Entry.model_validate(merged)
+                out.append(updated.model_dump(mode="json"))
+            else:
+                out.append(record)
+        if updated is None:
+            return None
+        rewrite_jsonl(self.paths.entries, out)
+        return updated

app/store_traces.py

@@ -0,0 +1,96 @@
+"""Load coach brief and persist coach debug traces as a ring buffer.
+
+Traces are append-only, then trimmed to the newest configured limit.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+from typing import Any
+
+from app.fsutil import append_jsonl, read_jsonl, rewrite_jsonl
+from app.paths import Paths
+
+DEFAULT_BRIEF = """# Coach brief (generic)
+
+Role: brief checklist coach. No pity. No pep talk.
+Never invent statistics. Use only EVIDENCE and SERVER_PICKS.
+Never cancel committed admin/travel process merely from fear.
+If self-harm language appears: crisis redirect only.
+
+## Bricks
+- A: admin/critical checklist item
+- B: environment/hygiene one act
+- C: build/earn one ship
+- D: logistics one line
+- E: boundary one line then silence
+- S: stop spiral — water/shower/sleep
+
+Court closed by default. Daydream requires same-day brick.
+Urge: delay then brick; ceiling one session/day (policy text only).
+"""
+
+
+class TraceStore:
+    """Read coach brief and manage the traces JSONL ring buffer."""
+
+    def __init__(self, paths: Paths, *, trace_limit: int = 200) -> None:
+        self.paths = paths
+        self.trace_limit = trace_limit
+
+    def ensure_brief(self) -> None:
+        """Seed a generic brief when missing."""
+
+        if not self.paths.coach_brief.exists():
+            self.paths.coach_brief.write_text(DEFAULT_BRIEF, encoding="utf-8")
+
+    def read_brief(self, *, max_bytes: int = 32_768) -> str:
+        """Read brief truncated to max_bytes with a truncation marker."""
+
+        self.ensure_brief()
+        raw = self.paths.coach_brief.read_bytes()
+        if len(raw) > max_bytes:
+            text = raw[:max_bytes].decode("utf-8", errors="replace")
+            return text + "\n[TRUNCATED]\n"
+        return raw.decode("utf-8", errors="replace")
+
+    def brief_meta(self, *, include_full: bool) -> dict[str, Any]:
+        """Return brief hash, excerpt, and optional full body."""
+
+        brief = self.read_brief()
+        digest = hashlib.sha256(brief.encode("utf-8")).hexdigest()
+        return {
+            "brief_sha256": digest,
+            "brief_excerpt": brief[:500],
+            "brief_full": brief if include_full else None,
+        }
+
+    def append_trace(self, trace: dict[str, Any]) -> None:
+        """Append one trace and trim to the newest N records."""
+
+        append_jsonl(self.paths.traces, trace)
+        records = read_jsonl(self.paths.traces)
+        if len(records) > self.trace_limit:
+            rewrite_jsonl(self.paths.traces, records[-self.trace_limit :])
+
+    def list_traces(self, limit: int = 20) -> list[dict[str, Any]]:
+        """Return newest-first trace summaries."""
+
+        records = read_jsonl(self.paths.traces)
+        records.reverse()
+        return records[:limit]
+
+    def get_trace(self, trace_id: str) -> dict[str, Any] | None:
+        """Return one full trace by id."""
+
+        for record in reversed(read_jsonl(self.paths.traces)):
+            if record.get("trace_id") == trace_id:
+                return record
+        return None
+
+    def latest_trace(self) -> dict[str, Any] | None:
+        """Return the newest trace, or None."""
+
+        records = read_jsonl(self.paths.traces)
+        return records[-1] if records else None

docs/BACKEND.md

@@ -0,0 +1,268 @@
+# Backend Design Document
+## Habit Journal (Discrete PWA) — Hugging Face Spaces
+
+**Version:** 2.0.0  
+**Scope:** Backend API, storage, auth, server-side statistics, coach, traces,
+debugging, export, and operations. The frontend is separate.
+
+This is the repository specification. If implementation and this document
+conflict, this document wins until a human changes it.
+
+## Implementation rules
+
+- Work in the vertical slices listed below and keep each diff small.
+- Prefer stdlib, FastAPI, httpx, Pydantic, and pydantic-settings.
+- Use no SQL database, Redis, Celery, DI framework, or repository hierarchy.
+- Every module has a short purpose docstring and public functions are typed.
+- Request and response bodies use Pydantic models.
+- Secrets come only from environment variables. Never log passwords or keys.
+- Keep generic Habit Journal naming and private biography out of git.
+- Disable OpenAPI, Swagger, and ReDoc when `ENV=prod`.
+- After each slice, the app must import and start with uvicorn.
+- Target at most about 15 Python files under `app/`; merge stores if needed.
+
+## Runtime contract
+
+On a fresh Hugging Face Docker Space with `APP_PASSWORD` and
+`APP_SECRET_KEY`, startup creates `/data/loop_logger`, login works, entries and
+daily rows survive restart, statistics are computed in Python, coach always
+returns model or backup text, and every debug paste bundle is understandable
+without prior user context.
+
+The Docker image uses Python 3.11 slim and runs:
+
+```text
+uvicorn app.main:app --host 0.0.0.0 --port ${PORT:-7860}
+```
+
+Required and optional environment:
+
+- `APP_PASSWORD`: bootstrap password when `config.json` does not exist
+- `APP_SECRET_KEY`: required session-signing secret
+- `DATA_ROOT=/data/loop_logger`
+- `ENV=prod`; `APP_NAME=Habit Journal`
+- `OPENROUTER_API_KEY`; `OPENROUTER_MODEL=openrouter/free`
+- `OPENROUTER_BASE_URL=https://openrouter.ai/api/v1`
+- `COACH_TIMEOUT_SEC=25`; `COACH_TEMPERATURE=0.3`
+- `COACH_MAX_TOKENS=400`; `COACH_HISTORY_K=10`
+- `COACH_TRACE_LIMIT=200`
+- `MIN_STATS_N=5`; `STATS_SHRINK_K=3`; `MATCH_ALPHA=0.5`
+- `SESSION_MAX_AGE_SEC=604800`
+- `DEBUG_INCLUDE_FULL_BRIEF=false`
+- `LOGIN_RATE_LIMIT=10`; `LOGIN_RATE_WINDOW_SEC=900`
+
+## Flat architecture and durable files
+
+Routers call services, services call stores, and stores access the filesystem.
+Do not introduce circular imports. `main.py` owns wiring.
+
+`DATA_ROOT` contains `config.json`, `entries.jsonl`, `daily.jsonl`,
+`coach_brief.md`, and `traces.jsonl`. Each mutated path uses a lock. Entry
+creation appends one flushed JSONL line under lock. Updates and deletes read,
+map/filter, then write a flushed temporary file and `os.replace` it.
+
+If configuration is absent and `APP_PASSWORD` exists, create:
+
+```json
+{
+  "password_hash": "pbkdf2_sha256$260000$...",
+  "session_version": 1,
+  "created_at": "ISO-8601",
+  "updated_at": "ISO-8601"
+}
+```
+
+PBKDF2 uses SHA-256, at least 200,000 iterations, a random salt, and
+`hmac.compare_digest`. Without config and bootstrap password, readiness is
+false and protected data routes return `setup_required`.
+
+`coach_brief.md` is operator-editable, read to at most 32 KiB, and seeded with
+generic policy only. Debug includes its hash and first 500 characters unless
+`DEBUG_INCLUDE_FULL_BRIEF=true`.
+
+## API envelope and security
+
+Every success is `{"ok":true,"data":...,"error":null}`. Every failure is
+`{"ok":false,"data":null,"error":{"code":"...","message":"..."}}`.
+Codes are `unauthorized`, `forbidden`, `validation_error`, `not_found`,
+`setup_required`, `rate_limited`, and `internal`.
+
+Use Starlette signed cookie sessions containing `auth`, `sv`, and Unix `exp`.
+Reject expired sessions and sessions whose version differs from config.
+Cookies are HTTP-only, SameSite Lax, and secure in production. Rate-limit
+login attempts by IP in memory and return generic credential failures.
+Limit request bodies to 64 KiB. Do not mount `DATA_ROOT`. Best-effort strip
+emails and long digit runs from coach prompts. Never return stack traces.
+
+Unauthenticated routes are only `GET /api/health`, `GET /api/ready`,
+`POST /api/auth/login`, and `GET /api/auth/me`.
+
+Auth routes:
+
+- `POST /api/auth/login` with `{password}`
+- `POST /api/auth/logout`
+- `GET /api/auth/me`
+- `POST /api/auth/change-password` with old/new password; bump session version
+
+## Entries and daily rows
+
+Entry fields are `id`, `ts`, timestamps, `activity`, `happened`, `emotions`,
+`intensity`, `remedy`, `result`, `tags`, `notes`, and nested coach metadata.
+Activity is 1–500 characters, happened 1–4000, emotions at most 12,
+intensity 1–10, remedy and notes at most 2000, and tags at most 20. Normalize
+emotions to lowercase stripped text; normalize tags likewise and replace
+spaces with underscores. Results are `worked`, `partial`, `failed`, or
+`pending`; pending never enters probability denominators.
+
+Entries support create, list/filter, get, patch, delete, and per-entry coach.
+List filters are start/end date, tag, result, emotion, substring `q`, limit
+(50 default, 200 max), and offset; sort newest first.
+
+Daily rows are unique by date and contain brick, brick completion, corn
+sessions, delay flag, daydream, rerun, court, points, note, and update time.
+Enums are brick `A|B|C|D|E|S|none`, daydream `none|done|fc`, rerun `clean|R`,
+and court `closed|court`. Corn sessions range from 0 to 50.
+
+Server-authoritative daily points:
+
+```text
+2 if brick_done
++1 if corn_sessions == 0 or (corn_sessions <= 1 and delay_ok)
++1 if daydream != "fc"
++1 if rerun == "clean"
++1 if court == "closed"
+```
+
+For a seven-calendar-day window, missing days produce `incomplete`; otherwise
+28–42 is `strong`, 18–27 `mixed`, and 0–17 `escape_heavy`.
+
+## Server statistics and evidence
+
+`stats_math.py` contains pure functions only. Excluding pending outcomes:
+
+- `p_worked(r) = N(worked,r) / N(r)`
+- `p_helped(r) = N(worked|partial,r) / N(r)`
+- `p_failed(t) = N(failed,t) / N(t)`
+- `rank(r) = p_helped(r) * n/(n+k)`
+- `match(r,T) = |T intersect helped_tags(r)| / max(|T|,1)`
+- `pick(r) = rank(r) * (1 + alpha*match(r,T))`
+
+Intensity buckets are 1–3, 4–6, and 7–10. Emotions are multi-label.
+Top coach tables hide remedies below `min_n`. Server picks are the top five by
+pick score. `DATA_THIN` means fewer than ten scored entries.
+
+Daily statistics include brick done, corn okay, no FC, clean rerun, closed
+court, and average points rates. `/api/stats` returns counts, outcomes,
+breakdowns, daily rates, formula strings, and generation time.
+`/api/stats/remedies` returns rank-sorted remedies. The model never computes
+statistics.
+
+## Coach
+
+Coach is an optional OpenRouter selector/copywriter over server evidence and
+server picks. A hardcoded ordered rule table is mandatory and always available.
+A valid coach request always returns HTTP 200 with non-empty text and source
+`model`, `backup`, or `model_unparsed_fallback`.
+
+Pipeline: resolve current input; load last K truncated entries and today's
+daily row; compute evidence and picks; build system/user messages; start a
+trace; call OpenRouter; strictly parse the template; fall back on empty,
+failed, timed-out, or unparseable output; set flags; append/ring-trim trace;
+optionally persist coach metadata on the entry.
+
+OpenRouter receives `POST {BASE}/chat/completions`, Bearer authorization,
+JSON content type, `X-Title: Habit Journal`, configured model, messages,
+temperature, token limit, and timeout.
+
+Required model output:
+
+```text
+LOOP: <tag>
+FEELING: <word>
+INTENSITY_GUESS: <1-10|n/a>
+REMEDY: <one concrete action>
+NEXT_BRICK: <A|B|C|D|E|S|specific>
+DO_NOT: <one thing>
+LINE: <12 words or fewer>
+NOTE: <optional 20 words or fewer>
+DATA_THIN: <true|false>
+```
+
+The prompt forbids invented numbers, canceling committed processes merely from
+fear, pity, and pep talks. Self-harm language triggers crisis redirection.
+Rules are ordered: crisis, court, rerun, rage_movie, daydream/fc, urge/corn,
+bully, shame, spain/admin, build/earn, home, loneliness, default. Non-crisis
+rules may substitute server pick number one.
+
+Flags include `DATA_THIN`, `PARSE_FAIL`, `TIMEOUT`, `EMPTY_MODEL`,
+`NO_API_KEY`, and `OUT_OF_EVIDENCE`.
+
+## Traces, debug, export, and operations
+
+Every coach call stores request, current situation, truncated history,
+evidence, picks, brief hash/excerpt/full value according to config, both
+prompts, model settings, latency/status/error, raw output, parse, source,
+backup rule, flags, app version, and final text. Keep the newest configured N.
+
+Authenticated debug endpoints list traces, return full traces, return one or
+the newest context-free Markdown paste bundle, preview prompts without a model
+call, and test coach with optional forced backup. The paste bundle contains:
+how an external AI should help, authoritative formulas, evidence, picks,
+current input, history, prompts, raw response, parse/source/flags, final text,
+and a reviewer checklist for invented numbers, ignored picks, format,
+fallback quality, evidence length, and weight tuning.
+
+Authenticated exports provide entries JSONL, entries CSV, and daily JSONL.
+CSV has a fixed entry header and pipe-joins emotions and tags.
+
+Health is public. Readiness reports writable data plus config presence.
+Authenticated settings status reports app name, coach configured state, model,
+data status, environment, and app version—never an API key.
+
+Mount `./static` at `/` only when present, after `/api` routes. Never mount the
+data directory.
+
+## Default generic policy
+
+- A: admin/critical checklist item
+- B: environment/hygiene one act
+- C: build/earn one ship
+- D: logistics one line
+- E: boundary one line then silence
+- S: stop spiral — water/shower/sleep
+
+Court is closed by default. Daydream requires a same-day brick. Urge policy is
+delay then brick, with a ceiling of one session per day expressed only in
+policy text.
+
+## Ordered implementation slices
+
+1. S0: Dockerfile, requirements, settings, paths, fs utilities, health/ready
+2. S1: password hashing, config store, auth, session, dependencies
+3. S2: entries CRUD JSONL
+4. S3: daily upsert, points, range band
+5. S4: pure statistics and stats routes
+6. S5: evidence and server picks
+7. S6: rule backup and offline coach
+8. S7: OpenRouter, parser, flags
+9. S8: traces and debug paste endpoints
+10. S9: export and settings status
+11. S10: seed brief, production checks, error polish
+
+Do not begin S7 before S6. Tests must cover daily point edge cases,
+hand-calculated helped/rank math, and backup rule precedence; an authenticated
+temporary-data API test is optional.
+
+## Acceptance
+
+Storage survives restart; good login sets a cookie and bad login is 401;
+unauthenticated data access is 401; daily and probability fixtures match;
+coach without a key and failed model calls return backup text; evidence and
+debug paste contain probabilities and server picks; paste contains formulas;
+authenticated export works; production OpenAPI is absent; and no secret is
+committed.
+
+Explicit decisions are final: JSONL only, server-only math, probabilities and
+picks passed to the model, hardcoded backup, context-free debug bundles,
+single shared password, entry result as truth, and private life details only
+in the operator brief on disk.

requirements.txt

@@ -0,0 +1,7 @@
+fastapi
+uvicorn[standard]
+pydantic
+pydantic-settings
+httpx
+python-multipart
+itsdangerous

tests/test_backup.py

@@ -0,0 +1,38 @@
+"""Unit tests for backup rule priority selection."""
+
+from app.backup_replies import backup_reply, select_backup_rule
+
+
+def test_court_beats_default() -> None:
+    rule = select_backup_rule("I opened court again", [])
+    assert rule["id"] == "court"
+
+
+def test_tag_match_rerun() -> None:
+    rule = select_backup_rule("something vague", ["rerun"])
+    assert rule["id"] == "rerun"
+
+
+def test_default_and_server_pick() -> None:
+    text, rule_id = backup_reply(
+        "ordinary day",
+        [],
+        [{"remedy_key": "walk outside", "pick": 0.5, "n": 5, "p_helped": 0.8}],
+        data_thin=True,
+    )
+    assert rule_id == "default"
+    assert "REMEDY: walk outside" in text
+    assert "DATA_THIN: true" in text
+    assert text.startswith("LOOP:")
+
+
+def test_crisis_ignores_server_pick() -> None:
+    text, rule_id = backup_reply(
+        "I want to kill myself",
+        [],
+        [{"remedy_key": "walk outside", "pick": 0.5, "n": 5, "p_helped": 0.8}],
+        data_thin=False,
+    )
+    assert rule_id == "crisis"
+    assert "walk outside" not in text
+    assert "crisis" in text.lower()

tests/test_daily_points.py

@@ -0,0 +1,82 @@
+"""Unit tests for server-authoritative daily point scoring."""
+
+from app.models import Court, Daydream, Rerun
+from app.store_daily import compute_daily_points, week_band
+
+
+def test_max_points() -> None:
+    assert (
+        compute_daily_points(
+            brick_done=True,
+            corn_sessions=0,
+            delay_ok=True,
+            daydream=Daydream.none,
+            rerun=Rerun.clean,
+            court=Court.closed,
+        )
+        == 6
+    )
+
+
+def test_corn_one_with_delay_ok() -> None:
+    assert (
+        compute_daily_points(
+            brick_done=False,
+            corn_sessions=1,
+            delay_ok=True,
+            daydream=Daydream.done,
+            rerun=Rerun.clean,
+            court=Court.closed,
+        )
+        == 4
+    )
+
+
+def test_corn_one_without_delay() -> None:
+    assert (
+        compute_daily_points(
+            brick_done=False,
+            corn_sessions=1,
+            delay_ok=False,
+            daydream=Daydream.none,
+            rerun=Rerun.clean,
+            court=Court.closed,
+        )
+        == 3
+    )
+
+
+def test_corn_two_loses_point() -> None:
+    assert (
+        compute_daily_points(
+            brick_done=True,
+            corn_sessions=2,
+            delay_ok=True,
+            daydream=Daydream.none,
+            rerun=Rerun.clean,
+            court=Court.closed,
+        )
+        == 5
+    )
+
+
+def test_fc_and_court_and_rerun() -> None:
+    assert (
+        compute_daily_points(
+            brick_done=True,
+            corn_sessions=0,
+            delay_ok=True,
+            daydream=Daydream.fc,
+            rerun=Rerun.R,
+            court=Court.court,
+        )
+        == 3
+    )
+
+
+def test_week_band() -> None:
+    assert week_band(6, 42) == "incomplete"
+    assert week_band(7, 28) == "strong"
+    assert week_band(7, 27) == "mixed"
+    assert week_band(7, 18) == "mixed"
+    assert week_band(7, 17) == "escape_heavy"

tests/test_stats_math.py

@@ -0,0 +1,97 @@
+"""Unit tests for pure stats math fixtures."""
+
+from app.stats_math import by_remedy, data_thin, server_picks
+
+
+def _entries() -> list[dict]:
+    # remedy "walk": 4 worked, 1 failed => n=5, p_helped=0.8
+    # remedy "nap": 2 partial, 1 failed, 2 pending => scored n=3 (below min_n=5)
+    rows = []
+    for _ in range(4):
+        rows.append(
+            {
+                "remedy": " Walk ",
+                "result": "worked",
+                "emotions": ["calm"],
+                "tags": ["rerun"],
+                "intensity": 5,
+            }
+        )
+    rows.append(
+        {
+            "remedy": "walk",
+            "result": "failed",
+            "emotions": ["shame"],
+            "tags": ["urge"],
+            "intensity": 8,
+        }
+    )
+    for _ in range(2):
+        rows.append(
+            {
+                "remedy": "nap",
+                "result": "partial",
+                "emotions": ["tired"],
+                "tags": ["home"],
+                "intensity": 2,
+            }
+        )
+    rows.append(
+        {
+            "remedy": "nap",
+            "result": "failed",
+            "emotions": ["tired"],
+            "tags": ["home"],
+            "intensity": 2,
+        }
+    )
+    rows.append(
+        {
+            "remedy": "nap",
+            "result": "pending",
+            "emotions": ["tired"],
+            "tags": ["home"],
+            "intensity": 2,
+        }
+    )
+    rows.append(
+        {
+            "remedy": "nap",
+            "result": "pending",
+            "emotions": ["tired"],
+            "tags": ["home"],
+            "intensity": 2,
+        }
+    )
+    return rows
+
+
+def test_p_helped_and_rank() -> None:
+    rows = by_remedy(_entries(), min_n=5, shrink_k=3)
+    assert len(rows) == 1
+    walk = rows[0]
+    assert walk["key"] == "walk"
+    assert walk["n"] == 5
+    assert walk["p_worked"] == 0.8
+    assert walk["p_helped"] == 0.8
+    assert abs(walk["rank"] - (0.8 * 5 / 8)) < 1e-9
+
+
+def test_data_thin() -> None:
+    assert data_thin(9) is True
+    assert data_thin(10) is False
+
+
+def test_server_picks_tag_boost() -> None:
+    picks = server_picks(
+        _entries(),
+        ["rerun"],
+        min_n=5,
+        shrink_k=3,
+        match_alpha=0.5,
+    )
+    assert len(picks) == 1
+    assert picks[0]["remedy_key"] == "walk"
+    assert picks[0]["match"] == 1.0
+    expected = (0.8 * 5 / 8) * (1 + 0.5 * 1.0)
+    assert abs(picks[0]["pick"] - expected) < 1e-9