kuhyx/diet-guard
1
0
Fork 0
mirror of https://github.com/kuhyx/diet-guard.git synced 2026-11-10 19:27:14 +01:00
Code Issues Actions 7 Packages Projects Releases Wiki Activity

Make PC sync event-driven; retire the 15-min timer

Browse Source 
The periodic diet-guard-sync.timer paid a network round-trip every ~15 min to
approximate the two moments the shared log actually changes on this device.
Sync now fires on those events instead:

* before a lock -- already present in _cli_gate._should_lock, which pulls only
  when a lock is otherwise due, so a phone-logged meal clears the lock and a
  not-due tick never touches the network;
* after a meal is logged here -- new _sync_events.publish_after_log, wired into
  all three write paths (_cli_log, _cli_gate, _mcp). This is what stops the
  PHONE nagging for a slot already logged on the PC, the mirror image of the
  rule that keeps the phone's own tick unconditional.

The gate's publish sits in cmd_gate AFTER MealGate.run() returns, never in the
unlock handler: that runs on the Tk thread with the window still up, so a
hanging network call would keep the user locked in behind a successful log. It
also covers the "Synced from another device" unlock, which hooking log_meal
would miss.

Accepted trade: a PC that is on but neither logging nor locking no longer
pulls, so an inbound phone edit lands at the next log or lock rather than
within 15 min. Bounded by slot spacing -- an unlogged day raises a lock.

install.sh now disables and removes the installed timer (guarded, since a fresh
machine has no such unit and `disable` would abort under `set -euo pipefail`);
the .service stays for manual one-off ticks.

Every test that logs a meal would otherwise reach the network through the new
call sites, so conftest patches publish_after_log at each of the three call
sites -- not on _sync_events, so test_sync_events still exercises the real
helper. Suite stays at ~8s, which is the actual leak detector.

Two files had to be split to commit at all, since file-length passes filenames
and both were already over the 250-line cap:
  _cli.py 466 -> 207, + _cli_args.py, _cli_init.py, _cli_log.py
  _mcp.py 370 -> 194, + _mcp_server.py (singleton + annotations), _mcp_read.py
Their conftest/test patch targets moved in the same commit; a stale target
redirects nothing and fails open.

Verified: 701 python tests at 100% branch coverage, 555 flutter tests,
`python -m diet_guard sync` -> 402 entries/exit 0, `gate --check` -> exit 0,
all five MCP tools still register, live ~/.local/share/diet_guard byte-identical.
Over-cap files: 44 -> 42.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RA9ynjqXPnexRQnR5oAuCC
This commit is contained in:
Krzysztof Rudnicki 2026-08-15 19:21:30 +02:00
parent 1e201acbdc
commit 38d66a8492
19 changed files with 928 additions and 547 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
CLAUDE.md
View File

@ -36,13 +36,43 @@ see the unit file's comments and `wait_for_display()`.
## Cross-device sync
`diet-guard-sync.timer` runs `python -m diet_guard sync` every ~15 min,
headless. It pulls every other device's log from `kuhyx/syncs`
A sync tick pulls every other device's log from `kuhyx/syncs`
(`diet-guard-sync/`, dumb file storage via the REST Contents API), merges
(`_sync_merge.merge_logs`: union by `id`, tombstone wins, legacy
`(time, desc)` dedup), **re-signs every persisted entry**, rebuilds the food
bank, pushes back.
**On the PC, sync is event-driven — there is no periodic timer.** The old
`diet-guard-sync.timer` (~15 min) was deleted: it paid a network round-trip on
every idle tick to approximate the two moments state actually changes here.
- **Before a lock** — `_cli_gate._should_lock` runs the cheap local
`gate_is_due()` first and pulls *only* when a lock is otherwise due, so a
meal just logged on the phone clears the lock and a not-due tick never
touches the network.
- **After a meal is logged** — `_sync_events.publish_after_log`, called from
all three write paths (`_cli`, `_cli_gate`, `_mcp`). This is what stops the
*phone* nagging for a slot already logged on the PC.
- The gate's publish lives in `cmd_gate` **after `MealGate.run()` returns**,
never in the unlock handler: that runs on the Tk thread with the window still
up, so a hanging network call would keep the user locked in behind a
successful log. It also covers the "Synced from another device" unlock, which
hooking `log_meal` would miss.
- Accepted trade: a PC that is on but neither logging nor locking does not
pull, so an inbound phone edit lands at the next log or lock rather than
within 15 min. Bounded by slot spacing — an unlogged day raises a lock.
- **`gate --demo` publishes too.** The call sits after the `finally`, so it
runs in demo as well. "Safe, closeable" means it cannot trap you at the
keyboard — not that it is offline: demo already writes real entries through
`log_meal` (only the *slot* is synthetic), so publishing them is consistent
rather than a new leak.
- `pull_shared_log()` is misnamed: it runs a *full* tick (pull, merge, re-sign,
persist, **push**, plus budget and both food banks), not just a pull.
- Every test that logs a meal would otherwise hit the network through these new
call sites; `conftest._isolate_state` patches `publish_after_log` at each of
the three call sites (not on `_sync_events`, so a direct test still exercises
the real helper).
- **Re-sign on every merge, not just phone-origin entries.** `_entry_is_valid()`
drops unsigned entries once a machine has the HMAC key, and the phone never
holds that key — skipping the re-sign loses every phone-logged meal on the

2
app/pubspec.yaml
View File

@ -2,7 +2,7 @@ name: diet_guard_app
description: "Companion phone app for diet_guard: log meals on the go."
publish_to: 'none'
version: 1.0.15+1
version: 1.0.16+1
environment:
sdk: ^3.12.2

14
diet-guard-sync.timer
View File

@ -1,14 +0,0 @@
[Unit]
Description=Periodically sync the Diet Guard log with other devices
[Timer]
# Every 15 minutes -- matches the phone app's WorkManager periodic sync (15
# min is also Android's WorkManager periodic floor), so both sides converge
# on roughly the same cadence. Persistent catches a run missed while the
# machine was suspended; OnCalendar re-arms every period regardless (see
# diet-guard-gate.timer's comment for why OnCalendar over OnBootSec here too).
OnCalendar=*-*-* *:00/15:00
Persistent=true
[Install]
WantedBy=timers.target

289
diet_guard/_cli.py
View File

