Add sync-ready log schema: stable id, tombstone undo, composite components

Lays the Python-side groundwork for the phone companion app's eventual
log sync (Milestone 0 of the diet-app-as-wise-balloon plan):
- log_meal() stamps every entry with a UUID id for union-by-id merging
  across devices.
- undo_last_today() now tombstones (deleted: true, re-signed) instead of
  popping the entry, so a sync merge can't resurrect a stale copy of
  something already undone on another device.
- log_meal() accepts components, the full per-item macros for a
  composite meal, so a food bank rebuilt purely by replaying the log can
  recover every component's standalone nutrition.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FU3f5KQ1GHXsbbSecfVEyF
This commit is contained in:
Krzysztof kuhy Rudnicki 2026-06-22 18:22:24 +02:00
parent 8ed2082854
commit 888c877048
4 changed files with 174 additions and 20 deletions

View File

@ -17,7 +17,7 @@ from diet_guard._budget import BudgetError, daily_budget, protein_target_g
from diet_guard._foodbank import remember_food, remember_meal from diet_guard._foodbank import remember_food, remember_meal
from diet_guard._gatelock_nutrition import _GateNutrition from diet_guard._gatelock_nutrition import _GateNutrition
from diet_guard._gatelock_ui import ERR, FG, UNIT_GRAMS from diet_guard._gatelock_ui import ERR, FG, UNIT_GRAMS
from diet_guard._meal import MealItem, meal_total from diet_guard._meal import MealItem, item_to_component, meal_total
from diet_guard._resolve import lookup_candidates from diet_guard._resolve import lookup_candidates
from diet_guard._slots import slot_label from diet_guard._slots import slot_label
from diet_guard._state import ( from diet_guard._state import (
@ -220,7 +220,12 @@ class _GateMealFlow(_GateNutrition):
name = self._meal_name() or _DEFAULT_MEAL_NAME name = self._meal_name() or _DEFAULT_MEAL_NAME
count = len(self._state.meal_items) count = len(self._state.meal_items)
total = remember_meal(name, list(self._state.meal_items)) total = remember_meal(name, list(self._state.meal_items))
log_meal(name, total, self._slot_for_log()) log_meal(
name,
total,
self._slot_for_log(),
components=[item_to_component(item) for item in self._state.meal_items],
)
self._state.meal_items = [] self._state.meal_items = []
self._finish_slot(f"{name}: {total.kcal:g} kcal ({count} items)") self._finish_slot(f"{name}: {total.kcal:g} kcal ({count} items)")

View File

@ -63,3 +63,31 @@ def meal_total(items: Sequence[MealItem]) -> Nutrition:
grams=round(sum((item.nutrition.grams for item in items), 0.0), 1), grams=round(sum((item.nutrition.grams for item in items), 0.0), 1),
source=MEAL_SOURCE, source=MEAL_SOURCE,
) )
def item_to_component(item: MealItem) -> dict[str, object]:
"""Return a composite meal's per-component log record for ``item``.
The food bank's own ``components`` field (see
:func:`diet_guard._foodbank.remember_meal`) stores only component
*names* -- it is rebuilt from the log, not the other way round. This
record carries the component's full macros so a bank rebuilt purely by
replaying the log (the companion phone app's sync model) can recover
each component's standalone nutrition, not just the composite's summed
total.
Args:
item: One component of a composite meal.
Returns:
A plain dict of the component's name and macros, suitable for the
``components`` list passed to :func:`diet_guard._state.log_meal`.
"""
return {
"name": item.name,
"kcal": item.nutrition.kcal,
"protein_g": item.nutrition.protein_g,
"carbs_g": item.nutrition.carbs_g,
"fat_g": item.nutrition.fat_g,
"grams": item.nutrition.grams,
}

View File

@ -14,6 +14,7 @@ from datetime import datetime, timezone
import json import json
import logging import logging
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
import uuid
from gatelock.log_integrity import ( from gatelock.log_integrity import (
compute_entry_hmac, compute_entry_hmac,
@ -116,6 +117,8 @@ def log_meal(
description: str, description: str,
nutrition: Nutrition, nutrition: Nutrition,
slot: int | None = None, slot: int | None = None,
*,
components: list[dict[str, object]] | None = None,
) -> dict[str, object]: ) -> dict[str, object]:
"""Append a signed entry for ``description`` to today's log. """Append a signed entry for ``description`` to today's log.
@ -125,11 +128,17 @@ def log_meal(
slot: The meal-slot hour this entry satisfies (e.g. ``12`` for the slot: The meal-slot hour this entry satisfies (e.g. ``12`` for the
12:00 checkpoint). When None the entry still counts toward the 12:00 checkpoint). When None the entry still counts toward the
day's calories but does not mark any slot as logged. day's calories but does not mark any slot as logged.
components: For a composite (multi-item) meal, each component's own
name and macros. Carried on the log entry itself -- not just the
food bank -- so a bank rebuilt purely by replaying the log (the
companion phone app's sync model) can recover every component's
standalone nutrition, not just the composite's summed total.
Returns: Returns:
The stored entry dict (carrying an ``hmac`` field when a key exists). The stored entry dict (carrying an ``hmac`` field when a key exists).
""" """
entry: dict[str, object] = { entry: dict[str, object] = {
"id": str(uuid.uuid4()),
"time": now_local().isoformat(timespec="seconds"), "time": now_local().isoformat(timespec="seconds"),
"desc": description, "desc": description,
"grams": nutrition.grams, "grams": nutrition.grams,
@ -141,6 +150,8 @@ def log_meal(
} }
if slot is not None: if slot is not None:
entry["slot"] = slot entry["slot"] = slot
if components is not None:
entry["components"] = list(components)
signature = compute_entry_hmac(entry) signature = compute_entry_hmac(entry)
if signature is not None: if signature is not None:
entry["hmac"] = signature entry["hmac"] = signature
@ -154,11 +165,21 @@ def log_meal(
def load_log() -> DayLog: def load_log() -> DayLog:
"""Return the log with only valid (untampered) entries retained.""" """Return the log with only valid, non-deleted entries retained.
A "deleted" entry is a tombstone left by :func:`undo_last_today`, not a
removal: it is kept on disk (and re-signed) rather than popped, so a sync
merge with another device can see the tombstone and not resurrect a
stale copy of the same entry. Readers simply filter it out here.
"""
raw = _read_raw_log() raw = _read_raw_log()
verified: DayLog = {} verified: DayLog = {}
for day, entries in raw.items(): for day, entries in raw.items():
kept = [entry for entry in entries if _entry_is_valid(entry)] kept = [
entry
for entry in entries
if _entry_is_valid(entry) and not entry.get("deleted")
]
if kept: if kept:
verified[day] = kept verified[day] = kept
return verified return verified
@ -246,23 +267,34 @@ def consumption_band() -> str:
def undo_last_today() -> dict[str, object] | None: def undo_last_today() -> dict[str, object] | None:
"""Remove and return today's most recently logged entry, if any. """Tombstone today's most recently logged, not-yet-undone entry.
Operates on the raw log so a mistaken entry can always be removed, even one Marks the entry ``deleted`` in place and re-signs it, rather than
that would not pass verification. physically removing it: a sync merge with another device only ever
*adds* entries it hasn't seen before, so a physical delete here would be
silently resurrected the next time that device's stale copy is pulled
back in. The tombstone travels with the entry instead, and every reader
(:func:`load_log`, the food-bank rebuild) already skips it.
Operates on the raw log so a mistaken entry can always be undone, even
one that would not pass verification.
Returns: Returns:
The removed entry, or None if nothing was logged today. The tombstoned entry, or None if nothing undoable was logged today.
""" """
log = _read_raw_log() log = _read_raw_log()
today = _today() today = _today()
entries = log.get(today) entries = log.get(today)
if not entries: if not entries:
return None return None
removed = entries.pop() for entry in reversed(entries):
if entries: if entry.get("deleted"):
log[today] = entries continue
else: entry["deleted"] = True
del log[today] entry.pop("hmac", None)
_write_log(log) signature = compute_entry_hmac(entry)
return removed if signature is not None:
entry["hmac"] = signature
_write_log(log)
return entry
return None

View File

@ -225,15 +225,51 @@ class TestBudgetViews:
assert consumption_band() == "OVER BUDGET" assert consumption_band() == "OVER BUDGET"
class TestIdAndComponents:
"""New per-entry fields the companion phone app's sync relies on."""
def test_entry_has_id(self) -> None:
"""Every logged entry carries a UUID id."""
entry = log_meal("toast", _nut(150), slot=8)
assert isinstance(entry["id"], str)
assert entry["id"]
def test_ids_are_unique(self) -> None:
"""Two entries never collide on id."""
first = log_meal("a", _nut(1), slot=8)
second = log_meal("b", _nut(1), slot=12)
assert first["id"] != second["id"]
def test_components_omitted_by_default(self) -> None:
"""A single-food entry carries no components field."""
entry = log_meal("toast", _nut(150), slot=8)
assert "components" not in entry
def test_components_carried_through(self) -> None:
"""A composite meal's component macros are stored on the entry."""
parts = [
{
"name": "chicken",
"kcal": 165.0,
"protein_g": 31.0,
"carbs_g": 0.0,
"fat_g": 3.6,
"grams": 100.0,
}
]
entry = log_meal("dinner", _nut(165), slot=20, components=parts)
assert entry["components"] == parts
class TestUndo: class TestUndo:
"""Removing the most recent entry.""" """Tombstoning the most recent entry."""
def test_nothing_to_undo(self) -> None: def test_nothing_to_undo(self) -> None:
"""An empty day undoes to None.""" """An empty day undoes to None."""
assert undo_last_today() is None assert undo_last_today() is None
def test_undo_leaves_earlier_entries(self) -> None: def test_undo_leaves_earlier_entries(self) -> None:
"""Undo removes only the last entry when others remain.""" """Undo tombstones only the last entry when others remain."""
log_meal("a", _nut(100), slot=8) log_meal("a", _nut(100), slot=8)
log_meal("b", _nut(200), slot=12) log_meal("b", _nut(200), slot=12)
removed = undo_last_today() removed = undo_last_today()
@ -241,8 +277,61 @@ class TestUndo:
assert removed["desc"] == "b" assert removed["desc"] == "b"
assert today_total_kcal() == 100.0 assert today_total_kcal() == 100.0
def test_undo_last_entry_clears_day(self) -> None: def test_undo_tombstones_in_place(self) -> None:
"""Undoing the only entry removes the day from the log.""" """Undoing the only entry keeps it on disk, marked deleted."""
log_meal("a", _nut(100), slot=8) log_meal("a", _nut(100), slot=8)
undo_last_today() undo_last_today()
assert _state._read_raw_log() == {} raw = _raw()
day = next(iter(raw))
assert len(raw[day]) == 1
assert raw[day][0]["deleted"] is True
def test_undo_tombstone_excluded_from_reads(self) -> None:
"""A tombstoned entry no longer counts toward totals or slots."""
log_meal("a", _nut(100), slot=8)
undo_last_today()
assert today_total_kcal() == 0.0
assert today_entries() == []
assert logged_slots_today() == set()
def test_undo_re_signs_the_tombstone(self) -> None:
"""The mutated (tombstoned) entry still carries a valid signature."""
log_meal("a", _nut(100), slot=8)
undo_last_today()
raw = _raw()
day = next(iter(raw))
assert "hmac" in raw[day][0]
def test_undo_unsigned_when_no_key(self) -> None:
"""Re-signing a tombstone with no key available leaves it unsigned."""
log_meal("a", _nut(100), slot=8)
with patch.object(_state, "compute_entry_hmac", return_value=None):
undo_last_today()
raw = _raw()
day = next(iter(raw))
assert "hmac" not in raw[day][0]
def test_undo_skips_already_tombstoned(self) -> None:
"""Undoing twice tombstones the prior entry, not the same one again."""
log_meal("a", _nut(100), slot=8)
log_meal("b", _nut(200), slot=12)
undo_last_today()
second = undo_last_today()
assert second is not None
assert second["desc"] == "a"
def test_undo_nothing_left_once_all_tombstoned(self) -> None:
"""Once every entry today is tombstoned, undo returns None."""
log_meal("a", _nut(100), slot=8)
undo_last_today()
assert undo_last_today() is None
class TestLoadLogSkipsTombstones:
"""``load_log`` filters out deleted entries the same way as invalid ones."""
def test_day_with_only_a_tombstone_is_omitted(self) -> None:
"""A day whose sole entry is tombstoned is dropped entirely."""
log_meal("a", _nut(100), slot=8)
undo_last_today()
assert load_log() == {}