@ -16,36 +16,26 @@ are left of the day's budget, plus which meal slots still need logging.
from __future__ import annotations
import argparse
from dataclasses import dataclass
import sys
from typing import TYPE_CHECKING
from diet_guard._budget import (
Biometrics,
BudgetFileCorruptError,
BudgetNotInitializedError,
compute_target_budget,
daily_budget,
protein_target_g,
write_budget,
)
from diet_guard._cli_averages import cmd_averages, register_averages_subparser
from diet_guard._cli_args import parse_args
from diet_guard._cli_averages import cmd_averages
from diet_guard._cli_gate import cmd_gate
from diet_guard._cli_sync import cmd_sync, register_sync_subparser
from diet_guard._foodbank import remember_food
from diet_guard._cli_init import cmd_init
from diet_guard._cli_log import ManualMacroArgs, Portion, cmd_ate
from diet_guard._cli_sync import cmd_sync
from diet_guard._gate import due_slots
from diet_guard._portions import (
DEFAULT_ITEM_GRAMS,
estimate_unit_grams,
)
from diet_guard._resolve import ManualMacros, resolve_nutrition
from diet_guard._slots import day_slots, slot_for_log, slot_label
from diet_guard._slots import day_slots, slot_label
from diet_guard._state import (
entry_kcal,
log_meal,
logged_slots_today,
now_local,
today_entries,
today_total_kcal,
today_total_macros,
@ -53,53 +43,13 @@ from diet_guard._state import (
)
if TYPE_CHECKING:
import argparse
from collections.abc import Callable
# Column width for a meal description in the status listing.
_DESC_WIDTH = 24
# An ISO timestamp formats as "YYYY-MM-DDTHH:MM:SS"; HH:MM is chars 11..16.
_TIME_SLICE = slice(11, 16)
# Accepted answers for the sex prompt that map to the male BMR constant.
_MALE_ANSWERS = {"m", "male"}
_FEMALE_ANSWERS = {"f", "female"}
@dataclass(frozen=True)
class _ManualMacros:
"""User-supplied calories/macros for ``ate``, all optional.
Grouping these keeps :func:`_cmd_ate` within the argument-count limit and
makes "manual values were supplied" a single, testable value object.
Attributes:
kcal: Calories entered manually (None means look the food up instead).
protein: Protein grams, recorded alongside ``kcal``.
carbs: Carbohydrate grams, recorded alongside ``kcal``.
fat: Fat grams, recorded alongside ``kcal``.
"""
kcal: float | None
protein: float | None
carbs: float | None
fat: float | None
@dataclass(frozen=True)
class _Portion:
"""How much was eaten and the basis for any typed macros.
Grouped so :func:`_cmd_ate` stays within the argument-count limit.
Attributes:
grams: Explicit grams eaten, or None.
count: Number of items eaten (an alternative to ``grams``), or None.
per_grams: Reference weight the typed macros are stated for (e.g. 100
for a per-100 g label), or None to treat the macros as totals.
"""
grams: float | None
count: float | None
per_grams: float | None
def _emit(text: str = "") -> None:
@ -117,88 +67,6 @@ def _ask(label: str) -> str:
return sys.stdin.readline().strip()
def _parse_args(argv: list[str]) -> argparse.Namespace:
"""Parse diet_guard CLI arguments."""
parser = argparse.ArgumentParser(
prog="diet_guard",
description="Log calories and check your daily budget.",
)
sub = parser.add_subparsers(dest="command", required=True)
sub.add_parser(
"init",
help="Compute your starting daily budget from biometrics.",
)
ate = sub.add_parser("ate", help="Log a meal you just ate.")
ate.add_argument("description", help='What you ate, e.g. "big mac".')
ate.add_argument(
"--grams",
type=float,
default=None,
help="Portion size in grams (default: OFF serving size, else 100 g).",
)
ate.add_argument(
"--kcal",
type=float,
default=None,
help="Calories entered manually; skips the food bank and OFF lookup.",
)
ate.add_argument(
"--protein",
type=float,
default=None,
help="Protein in grams (recorded with --kcal to seed the food bank).",
)
ate.add_argument(
"--carbs",
type=float,
default=None,
help="Carbohydrate in grams (recorded with --kcal).",
)
ate.add_argument(
"--fat",
type=float,
default=None,
help="Fat in grams (recorded with --kcal).",
)
ate.add_argument(
"--per",
type=float,
default=None,
help="Grams the macros are stated for (e.g. 100 for a per-100 g label);"
" the typed macros are scaled from this to how much you ate.",
)
ate.add_argument(
"--count",
type=float,
default=None,
help="Number of items eaten (e.g. 5 apples) instead of --grams;"
" multiplied by the staple's unit weight.",
)
sub.add_parser("status", help="Show today's calories and budget band.")
sub.add_parser("undo", help="Remove today's most recent entry.")
register_averages_subparser(sub)
register_sync_subparser(sub)
gate = sub.add_parser(
"gate",
help="Log-to-unlock screen gate (intended to be run by a timer).",
)
gate.add_argument(
"--check",
action="store_true",
help="Headless: exit 0 if NOT due, 1 if a lock is due. Prints, no window.",
)
gate.add_argument(
"--demo",
action="store_true",
help="Show the lock in safe demo mode (local grab + close button).",
)
return parser.parse_args(argv)
def _print_summary() -> None:
"""Print today's total and how much of the daily budget is left.
@ -232,137 +100,6 @@ def _print_entry_line(entry: dict[str, object]) -> None:
)
def _read_init_inputs() -> tuple[Biometrics, float, float] | None:
"""Prompt for biometrics on stdin; return (bio, activity, deficit) or None.
Returns None (after printing why) on any unparsable or out-of-range input,
so a typo never sets a wrong budget.
"""
try:
weight = float(_ask("weight in kg:"))
height = float(_ask("height in cm:"))
age = float(_ask("age in years:"))
sex_raw = _ask("sex (m/f):").lower()
activity = float(
_ask(
"activity factor "
"(1.2 sedentary / 1.375 light / 1.55 moderate / 1.725 active):",
),
)
deficit = float(_ask("daily deficit in kcal (e.g. 200):"))
except ValueError:
_emit("that was not a number; nothing was set.")
return None
if sex_raw in _MALE_ANSWERS:
is_male = True
elif sex_raw in _FEMALE_ANSWERS:
is_male = False
else:
_emit('sex must be "m" or "f"; nothing was set.')
return None
bio = Biometrics(
weight_kg=weight,
height_cm=height,
age_years=age,
is_male=is_male,
)
return bio, activity, deficit
def _cmd_init() -> int:
"""Compute the starting budget from biometrics and write it."""
inputs = _read_init_inputs()
if inputs is None:
return 2
bio, activity, deficit = inputs
budget = compute_target_budget(
bio,
activity_factor=activity,
deficit_kcal=deficit,
)
write_budget(budget, weight_kg=bio.weight_kg)
_emit(f"budget computed from your biometrics: {budget:g} kcal/day.")
_emit("edit it any time from the gate's calendar tab or the phone app.")
return 0
def _eaten_grams(
description: str,
portion: _Portion,
) -> tuple[float | None, str | None]:
"""Resolve how many grams were eaten, plus a note if a weight was assumed.
A count of items is turned into grams via the staple's unit weight; an
unknown item falls back to a default weight, with a note so the estimate is
never silent.
Args:
description: The food name (used to look up a per-item weight).
portion: The user's portion inputs.
Returns:
``(grams, note)`` where ``grams`` may be None (no portion given) and
``note`` is a one-line caveat to print, or None.
"""
if portion.count is not None:
unit = estimate_unit_grams(description)
if unit is None:
return (
portion.count * DEFAULT_ITEM_GRAMS,
f"(assumed {DEFAULT_ITEM_GRAMS:g} g per item; "
"pass --grams to be exact)",
)
return portion.count * unit, None
return portion.grams, None
def _cmd_ate(description: str, portion: _Portion, macros: _ManualMacros) -> int:
"""Resolve and log a meal, tag its slot, bank it, then print the total.
Resolution order is manual, then food bank, then the staple table, then
Open Food Facts (see :func:`resolve_nutrition`). A per-item count or a
per-reference macro basis is converted to the amount actually eaten first,
and the food is remembered so next time it is served from local history.
"""
eaten, note = _eaten_grams(description, portion)
if note is not None:
_emit(note)
manual_macros = (
ManualMacros(
kcal=macros.kcal,
protein=macros.protein or 0.0,
carbs=macros.carbs or 0.0,
fat=macros.fat or 0.0,
per_grams=portion.per_grams,
)
if macros.kcal is not None
else None
)
nutrition = resolve_nutrition(
description,
grams=eaten,
manual_macros=manual_macros,
)
if nutrition is None:
_emit(
f'no food bank, staple, or Open Food Facts match for "{description}". '
"re-run with --kcal <number> to log it manually.",
)
return 1
log_meal(description, nutrition, slot_for_log(now_local()))
remember_food(description, nutrition)
macro_str = f"P{nutrition.protein_g:g} C{nutrition.carbs_g:g} F{nutrition.fat_g:g}"
portion_str = f"{nutrition.grams:g} g" if nutrition.grams else "portion n/a"
_emit(
f"logged: {description} {nutrition.kcal:g} kcal "
f"({macro_str}) [{nutrition.source}, {portion_str}]",
)
_print_summary()
return 0
def _print_slot_status() -> None:
"""Print each meal slot as logged / DUE / upcoming for today."""
logged = logged_slots_today()
@ -422,10 +159,12 @@ def _cmd_undo() -> int:
def _dispatch_ate(args: argparse.Namespace) -> int:
"""Marshal ``ate``'s flags into value objects, then log the meal."""
return _cmd_ate(
return cmd_ate(
_emit,
_print_summary,
args.description,
_Portion(grams=args.grams, count=args.count, per_grams=args.per),
_ManualMacros(
Portion(grams=args.grams, count=args.count, per_grams=args.per),
ManualMacroArgs(
kcal=args.kcal,
protein=args.protein,
carbs=args.carbs,
@ -443,7 +182,7 @@ def _dispatch_gate(args: argparse.Namespace) -> int:
# tripped ruff's return-count limit at the seventh, and a table also has no
# unreachable trailing branch for the 100%-coverage gate to chase.
_COMMANDS: dict[str, Callable[[argparse.Namespace], int]] = {
"init": lambda _args: _cmd_init(),
"init": lambda _args: cmd_init(_emit, _ask),
"ate": _dispatch_ate,
"status": lambda _args: _cmd_status(),
"averages": lambda _args: cmd_averages(_emit),
@ -462,5 +201,5 @@ def main(argv: list[str] | None = None) -> int:
Returns:
A process exit code (0 on success).
"""
args = _parse_args(sys.argv[1:] if argv is None else argv)
args = parse_args(sys.argv[1:] if argv is None else argv)
return _COMMANDS[args.command](args)

100
diet_guard/_cli_args.py Normal file
View File

@ -0,0 +1,100 @@
"""Argument parsing for the diet_guard CLI.
Split out of :mod:`diet_guard._cli` to hold the repo's 250-line cap. This is
the declarative half -- every subcommand and flag the CLI accepts -- kept apart
from the handlers that act on them, so adding a flag never grows a file that
also contains logic.
Subparsers owned by a sibling handler register themselves through their own
module (``register_averages_subparser``, ``register_sync_subparser``) rather
than being spelled out here, so a subcommand's flags live next to its code.
"""
from __future__ import annotations
import argparse
from diet_guard._cli_averages import register_averages_subparser
from diet_guard._cli_sync import register_sync_subparser
def parse_args(argv: list[str]) -> argparse.Namespace:
"""Parse diet_guard CLI arguments."""
parser = argparse.ArgumentParser(
prog="diet_guard",
description="Log calories and check your daily budget.",
)
sub = parser.add_subparsers(dest="command", required=True)
sub.add_parser(
"init",
help="Compute your starting daily budget from biometrics.",
)
ate = sub.add_parser("ate", help="Log a meal you just ate.")
ate.add_argument("description", help='What you ate, e.g. "big mac".')
ate.add_argument(
"--grams",
type=float,
default=None,
help="Portion size in grams (default: OFF serving size, else 100 g).",
)
ate.add_argument(
"--kcal",
type=float,
default=None,
help="Calories entered manually; skips the food bank and OFF lookup.",
)
ate.add_argument(
"--protein",
type=float,
default=None,
help="Protein in grams (recorded with --kcal to seed the food bank).",
)
ate.add_argument(
"--carbs",
type=float,
default=None,
help="Carbohydrate in grams (recorded with --kcal).",
)
ate.add_argument(
"--fat",
type=float,
default=None,
help="Fat in grams (recorded with --kcal).",
)
ate.add_argument(
"--per",
type=float,
default=None,
help="Grams the macros are stated for (e.g. 100 for a per-100 g label);"
" the typed macros are scaled from this to how much you ate.",
)
ate.add_argument(
"--count",
type=float,
default=None,
help="Number of items eaten (e.g. 5 apples) instead of --grams;"
" multiplied by the staple's unit weight.",
)
sub.add_parser("status", help="Show today's calories and budget band.")
sub.add_parser("undo", help="Remove today's most recent entry.")
register_averages_subparser(sub)
register_sync_subparser(sub)
gate = sub.add_parser(
"gate",
help="Log-to-unlock screen gate (intended to be run by a timer).",
)
gate.add_argument(
"--check",
action="store_true",
help="Headless: exit 0 if NOT due, 1 if a lock is due. Prints, no window.",
)
gate.add_argument(
"--demo",
action="store_true",
help="Show the lock in safe demo mode (local grab + close button).",
)
return parser.parse_args(argv)

10
diet_guard/_cli_gate.py
View File

@ -14,6 +14,7 @@ from diet_guard._gate import gate_is_due
from diet_guard._gatelock import MealGate, acquire_gate_lock, release_gate_lock
from diet_guard._gatelock_support import wait_for_display
from diet_guard._sync import pull_shared_log
from diet_guard._sync_events import publish_after_log
if TYPE_CHECKING:
from collections.abc import Callable
@ -79,4 +80,13 @@ def cmd_gate(emit: Callable[[str], None], *, check: bool, demo: bool) -> int:
MealGate(demo_mode=demo).run()
finally:
release_gate_lock(handle)
# Publish only once ``run()`` has returned -- i.e. the mainloop has ended and
# the window is gone. Deliberately NOT inside the unlock handler: that runs
# on the Tk thread with the lock still up, so a slow or hanging network call
# would keep the user locked in behind a successful log. Here the worst case
# is a late publish. One call site covers every unlock reason, including the
# "Synced from another device" path, which hooking the log itself would miss.
reason = publish_after_log()
if reason is not None:
emit(f"logged locally, not yet published ({reason}).")
return 0

85
diet_guard/_cli_init.py Normal file
View File

@ -0,0 +1,85 @@
"""CLI handler for the ``init`` subcommand.
Split out of :mod:`diet_guard._cli` to hold the repo's 250-line cap, following
the same thin-per-subcommand-handler shape as ``_cli_gate.py`` (``gate``),
``_cli_log.py`` (``ate``) and ``_cli_sync.py`` (``sync``).
The biometrics prompted for here are used **once** and discarded: only the
computed budget (and the body weight, which the protein target needs) is ever
persisted. See ``CLAUDE.md``'s "Biometrics are used once and discarded".
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from diet_guard._budget import Biometrics, compute_target_budget, write_budget
if TYPE_CHECKING:
from collections.abc import Callable
# Accepted answers for the sex prompt that map to the male BMR constant.
MALE_ANSWERS = {"m", "male"}
FEMALE_ANSWERS = {"f", "female"}
def read_init_inputs(
emit: Callable[[str], None],
ask: Callable[[str], str],
) -> tuple[Biometrics, float, float] | None:
"""Prompt for biometrics on stdin; return (bio, activity, deficit) or None.
Returns None (after printing why) on any unparsable or out-of-range input,
so a typo never sets a wrong budget.
"""
try:
weight = float(ask("weight in kg:"))
height = float(ask("height in cm:"))
age = float(ask("age in years:"))
sex_raw = ask("sex (m/f):").lower()
activity = float(
ask(
"activity factor "
"(1.2 sedentary / 1.375 light / 1.55 moderate / 1.725 active):",
),
)
deficit = float(ask("daily deficit in kcal (e.g. 200):"))
except ValueError:
emit("that was not a number; nothing was set.")
return None
if sex_raw in MALE_ANSWERS:
is_male = True
elif sex_raw in FEMALE_ANSWERS:
is_male = False
else:
emit('sex must be "m" or "f"; nothing was set.')
return None
bio = Biometrics(
weight_kg=weight,
height_cm=height,
age_years=age,
is_male=is_male,
)
return bio, activity, deficit
def cmd_init(
emit: Callable[[str], None],
ask: Callable[[str], str],
) -> int:
"""Compute the starting budget from biometrics and write it."""
inputs = read_init_inputs(emit, ask)
if inputs is None:
return 2
bio, activity, deficit = inputs
budget = compute_target_budget(
bio,
activity_factor=activity,
deficit_kcal=deficit,
)
write_budget(budget, weight_kg=bio.weight_kg)
emit(f"budget computed from your biometrics: {budget:g} kcal/day.")
emit("edit it any time from the gate's calendar tab or the phone app.")
return 0

162
diet_guard/_cli_log.py Normal file
View File

@ -0,0 +1,162 @@
"""CLI handler for the ``ate`` subcommand.
Split out of :mod:`diet_guard._cli` to hold the repo's 250-line cap, following
the same thin-per-subcommand-handler shape as ``_cli_gate.py`` (``gate``) and
``_cli_sync.py`` (``sync``). The portion/macro value objects live here rather
than in ``_cli`` because ``_cmd_ate`` is their only consumer -- the argparse
layer builds them and hands them straight over.
This is also where a local write turns into a publish: see
:func:`diet_guard._sync_events.publish_after_log` for why that is event-driven
rather than a periodic timer.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING
from diet_guard._foodbank import remember_food
from diet_guard._portions import DEFAULT_ITEM_GRAMS, estimate_unit_grams
from diet_guard._resolve import ManualMacros, resolve_nutrition
from diet_guard._slots import slot_for_log
from diet_guard._state import log_meal, now_local
from diet_guard._sync_events import publish_after_log
if TYPE_CHECKING:
from collections.abc import Callable
@dataclass(frozen=True)
class ManualMacroArgs:
"""User-supplied calories/macros for ``ate``, all optional.
Grouping these keeps :func:`cmd_ate` within the argument-count limit and
makes "manual values were supplied" a single, testable value object.
Attributes:
kcal: Calories entered manually (None means look the food up instead).
protein: Protein grams, recorded alongside ``kcal``.
carbs: Carbohydrate grams, recorded alongside ``kcal``.
fat: Fat grams, recorded alongside ``kcal``.
"""
kcal: float | None
protein: float | None
carbs: float | None
fat: float | None
@dataclass(frozen=True)
class Portion:
"""How much was eaten and the basis for any typed macros.
Grouped so :func:`cmd_ate` stays within the argument-count limit.
Attributes:
grams: Explicit grams eaten, or None.
count: Number of items eaten (an alternative to ``grams``), or None.
per_grams: Reference weight the typed macros are stated for (e.g. 100
for a per-100 g label), or None to treat the macros as totals.
"""
grams: float | None
count: float | None
per_grams: float | None
def eaten_grams(
description: str,
portion: Portion,
) -> tuple[float | None, str | None]:
"""Resolve how many grams were eaten, plus a note if a weight was assumed.
A count of items is turned into grams via the staple's unit weight; an
unknown item falls back to a default weight, with a note so the estimate is
never silent.
Args:
description: The food name (used to look up a per-item weight).
portion: The user's portion inputs.
Returns:
``(grams, note)`` where ``grams`` may be None (no portion given) and
``note`` is a one-line caveat to print, or None.
"""
if portion.count is not None:
unit = estimate_unit_grams(description)
if unit is None:
return (
portion.count * DEFAULT_ITEM_GRAMS,
f"(assumed {DEFAULT_ITEM_GRAMS:g} g per item; "
"pass --grams to be exact)",
)
return portion.count * unit, None
return portion.grams, None
def cmd_ate(
emit: Callable[[str], None],
print_summary: Callable[[], None],
description: str,
portion: Portion,
macros: ManualMacroArgs,
) -> int:
"""Resolve and log a meal, tag its slot, bank it, publish, print the total.
Resolution order is manual, then food bank, then the staple table, then
Open Food Facts (see :func:`resolve_nutrition`). A per-item count or a
per-reference macro basis is converted to the amount actually eaten first,
and the food is remembered so next time it is served from local history.
Args:
emit: A one-line output sink (``_cli._emit``, passed in rather than
imported -- see ``_cli_sync.cmd_sync`` for why).
print_summary: Prints today's total and remaining budget.
description: The food as the user typed it.
portion: How much was eaten.
macros: Any manually supplied calories/macros.
Returns:
0 once logged, or 1 when the food could not be resolved at all.
"""
eaten, note = eaten_grams(description, portion)
if note is not None:
emit(note)
manual_macros = (
ManualMacros(
kcal=macros.kcal,
protein=macros.protein or 0.0,
carbs=macros.carbs or 0.0,
fat=macros.fat or 0.0,
per_grams=portion.per_grams,
)
if macros.kcal is not None
else None
)
nutrition = resolve_nutrition(
description,
grams=eaten,
manual_macros=manual_macros,
)
if nutrition is None:
emit(
f'no food bank, staple, or Open Food Facts match for "{description}". '
"re-run with --kcal <number> to log it manually.",
)
return 1
log_meal(description, nutrition, slot_for_log(now_local()))
remember_food(description, nutrition)
macro_str = f"P{nutrition.protein_g:g} C{nutrition.carbs_g:g} F{nutrition.fat_g:g}"
portion_str = f"{nutrition.grams:g} g" if nutrition.grams else "portion n/a"
emit(
f"logged: {description} {nutrition.kcal:g} kcal "
f"({macro_str}) [{nutrition.source}, {portion_str}]",
)
# Publish straight away rather than waiting for a periodic tick: until this
# lands, the phone still believes this slot is unlogged and will nag for it.
reason = publish_after_log()
if reason is not None:
emit(f"logged locally, not yet published ({reason}).")
print_summary()
return 0

230
diet_guard/_mcp.py
View File

@ -33,211 +33,24 @@ Safety invariants (do not break when adding tools):
from __future__ import annotations
import logging
import sys
from typing import Any
from mcp.server import MCPServer
from mcp.types import ToolAnnotations
from pydantic import BaseModel
from diet_guard._averages import (
PeriodAverage,
band_label,
monthly_average,
weekly_average,
)
from diet_guard._budget import BudgetError, current_schedule, daily_budget
from diet_guard._gate import due_slots
# Importing these IS what registers the read tools: the module-level
# ``@mcp.tool`` decorators in ``_mcp_read`` run against the shared singleton at
# import time. They are re-exported via ``__all__`` rather than imported bare so
# the names are genuinely used -- drop this and the server starts write-only.
from diet_guard._mcp_read import get_averages, get_slots, get_status, list_today
from diet_guard._mcp_server import APPENDS, logger, mcp
from diet_guard._resolve import ManualMacros, resolve_nutrition
from diet_guard._slots import current_slot, day_slots, slot_for_log, slot_label
from diet_guard._state import (
consumption_band,
load_log,
logged_slots_today,
now_local,
today_entries,
today_total_kcal,
today_total_macros,
)
from diet_guard._slots import slot_for_log
from diet_guard._state import (
log_meal as record_meal,
)
# Log to STDERR only -- STDOUT carries the MCP JSON-RPC protocol frames, so a
# single stray stdout write would corrupt the stream and kill the session.
logging.basicConfig(
level=logging.INFO,
stream=sys.stderr,
format="%(asctime)s [%(levelname)s] diet-guard-mcp: %(message)s",
)
logger = logging.getLogger(__name__)
mcp = MCPServer("diet-guard")
# Machine-readable versions of what the docstrings already say, so a client
# can decide what to run unattended. The day a "read" tool starts writing,
# these annotations are what has to change with it.
_READS_ONLY = ToolAnnotations(
read_only_hint=True,
idempotent_hint=True,
open_world_hint=False,
)
# log_meal is the one tool that writes. It appends and never rewrites or
# deletes, so it is not destructive; it is not idempotent either, because a
# second confirmed call logs a second meal. It reaches Open Food Facts when
# macros are not supplied, which is exactly what open_world means.
_APPENDS = ToolAnnotations(
read_only_hint=False,
destructive_hint=False,
idempotent_hint=False,
open_world_hint=True,
)
def _entry_view(entry: dict[str, object]) -> dict[str, Any]:
"""Project a stored log entry to a safe, JSON-friendly summary.
Deliberately drops the ``hmac`` field and any unknown keys: the signature
is internal integrity metadata, not something a client needs.
Args:
entry: A stored food-log entry.
Returns:
A flat dict of the display-relevant fields.
"""
return {
"time": entry.get("time"),
"desc": entry.get("desc"),
"kcal": entry.get("kcal"),
"protein_g": entry.get("protein_g"),
"carbs_g": entry.get("carbs_g"),
"fat_g": entry.get("fat_g"),
"grams": entry.get("grams"),
"source": entry.get("source"),
"slot": entry.get("slot"),
}
# ──────────────────────────────────────────────────────────────
# Read tools (consumed intake + qualitative band only; NEVER the budget number)
# ──────────────────────────────────────────────────────────────
@mcp.tool(title="Today's intake status", annotations=_READS_ONLY)
def get_status() -> dict[str, Any]:
"""Return today's intake status without ever revealing the daily budget.
Reports the calories and macros *consumed* so far, the qualitative
:func:`consumption_band` (``"on track"`` / ``"approaching limit"`` /
``"OVER BUDGET"``, or ``None`` when no budget has been set yet), and the
meal-slot picture (which slots are due, which are already logged, and the
current slot). The raw budget number is intentionally withheld -- only the
band is exposed, mirroring how the CLI status shows a label to an automated
caller rather than the anchor number.
"""
try:
band: str | None = consumption_band()
except BudgetError:
# No budget set (or a corrupt file): surface the absence, not a number.
band = None
return {
"consumed_kcal": today_total_kcal(),
"consumed_macros_g": _macros_dict(today_total_macros()),
"consumption_band": band,
"budget_initialized": band is not None,
"due_slots": [slot_label(slot) for slot in due_slots()],
"logged_slots": sorted(logged_slots_today()),
"current_slot": current_slot(now_local()),
}
@mcp.tool(title="Meals logged today", annotations=_READS_ONLY)
def list_today() -> dict[str, Any]:
"""List today's logged meals (valid entries only), newest last.
Returns the per-entry description, calories, macros, portion, source, and
slot -- the same data the CLI ``status`` listing renders, minus the internal
HMAC signature.
"""
entries = today_entries()
return {
"count": len(entries),
"entries": [_entry_view(entry) for entry in entries],
}
def _period_view(period: PeriodAverage) -> dict[str, Any]:
"""Render one period for the wire, WITHOUT its average budget.
``avg_budget`` is dropped on purpose: it is the daily budget by another
name, and the module invariant is that the number never leaves via this
interface. What remains -- the mean intake plus the qualitative band -- is
the same trade ``get_status`` already makes (a band plus a consumed number
bounds the budget loosely; the exact figure stays on the machine).
"""
return {
"start": period.start,
"end": period.end,
"logged_days": period.logged_days,
"elapsed_days": period.elapsed_days,
"avg_kcal": period.avg_kcal,
"band": None if period.band is None else period.band.value,
"band_label": band_label(period.band),
}
@mcp.tool(title="Weekly and monthly average intake", annotations=_READS_ONLY)
def get_averages() -> dict[str, Any]:
"""Return average kcal/day for this and last week, and this and last month.
Each period's average is over its *logged* days only and is classified
``"under"`` / ``"slightly_over"`` / ``"very_over"`` against the mean of the
budgets that applied on those same days -- never against today's budget, so
a budget edit does not retroactively reclassify past weeks.
Every period ends at **yesterday**: today is still being logged, and a
half-logged day would drag the mean far enough to flip the band. A period
with no finished logged day yet reports ``avg_kcal: null`` rather than a
flattering zero.
The per-period average budget is deliberately not returned (see the module
docstring's budget invariant); only the intake and the band are.
"""
try:
schedule = current_schedule(default=daily_budget())
except BudgetError:
# No budget set, or a corrupt file: no yardstick, so no bands.
return {"budget_initialized": False, "periods": {}}
log = load_log()
periods = {
"this_week": weekly_average(log, schedule=schedule, weeks_ago=0),
"last_week": weekly_average(log, schedule=schedule, weeks_ago=1),
"this_month": monthly_average(log, schedule=schedule, months_ago=0),
"last_month": monthly_average(log, schedule=schedule, months_ago=1),
}
return {
"budget_initialized": True,
"excludes_today": True,
"periods": {name: _period_view(period) for name, period in periods.items()},
}
@mcp.tool(title="Meal slots for the day", annotations=_READS_ONLY)
def get_slots() -> dict[str, Any]:
"""Return the day's fixed meal slots and which one is current.
Pure schedule information (08:00 / 12:00 / 16:00 / 20:00 by default) with no
budget or intake data attached.
"""
return {
"day_slots": [
{"hour": slot, "label": slot_label(slot)} for slot in day_slots()
],
"current_slot": current_slot(now_local()),
}
from diet_guard._state import now_local
from diet_guard._sync_events import publish_after_log
# ──────────────────────────────────────────────────────────────
# Gated write tool (preview unless confirm=True; NEVER allowlist this)
@ -260,7 +73,7 @@ class Macros(BaseModel):
fat: float = 0.0
@mcp.tool(title="Log a meal (gated write)", annotations=_APPENDS)
@mcp.tool(title="Log a meal (gated write)", annotations=APPENDS)
def log_meal(
description: str,
grams: float | None = None,
@ -343,6 +156,9 @@ def log_meal(
logger.warning("log_meal write failed: %s", exc)
return {"ok": False, "reason": "could not write the food log."}
logger.info("log_meal applied: %s (%g kcal)", description, nutrition.kcal)
# Same rule as the CLI: publish on the event, not on a clock. Fail-closed,
# so a sync outage still reports the local write as applied.
published = publish_after_log() is None
return {
"ok": True,
"applied": True,
@ -351,15 +167,10 @@ def log_meal(
"logged": resolved,
"target_slot": target_slot,
"signed": "hmac" in entry,
"published": published,
}
def _macros_dict(macros: tuple[float, float, float]) -> dict[str, float]:
"""Return a ``(protein, carbs, fat)`` triple as a labelled dict."""
protein, carbs, fat = macros
return {"protein": protein, "carbs": carbs, "fat": fat}
def main() -> None:
"""Run the MCP server over stdio (STDOUT = JSON-RPC, STDERR = logs)."""
logger.info("Starting diet-guard MCP server (python=%s)", sys.executable)
@ -368,3 +179,16 @@ def main() -> None:
if __name__ == "__main__":
main()
#: Re-exported so ``from diet_guard._mcp import get_status`` keeps working for
#: the tests and any client that imported them from here before the split.
__all__ = [
"Macros",
"get_averages",
"get_slots",
"get_status",
"list_today",
"log_meal",
"main",
]

186
diet_guard/_mcp_read.py Normal file
View File

@ -0,0 +1,186 @@
"""The MCP server's read-only tools.
Split from :mod:`diet_guard._mcp` (which now holds the gated write tool) to
hold the repo's 250-line cap. Importing this module is what *registers* these
tools -- the ``@mcp.tool`` decorators run at import time against the singleton
in :mod:`diet_guard._mcp_server` -- so ``_mcp`` imports it for its side effect
and must keep doing so, however unused the name looks.
The budget invariant lives here as much as in ``_mcp``: no tool in this file
returns the raw budget, the remaining-budget number, the stored body weight or
the protein target. Only consumed intake and the qualitative band go out.
"""
from __future__ import annotations
from typing import Any
from diet_guard._averages import (
PeriodAverage,
band_label,
monthly_average,
weekly_average,
)
from diet_guard._budget import BudgetError, current_schedule, daily_budget
from diet_guard._gate import due_slots
from diet_guard._mcp_server import READS_ONLY, mcp
from diet_guard._slots import current_slot, day_slots, slot_label
from diet_guard._state import (
consumption_band,
load_log,
logged_slots_today,
now_local,
today_entries,
today_total_kcal,
today_total_macros,
)
def _macros_dict(macros: tuple[float, float, float]) -> dict[str, float]:
"""Return a ``(protein, carbs, fat)`` triple as a labelled dict."""
protein, carbs, fat = macros
return {"protein": protein, "carbs": carbs, "fat": fat}
def _entry_view(entry: dict[str, object]) -> dict[str, Any]:
"""Project a stored log entry to a safe, JSON-friendly summary.
Deliberately drops the ``hmac`` field and any unknown keys: the signature
is internal integrity metadata, not something a client needs.
Args:
entry: A stored food-log entry.
Returns:
A flat dict of the display-relevant fields.
"""
return {
"time": entry.get("time"),
"desc": entry.get("desc"),
"kcal": entry.get("kcal"),
"protein_g": entry.get("protein_g"),
"carbs_g": entry.get("carbs_g"),
"fat_g": entry.get("fat_g"),
"grams": entry.get("grams"),
"source": entry.get("source"),
"slot": entry.get("slot"),
}
# ──────────────────────────────────────────────────────────────
# Read tools (consumed intake + qualitative band only; NEVER the budget number)
# ──────────────────────────────────────────────────────────────
@mcp.tool(title="Today's intake status", annotations=READS_ONLY)
def get_status() -> dict[str, Any]:
"""Return today's intake status without ever revealing the daily budget.
Reports the calories and macros *consumed* so far, the qualitative
:func:`consumption_band` (``"on track"`` / ``"approaching limit"`` /
``"OVER BUDGET"``, or ``None`` when no budget has been set yet), and the
meal-slot picture (which slots are due, which are already logged, and the
current slot). The raw budget number is intentionally withheld -- only the
band is exposed, mirroring how the CLI status shows a label to an automated
caller rather than the anchor number.
"""
try:
band: str | None = consumption_band()
except BudgetError:
# No budget set (or a corrupt file): surface the absence, not a number.
band = None
return {
"consumed_kcal": today_total_kcal(),
"consumed_macros_g": _macros_dict(today_total_macros()),
"consumption_band": band,
"budget_initialized": band is not None,
"due_slots": [slot_label(slot) for slot in due_slots()],
"logged_slots": sorted(logged_slots_today()),
"current_slot": current_slot(now_local()),
}
@mcp.tool(title="Meals logged today", annotations=READS_ONLY)
def list_today() -> dict[str, Any]:
"""List today's logged meals (valid entries only), newest last.
Returns the per-entry description, calories, macros, portion, source, and
slot -- the same data the CLI ``status`` listing renders, minus the internal
HMAC signature.
"""
entries = today_entries()
return {
"count": len(entries),
"entries": [_entry_view(entry) for entry in entries],
}
def _period_view(period: PeriodAverage) -> dict[str, Any]:
"""Render one period for the wire, WITHOUT its average budget.
``avg_budget`` is dropped on purpose: it is the daily budget by another
name, and the module invariant is that the number never leaves via this
interface. What remains -- the mean intake plus the qualitative band -- is
the same trade ``get_status`` already makes (a band plus a consumed number
bounds the budget loosely; the exact figure stays on the machine).
"""
return {
"start": period.start,
"end": period.end,
"logged_days": period.logged_days,
"elapsed_days": period.elapsed_days,
"avg_kcal": period.avg_kcal,
"band": None if period.band is None else period.band.value,
"band_label": band_label(period.band),
}
@mcp.tool(title="Weekly and monthly average intake", annotations=READS_ONLY)
def get_averages() -> dict[str, Any]:
"""Return average kcal/day for this and last week, and this and last month.
Each period's average is over its *logged* days only and is classified
``"under"`` / ``"slightly_over"`` / ``"very_over"`` against the mean of the
budgets that applied on those same days -- never against today's budget, so
a budget edit does not retroactively reclassify past weeks.
Every period ends at **yesterday**: today is still being logged, and a
half-logged day would drag the mean far enough to flip the band. A period
with no finished logged day yet reports ``avg_kcal: null`` rather than a
flattering zero.
The per-period average budget is deliberately not returned (see the module
docstring's budget invariant); only the intake and the band are.
"""
try:
schedule = current_schedule(default=daily_budget())
except BudgetError:
# No budget set, or a corrupt file: no yardstick, so no bands.
return {"budget_initialized": False, "periods": {}}
log = load_log()
periods = {
"this_week": weekly_average(log, schedule=schedule, weeks_ago=0),
"last_week": weekly_average(log, schedule=schedule, weeks_ago=1),
"this_month": monthly_average(log, schedule=schedule, months_ago=0),
"last_month": monthly_average(log, schedule=schedule, months_ago=1),
}
return {
"budget_initialized": True,
"excludes_today": True,
"periods": {name: _period_view(period) for name, period in periods.items()},
}
@mcp.tool(title="Meal slots for the day", annotations=READS_ONLY)
def get_slots() -> dict[str, Any]:
"""Return the day's fixed meal slots and which one is current.
Pure schedule information (08:00 / 12:00 / 16:00 / 20:00 by default) with no
budget or intake data attached.
"""
return {
"day_slots": [
{"hour": slot, "label": slot_label(slot)} for slot in day_slots()
],
"current_slot": current_slot(now_local()),
}

49
diet_guard/_mcp_server.py Normal file
View File

@ -0,0 +1,49 @@
"""The MCP server singleton and its shared tool annotations.
Extracted from :mod:`diet_guard._mcp` to hold the repo's 250-line cap. It has
to be a *separate* module rather than a section of ``_mcp``: the tool modules
need ``mcp`` at import time to register themselves with ``@mcp.tool``, so
keeping the singleton alongside them would make every split a circular import.
Also the single place stdio logging is configured, which is load-bearing:
STDOUT carries the MCP JSON-RPC protocol frames, so a stray write there
corrupts the stream and kills the session. Everything logs to STDERR.
"""
from __future__ import annotations
import logging
import sys
from mcp.server import MCPServer
from mcp.types import ToolAnnotations
# Log to STDERR only -- STDOUT carries the MCP JSON-RPC protocol frames, so a
# single stray stdout write would corrupt the stream and kill the session.
logging.basicConfig(
level=logging.INFO,
stream=sys.stderr,
format="%(asctime)s [%(levelname)s] diet-guard-mcp: %(message)s",
)
logger = logging.getLogger("diet_guard._mcp")
mcp = MCPServer("diet-guard")
# Machine-readable versions of what the docstrings already say, so a client
# can decide what to run unattended. The day a "read" tool starts writing,
# these annotations are what has to change with it.
READS_ONLY = ToolAnnotations(
read_only_hint=True,
idempotent_hint=True,
open_world_hint=False,
)
# log_meal is the one tool that writes. It appends and never rewrites or
# deletes, so it is not destructive; it is not idempotent either, because a
# second confirmed call logs a second meal. It reaches Open Food Facts when
# macros are not supplied, which is exactly what open_world means.
APPENDS = ToolAnnotations(
read_only_hint=False,
destructive_hint=False,
idempotent_hint=False,
open_world_hint=True,
)

50
diet_guard/_sync_events.py Normal file
View File

@ -0,0 +1,50 @@
"""Event-driven sync triggers: publish when state changes, not on a clock.
diet_guard used to run a ``diet-guard-sync.timer`` every ~15 min. That cadence
was a poor proxy for the two moments the shared log actually changes on this
device, and it paid a network round-trip on every idle tick in between:
* **before a lock** -- already handled in :func:`diet_guard._cli_gate._should_lock`,
which pulls only when a lock is otherwise due, so a meal just logged on the
phone clears the lock without a manual re-entry.
* **after a meal is logged here** -- this module. Publishing immediately is
what stops the *phone* nagging for a slot already logged on the PC; the
mirror image of the rule in ``CLAUDE.md`` that keeps the phone's tick
unconditional.
Both go through :func:`~diet_guard._sync.pull_shared_log`, which despite its
name runs a *full* tick (pull, merge, re-sign, persist, push, plus budget and
both food banks) and fails closed -- a sync outage returns a reason string
rather than raising, so it can never turn a successful local log into an error.
The trade this makes explicit: a PC that is on but neither logging nor locking
no longer pulls at all, so an inbound phone edit lands at the next log or lock
rather than within 15 min. That is bounded by the meal-slot spacing, since a
day that goes unlogged raises a lock by construction.
"""
from __future__ import annotations
import logging
from diet_guard._sync import pull_shared_log
_logger = logging.getLogger(__name__)
def publish_after_log() -> str | None:
"""Run a full sync tick after a local write, swallowing any outage.
Call this *after* the local log write is durable, never before: the local
entry is the source of truth for this device, and a network failure must
leave it intact and the caller's own success path untouched.
Returns:
``None`` when the tick completed, or a short human-readable reason when
it could not -- callers that have somewhere to show it (the CLI) print
it; callers that do not (the gate, the MCP tool) let it fall to the log.
"""
reason = pull_shared_log()
if reason is not None:
_logger.info("post-log sync skipped: %s", reason)
return reason

8
diet_guard/tests/conftest.py
View File

@ -144,6 +144,14 @@ def _isolate_state(tmp_path: Path) -> Iterator[None]:
"diet_guard._sync_client.CONFIG_FILE",
tmp_path / "nonexistent-firebase.json",
),
# Logging a meal now publishes immediately (`_sync_events`), so every
# test that logs one would otherwise open a real connection through a
# path that has nothing to do with what it is asserting. Patched at the
# *call sites* rather than on `_sync_events`, so a test that imports
# `publish_after_log` directly still exercises the real helper.
patch("diet_guard._cli_log.publish_after_log", return_value=None),
patch("diet_guard._cli_gate.publish_after_log", return_value=None),
patch("diet_guard._mcp.publish_after_log", return_value=None),
):
yield

45
diet_guard/tests/test_cli.py
View File

@ -11,13 +11,14 @@ import io
from typing import TYPE_CHECKING
from unittest.mock import patch
from diet_guard import _cli
from diet_guard import _cli, _cli_log
from diet_guard._budget import (
BudgetFileCorruptError,
BudgetNotInitializedError,
write_budget,
)
from diet_guard._cli import _eaten_grams, _Portion, main
from diet_guard._cli import main
from diet_guard._cli_log import Portion, eaten_grams
from diet_guard._estimator import Nutrition
if TYPE_CHECKING:
@ -37,23 +38,21 @@ class TestEatenGrams:
def test_count_of_known_staple(self) -> None:
"""A count of a known staple multiplies by its unit weight, no note."""
grams, note = _eaten_grams(
"apple", _Portion(grams=None, count=5, per_grams=None)
)
grams, note = eaten_grams("apple", Portion(grams=None, count=5, per_grams=None))
assert grams == 5 * 182
assert note is None
def test_count_of_unknown_item_warns(self) -> None:
"""A count of an unknown item uses the default and flags the assumption."""
grams, note = _eaten_grams(
"mystery", _Portion(grams=None, count=3, per_grams=None)
grams, note = eaten_grams(
"mystery", Portion(grams=None, count=3, per_grams=None)
)
assert grams is not None
assert note is not None
def test_explicit_grams(self) -> None:
"""An explicit gram portion passes straight through."""
grams, note = _eaten_grams("x", _Portion(grams=300, count=None, per_grams=None))
grams, note = eaten_grams("x", Portion(grams=300, count=None, per_grams=None))
assert grams == 300
assert note is None
@ -113,22 +112,46 @@ class TestAte:
def test_logs_and_summarizes(self, capsys: pytest.CaptureFixture[str]) -> None:
"""A resolved meal is logged, banked, and summarized."""
write_budget(2000)
with patch.object(_cli, "resolve_nutrition", return_value=_NUT):
with patch.object(_cli_log, "resolve_nutrition", return_value=_NUT):
assert main(["ate", "big mac"]) == 0
assert "logged:" in capsys.readouterr().out
def test_publishes_after_logging(self) -> None:
"""The meal is published immediately, not left for a periodic tick."""
write_budget(2000)
with (
patch.object(_cli_log, "resolve_nutrition", return_value=_NUT),
patch.object(_cli_log, "publish_after_log", return_value=None) as publish,
):
assert main(["ate", "big mac"]) == 0
publish.assert_called_once_with()
def test_sync_outage_still_logs_locally(
self, capsys: pytest.CaptureFixture[str]
) -> None:
"""A failed publish reports itself without failing the local log."""
write_budget(2000)
with (
patch.object(_cli_log, "resolve_nutrition", return_value=_NUT),
patch.object(_cli_log, "publish_after_log", return_value="no token"),
):
assert main(["ate", "big mac"]) == 0
out = capsys.readouterr().out
assert "logged:" in out
assert "not yet published (no token)" in out
def test_note_printed_for_assumed_weight(
self, capsys: pytest.CaptureFixture[str]
) -> None:
"""An assumed per-item weight prints its caveat."""
write_budget(2000)
with patch.object(_cli, "resolve_nutrition", return_value=_NUT):
with patch.object(_cli_log, "resolve_nutrition", return_value=_NUT):
main(["ate", "mystery", "--count", "3"])
assert "assumed" in capsys.readouterr().out
def test_unresolved_food(self, capsys: pytest.CaptureFixture[str]) -> None:
"""An unresolvable food returns a failure and a manual-entry hint."""
with patch.object(_cli, "resolve_nutrition", return_value=None):
with patch.object(_cli_log, "resolve_nutrition", return_value=None):
assert main(["ate", "nonsense"]) == 1
assert "--kcal" in capsys.readouterr().out

44
diet_guard/tests/test_cli_gate.py
View File

@ -80,6 +80,50 @@ class TestCmdGate:
factory.assert_not_called()
assert "no lock needed" in lines[0]
def test_publishes_only_after_the_window_closes(self) -> None:
"""The publish must not run while the lock is still up.
A network call on the Tk thread would keep the user locked in behind a
successful log, so ordering here is the actual guarantee, not the call.
"""
calls: list[str] = []
gate = MagicMock()
gate.run.side_effect = lambda: calls.append("run")
with (
patch.object(_cli_gate, "gate_is_due", return_value=True),
_no_sync(),
patch.object(_cli_gate, "MealGate", return_value=gate),
patch.object(_cli_gate, "acquire_gate_lock", return_value=MagicMock()),
patch.object(
_cli_gate,
"release_gate_lock",
side_effect=lambda _h: calls.append("release"),
),
patch.object(_cli_gate, "wait_for_display", return_value=True),
patch.object(
_cli_gate,
"publish_after_log",
side_effect=lambda: calls.append("publish"),
),
):
assert cmd_gate([].append, check=False, demo=False) == 0
assert calls == ["run", "release", "publish"]
def test_publish_outage_reported_after_unlock(self) -> None:
"""A failed post-unlock publish reports without re-trapping the user."""
lines: list[str] = []
with (
patch.object(_cli_gate, "gate_is_due", return_value=True),
_no_sync(),
patch.object(_cli_gate, "MealGate", return_value=MagicMock()),
patch.object(_cli_gate, "acquire_gate_lock", return_value=MagicMock()),
patch.object(_cli_gate, "release_gate_lock"),
patch.object(_cli_gate, "wait_for_display", return_value=True),
patch.object(_cli_gate, "publish_after_log", return_value="offline"),
):
assert cmd_gate(lines.append, check=False, demo=False) == 0
assert "not yet published (offline)" in lines[-1]
def test_gate_already_running(self) -> None:
"""A held single-instance lock means a second window is not opened."""
lines: list[str] = []

62
diet_guard/tests/test_mcp.py
View File

@ -10,7 +10,7 @@ from __future__ import annotations
from unittest.mock import patch
from diet_guard import _mcp
from diet_guard import _mcp, _mcp_read
from diet_guard._budget import BudgetNotInitializedError
from diet_guard._estimator import Nutrition
@ -29,14 +29,16 @@ def _nutrition(kcal: float = 250.0) -> Nutrition:
class TestGetStatus:
def test_band_and_intake(self) -> None:
with (
patch.object(_mcp, "today_total_kcal", return_value=640.0),
patch.object(_mcp, "today_total_macros", return_value=(30.0, 50.0, 12.0)),
patch.object(_mcp, "consumption_band", return_value="on track"),
patch.object(_mcp, "due_slots", return_value=(12,)),
patch.object(_mcp, "logged_slots_today", return_value={8}),
patch.object(_mcp, "current_slot", return_value=12),
patch.object(_mcp_read, "today_total_kcal", return_value=640.0),
patch.object(
_mcp_read, "today_total_macros", return_value=(30.0, 50.0, 12.0)
),
patch.object(_mcp_read, "consumption_band", return_value="on track"),
patch.object(_mcp_read, "due_slots", return_value=(12,)),
patch.object(_mcp_read, "logged_slots_today", return_value={8}),
patch.object(_mcp_read, "current_slot", return_value=12),
patch.object(_mcp, "now_local"),
patch.object(_mcp, "slot_label", return_value="12:00"),
patch.object(_mcp_read, "slot_label", return_value="12:00"),
):
out = _mcp.get_status()
assert out["consumed_kcal"] == 640.0
@ -49,16 +51,16 @@ class TestGetStatus:
def test_no_budget_hides_band(self) -> None:
with (
patch.object(_mcp, "today_total_kcal", return_value=0.0),
patch.object(_mcp, "today_total_macros", return_value=(0.0, 0.0, 0.0)),
patch.object(_mcp_read, "today_total_kcal", return_value=0.0),
patch.object(_mcp_read, "today_total_macros", return_value=(0.0, 0.0, 0.0)),
patch.object(
_mcp,
_mcp_read,
"consumption_band",
side_effect=BudgetNotInitializedError(),
),
patch.object(_mcp, "due_slots", return_value=()),
patch.object(_mcp, "logged_slots_today", return_value=set()),
patch.object(_mcp, "current_slot", return_value=None),
patch.object(_mcp_read, "due_slots", return_value=()),
patch.object(_mcp_read, "logged_slots_today", return_value=set()),
patch.object(_mcp_read, "current_slot", return_value=None),
patch.object(_mcp, "now_local"),
):
out = _mcp.get_status()
@ -84,7 +86,7 @@ class TestListToday:
"hmac": "deadbeef",
},
]
with patch.object(_mcp, "today_entries", return_value=entries):
with patch.object(_mcp_read, "today_entries", return_value=entries):
out = _mcp.list_today()
assert out["count"] == 1
view = out["entries"][0]
@ -94,7 +96,7 @@ class TestListToday:
assert "hmac" not in view
def test_empty(self) -> None:
with patch.object(_mcp, "today_entries", return_value=[]):
with patch.object(_mcp_read, "today_entries", return_value=[]):
out = _mcp.list_today()
assert out == {"count": 0, "entries": []}
@ -102,10 +104,10 @@ class TestListToday:
class TestGetSlots:
def test_lists_slots_and_current(self) -> None:
with (
patch.object(_mcp, "day_slots", return_value=(8, 12, 16, 20)),
patch.object(_mcp, "current_slot", return_value=16),
patch.object(_mcp_read, "day_slots", return_value=(8, 12, 16, 20)),
patch.object(_mcp_read, "current_slot", return_value=16),
patch.object(_mcp, "now_local"),
patch.object(_mcp, "slot_label", side_effect=lambda s: f"{s:02d}:00"),
patch.object(_mcp_read, "slot_label", side_effect=lambda s: f"{s:02d}:00"),
):
out = _mcp.get_slots()
assert out["current_slot"] == 16
@ -179,6 +181,28 @@ class TestLogMealGate:
assert out["applied"] is True
assert out["signed"] is False
def test_confirm_publishes_and_reports_it(self) -> None:
"""A successful write publishes immediately and says so."""
with (
patch.object(_mcp, "resolve_nutrition", return_value=_nutrition()),
patch.object(_mcp, "record_meal", return_value={"desc": "apple"}),
patch.object(_mcp, "publish_after_log", return_value=None) as publish,
):
out = _mcp.log_meal("apple", slot=8, confirm=True)
publish.assert_called_once_with()
assert out["published"] is True
def test_confirm_applies_even_when_publish_fails(self) -> None:
"""A sync outage leaves the local write applied, flagged unpublished."""
with (
patch.object(_mcp, "resolve_nutrition", return_value=_nutrition()),
patch.object(_mcp, "record_meal", return_value={"desc": "apple"}),
patch.object(_mcp, "publish_after_log", return_value="offline"),
):
out = _mcp.log_meal("apple", slot=8, confirm=True)
assert out["applied"] is True
assert out["published"] is False
def test_confirm_write_failure_returns_gracefully(self) -> None:
with (
patch.object(_mcp, "resolve_nutrition", return_value=_nutrition()),

30
diet_guard/tests/test_mcp_averages.py
View File

@ -4,7 +4,7 @@ from __future__ import annotations
from unittest.mock import patch
from diet_guard import _mcp
from diet_guard import _mcp, _mcp_read
from diet_guard._averages import AverageBand, PeriodAverage
from diet_guard._budget import BudgetNotInitializedError
@ -28,12 +28,12 @@ def _period(
class TestPeriodView:
def test_never_carries_the_average_budget(self) -> None:
# The module invariant: no tool hands an automated caller the budget.
view = _mcp._period_view(_period())
view = _mcp_read._period_view(_period())
assert "avg_budget" not in view
assert 2000.0 not in view.values()
def test_carries_intake_band_and_coverage(self) -> None:
view = _mcp._period_view(_period())
view = _mcp_read._period_view(_period())
assert view["avg_kcal"] == 2100.0
assert view["band"] == "slightly_over"
assert view["band_label"] == "slightly over"
@ -43,7 +43,7 @@ class TestPeriodView:
assert view["end"] == "2026-01-11"
def test_empty_period_serializes_as_null_not_zero(self) -> None:
view = _mcp._period_view(_period(avg_kcal=None, band=None))
view = _mcp_read._period_view(_period(avg_kcal=None, band=None))
assert view["avg_kcal"] is None
assert view["band"] is None
assert view["band_label"] == "no data"
@ -52,11 +52,11 @@ class TestPeriodView:
class TestGetAverages:
def test_returns_all_four_periods(self) -> None:
with (
patch.object(_mcp, "daily_budget", return_value=2000),
patch.object(_mcp, "current_schedule"),
patch.object(_mcp, "load_log", return_value={}),
patch.object(_mcp, "weekly_average", return_value=_period()),
patch.object(_mcp, "monthly_average", return_value=_period()),
patch.object(_mcp_read, "daily_budget", return_value=2000),
patch.object(_mcp_read, "current_schedule"),
patch.object(_mcp_read, "load_log", return_value={}),
patch.object(_mcp_read, "weekly_average", return_value=_period()),
patch.object(_mcp_read, "monthly_average", return_value=_period()),
):
out = _mcp.get_averages()
assert out["budget_initialized"] is True
@ -70,7 +70,7 @@ class TestGetAverages:
def test_no_budget_degrades_to_empty_periods(self) -> None:
with patch.object(
_mcp,
_mcp_read,
"daily_budget",
side_effect=BudgetNotInitializedError,
):
@ -79,11 +79,11 @@ class TestGetAverages:
def test_no_response_field_leaks_the_budget_number(self) -> None:
with (
patch.object(_mcp, "daily_budget", return_value=2000),
patch.object(_mcp, "current_schedule"),
patch.object(_mcp, "load_log", return_value={}),
patch.object(_mcp, "weekly_average", return_value=_period()),
patch.object(_mcp, "monthly_average", return_value=_period()),
patch.object(_mcp_read, "daily_budget", return_value=2000),
patch.object(_mcp_read, "current_schedule"),
patch.object(_mcp_read, "load_log", return_value={}),
patch.object(_mcp_read, "weekly_average", return_value=_period()),
patch.object(_mcp_read, "monthly_average", return_value=_period()),
):
out = _mcp.get_averages()
flat = repr(out)

54
diet_guard/tests/test_sync_events.py Normal file
View File

@ -0,0 +1,54 @@
"""Tests for the event-driven post-log sync trigger.
These exercise the real :func:`publish_after_log`, not the conftest stub: the
autouse fixture patches it at each *call site*, so importing it here still
reaches the genuine helper (see ``conftest._isolate_state``).
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from unittest.mock import patch
from diet_guard import _sync_events
from diet_guard._sync_events import publish_after_log
if TYPE_CHECKING:
import pytest
class TestPublishAfterLog:
"""A full tick fired by a local write, which must never raise."""
def test_success_returns_none(self) -> None:
"""A completed tick reports no reason."""
with patch.object(
_sync_events,
"pull_shared_log",
return_value=None,
) as pull:
assert publish_after_log() is None
pull.assert_called_once_with()
def test_outage_returns_reason(self) -> None:
"""A failed tick surfaces the reason instead of raising."""
with patch.object(
_sync_events,
"pull_shared_log",
return_value="sync unavailable (no token)",
):
assert publish_after_log() == "sync unavailable (no token)"
def test_outage_is_logged(self, caplog: pytest.LogCaptureFixture) -> None:
"""The reason reaches the log for callers with nowhere to show it."""
with (
patch.object(
_sync_events,
"pull_shared_log",
return_value="sync unavailable (boom)",
),
caplog.at_level(logging.INFO, logger="diet_guard._sync_events"),
):
publish_after_log()
assert "boom" in caplog.text

21
install.sh
View File

@ -27,7 +27,6 @@ readonly REPO_DIR="$SCRIPT_DIR"
readonly SERVICE_SRC="$SCRIPT_DIR/diet-guard-gate.service"
readonly TIMER_SRC="$SCRIPT_DIR/diet-guard-gate.timer"
readonly SYNC_SERVICE_SRC="$SCRIPT_DIR/diet-guard-sync.service"
readonly SYNC_TIMER_SRC="$SCRIPT_DIR/diet-guard-sync.timer"
readonly SYSTEMD_USER_DIR="$HOME/.config/systemd/user"
readonly DATA_DIR="$HOME/.local/share/diet_guard"
readonly BUDGET_FILE="$DATA_DIR/.budget"
@ -60,18 +59,26 @@ systemctl --user daemon-reload
systemctl --user enable --now diet-guard-gate.timer
echo " Timer enabled and started (fires the gate every ~30 min)."
# 4. systemd user timer + service (sync) -------------------------------------
echo "[4/5] Installing the sync's systemd user timer + service..."
# 4. Retire the old periodic sync timer ---------------------------------------
# Sync is event-driven now (before a lock, after a meal is logged -- see
# diet_guard/_sync_events.py), so the ~15 min timer is gone. Removing the unit
# from the repo does NOT stop an already-installed copy: without this block it
# would keep firing forever from $SYSTEMD_USER_DIR with no unit file in git to
# explain it. `|| true` because a fresh machine has no such unit and `disable`
# exits non-zero there, which would abort the install under `set -euo pipefail`.
echo "[4/5] Removing the retired periodic sync timer (sync is event-driven)..."
systemctl --user disable --now diet-guard-sync.timer 2>/dev/null || true
rm -f "$SYSTEMD_USER_DIR/diet-guard-sync.timer"
# The *service* stays: it is how a manual one-off tick is run
# (`systemctl --user start diet-guard-sync`). Only the timer is retired.
cp "$SYNC_SERVICE_SRC" "$SYSTEMD_USER_DIR/diet-guard-sync.service"
cp "$SYNC_TIMER_SRC" "$SYSTEMD_USER_DIR/diet-guard-sync.timer"
systemctl --user daemon-reload
systemctl --user enable --now diet-guard-sync.timer
echo " Timer enabled and started (syncs the log every ~15 min)."
echo " Periodic sync timer removed; 'python -m diet_guard sync' still works."
if [[ -e "$SYNC_TOKEN_FILE" ]]; then
echo " Sync token already present at $SYNC_TOKEN_FILE."
else
echo " No sync token yet at $SYNC_TOKEN_FILE -- sync will no-op (and log a"
echo " failure) on every tick until you create a fine-grained GitHub PAT"
echo " failure) after every meal you log, until you create a GitHub PAT"
echo " scoped to the diet-guard-sync repo's contents and save it there,"
echo " mode 600: chmod 600 \"$SYNC_TOKEN_FILE\""
fi