diff --git a/.gitignore b/.gitignore index 434f74c..36d3a8f 100644 --- a/.gitignore +++ b/.gitignore @@ -17,6 +17,8 @@ htmlcov/ # Runtime state files — personal workout/health data, never public. screen_locker/log.json +# ...and the ledger naming which of its entries record the same workout. +screen_locker/duplicate_credits.json # ...including the timestamped backups a correction script leaves behind. screen_locker/*.json.bak-* screen_locker/sick_history.json diff --git a/screen_locker/tests/test_annotate_duplicate_credits.py b/screen_locker/tests/test_annotate_duplicate_credits.py new file mode 100644 index 0000000..6f4ac97 --- /dev/null +++ b/screen_locker/tests/test_annotate_duplicate_credits.py @@ -0,0 +1,163 @@ +"""Tests for the duplicate-credit ledger script. + +The ledger is documentation, not behaviour -- nothing reads it to decide a +lock. Its one hard requirement is that it never contradicts +:func:`~screen_locker._weekly_check.count_day_credits`, because a ledger that +describes a different rule than the locker uses is worse than no ledger at all. +That invariant is enforced at runtime by ``check_ledger_agrees`` and pinned +here. +""" + +from __future__ import annotations + +import sys +from typing import Any, NoReturn + +import pytest + +from scripts.annotate_duplicate_credits import ( + build_ledger, + check_ledger_agrees, +) + + +def _raising_exit(code: object = None) -> NoReturn: + """Stand in for the real ``sys.exit``. + + Args: + code: The exit status, as ``sys.exit`` takes. + + Raises: + SystemExit: Always, exactly as ``sys.exit`` does. + """ + raise SystemExit(code) + + +@pytest.fixture +def restore_sys_exit(monkeypatch: pytest.MonkeyPatch) -> None: + """Give this test a ``sys.exit`` that actually exits. + + The autouse ``_block_real_tk_and_exit`` fixture in conftest patches + ``screen_locker.screen_lock.sys.exit``. That attribute path resolves to the + *global* ``sys`` module object, not a per-module copy, so the patch replaces + ``sys.exit`` for every module in the process for the duration of every test + -- including this script's. Without this fixture the guard under test is a + no-op and the test passes for the wrong reason. + + Args: + monkeypatch: pytest's attribute patcher, which undoes this on teardown. + """ + monkeypatch.setattr(sys, "exit", _raising_exit) + + +def _entry(wtype: str, timestamp: str) -> dict[str, Any]: + """Build a log entry with a workout_id, as the real writer does. + + Args: + wtype: The ``workout_data.type`` value. + timestamp: ISO-8601 timestamp. + + Returns: + A log entry dict. + """ + return { + "timestamp": timestamp, + "workout_data": {"type": wtype}, + "workout_id": f"{wtype}:{timestamp[:10]}", + } + + +class TestBuildLedger: + """The ledger names exactly the entries that share one credit.""" + + def test_stronglifts_pair_is_recorded(self) -> None: + """A phone/PC pair on one day becomes one ledger record.""" + logs = { + "2026-08-24": [ + _entry("pc_workout_verified", "2026-08-24T10:55:29+00:00"), + _entry("phone_verified", "2026-08-24T10:55:38+00:00"), + ], + } + ledger = build_ledger(logs) + assert list(ledger) == ["2026-08-24"] + record = ledger["2026-08-24"][0] + assert record["credit"] == "stronglifts" + # Earliest timestamp is the canonical entry; the later copy is absorbed. + assert record["counted"] == "pc_workout_verified:2026-08-24" + assert record["duplicate_of"] == ["phone_verified:2026-08-24"] + + def test_days_without_a_shared_credit_are_absent(self) -> None: + """Separate workouts produce no ledger entry at all.""" + logs = { + "2026-08-24": [ + _entry("phone_verified", "2026-08-24T10:00:00+00:00"), + _entry("runnerup_verified", "2026-08-24T19:00:00+00:00"), + ], + "2026-08-25": [_entry("manual_workout", "2026-08-25T10:00:00+00:00")], + } + assert build_ledger(logs) == {} + + def test_uncredited_entries_are_ignored(self) -> None: + """A skip marker earns no credit, so it can never share one.""" + logs = { + "2026-08-25": [ + _entry("relaxed_day_skip", "2026-08-25T18:00:00+00:00"), + _entry("relaxed_day_skip", "2026-08-25T19:00:00+00:00"), + ], + } + assert build_ledger(logs) == {} + + def test_entry_without_workout_id_falls_back_to_position(self) -> None: + """Legacy entries predating workout_id are still nameable.""" + logs = { + "2026-08-24": [ + { + "timestamp": "2026-08-24T10:00:00+00:00", + "workout_data": {"type": "phone_verified"}, + }, + { + "timestamp": "2026-08-24T11:00:00+00:00", + "workout_data": {"type": "pc_workout_verified"}, + }, + ], + } + record = build_ledger(logs)["2026-08-24"][0] + assert record["counted"] == "2026-08-24#0" + assert record["duplicate_of"] == ["2026-08-24#1"] + + +class TestLedgerAgreesWithTheCreditRule: + """The ledger's arithmetic must match count_day_credits exactly.""" + + def test_generated_ledger_agrees(self) -> None: + """A ledger built from a log always agrees with that log.""" + logs = { + "2026-08-24": [ + _entry("pc_workout_verified", "2026-08-24T10:55:29+00:00"), + _entry("phone_verified", "2026-08-24T10:55:38+00:00"), + ], + "2026-08-27": [_entry("manual_workout", "2026-08-27T18:45:32+00:00")], + } + check_ledger_agrees(logs, build_ledger(logs)) + + @pytest.mark.usefixtures("restore_sys_exit") + def test_over_claiming_ledger_is_refused(self) -> None: + """A ledger claiming a duplicate that is not one aborts the script.""" + logs = { + "2026-08-24": [ + _entry("runnerup_verified", "2026-08-24T07:00:00+00:00"), + _entry("runnerup_verified", "2026-08-24T19:00:00+00:00"), + ], + } + hand_written = { + "2026-08-24": [ + { + "credit": "runnerup_verified", + "counted": "runnerup_verified:2026-08-24", + "duplicate_of": ["runnerup_verified:2026-08-24"], + "note": "wrong: two runs in a day are two workouts", + }, + ], + } + with pytest.raises(SystemExit, match="Ledger disagrees"): + check_ledger_agrees(logs, hand_written) diff --git a/scripts/annotate_duplicate_credits.py b/scripts/annotate_duplicate_credits.py new file mode 100755 index 0000000..0d0c5b6 --- /dev/null +++ b/scripts/annotate_duplicate_credits.py @@ -0,0 +1,250 @@ +#!/usr/bin/env python3 +"""Record which log entries are two recordings of one workout. + +Written for the 2026-08-29 outage: a StrongLifts session reaches ``log.json`` +twice -- once as ``phone_verified`` from the phone sync, once as +``pc_workout_verified`` from the PC session sync -- and both are genuine, +signed records of the same hour in the gym. ``_weekly_check`` now collapses +them onto one credit, so the *numbers* are right; this makes the *history* say +so rather than leaving a reader to notice two entries share a duration. + +Annotates, never deletes, for two independent reasons. Deleting is +self-undoing: every re-ingest path dedups against ``log.json``'s own contents, +so ``_session_sync`` re-pulls a removed ``pc_workout_verified`` from the CRDT +store on the next 15-minute ``workout-sync.timer`` tick and re-applies its +shutdown credit. And it is the repo's stated position on signed data -- see +``scripts/revoke_sick_day.py``. No entry is modified, so no HMAC is touched. + +The ledger is descriptive: nothing reads it to decide a lock. If it ever +disagrees with :func:`~screen_locker._weekly_check.count_day_credits` the +script fails rather than writing, because a ledger contradicting the rule it +documents is worse than no ledger. + +Usage: + python3 scripts/annotate_duplicate_credits.py # show the diff + python3 scripts/annotate_duplicate_credits.py --apply # write it +""" + +from __future__ import annotations + +import argparse +from collections import defaultdict +from datetime import UTC, datetime +import json +import logging +from pathlib import Path +import shutil +import sys +from typing import TypedDict + +from screen_locker._log_io import load_workout_log +from screen_locker._weekly_check import count_day_credits, credit_key + +_logger = logging.getLogger(__name__) + +REPO_ROOT = Path(__file__).resolve().parent.parent +LOG_FILE = REPO_ROOT / "screen_locker" / "log.json" +LEDGER_FILE = REPO_ROOT / "screen_locker" / "duplicate_credits.json" + +# A credit is "shared" only once a second entry claims it; one entry per +# credit is the normal case and earns no ledger record. +_SHARED_BY = 2 + +_NOTE = ( + "Two ingestion paths recorded one workout. Both entries are genuine and " + "stay in the log; they share a single credit." +) + + +class CreditRecord(TypedDict): + """One credit that more than one log entry recorded. + + Typed rather than a bare dict so ``duplicate_of`` is known to be a list + of strings at both use sites -- otherwise ``len()`` and ``join()`` on it + need a typing suppression, which this repo bans outright. + """ + + credit: str + counted: str + duplicate_of: list[str] + note: str + + +def _entry_id(entry: dict, date_str: str, index: int) -> str: + """Return a stable identifier for a log entry. + + Args: + entry: The log entry. + date_str: The date key it is filed under. + index: Its position in that date's list, used when the entry predates + ``workout_id`` and so has nothing else to name it by. + + Returns: + The entry's ``workout_id`` when it has one, else ``#``. + """ + workout_id = entry.get("workout_id") + return str(workout_id) if workout_id else f"{date_str}#{index}" + + +def _shared_credits(date_str: str, entries: list[dict]) -> list[CreditRecord]: + """Return one record per credit that more than one entry shares. + + Args: + date_str: The date key the entries are filed under. + entries: That date's log entries. + + Returns: + A list of ledger records, empty when every entry earns its own credit. + """ + groups: dict[tuple[str, str], list[tuple[int, dict]]] = defaultdict(list) + for index, entry in enumerate(entries): + key = credit_key(date_str, index, entry) + if key is not None: + groups[key].append((index, entry)) + + records: list[CreditRecord] = [] + for key, members in sorted(groups.items()): + if len(members) < _SHARED_BY: + continue + # Earliest timestamp first: the canonical entry is the one that landed + # when the workout actually happened, not the copy a later sync filed. + ordered = sorted(members, key=lambda m: str(m[1].get("timestamp", ""))) + records.append( + { + "credit": key[0], + "counted": _entry_id(ordered[0][1], date_str, ordered[0][0]), + "duplicate_of": [ + _entry_id(entry, date_str, index) for index, entry in ordered[1:] + ], + "note": _NOTE, + } + ) + return records + + +def build_ledger(logs: dict[str, list[dict]]) -> dict[str, list[CreditRecord]]: + """Build the full ledger from a loaded workout log. + + Args: + logs: The normalised ``{date: [entry, ...]}`` log. + + Returns: + A date-keyed ledger holding only the days with a shared credit. + """ + return { + date_str: records + for date_str, entries in sorted(logs.items()) + if (records := _shared_credits(date_str, entries)) + } + + +def check_ledger_agrees( + logs: dict[str, list[dict]], + ledger: dict[str, list[CreditRecord]], +) -> None: + """Exit unless the ledger's arithmetic matches the canonical credit rule. + + For every day, the entries that earn a credit minus the ones the ledger + calls duplicates must equal ``count_day_credits``. If those disagree, the + ledger is describing a rule the locker does not use. + + Args: + logs: The normalised workout log. + ledger: The ledger built from it. + """ + for date_str, entries in logs.items(): + earning = sum( + 1 + for index, entry in enumerate(entries) + if credit_key(date_str, index, entry) is not None + ) + duplicates = sum( + len(record["duplicate_of"]) for record in ledger.get(date_str, []) + ) + expected = count_day_credits(date_str, entries) + if earning - duplicates != expected: + sys.exit( + f"Ledger disagrees with count_day_credits on {date_str}: " + f"{earning} earning - {duplicates} duplicate != {expected}. " + "Refusing to write a ledger that contradicts the credit rule." + ) + + +def _describe(ledger: dict[str, list[CreditRecord]]) -> str: + """Render the ledger as a human-readable summary. + + Args: + ledger: The ledger to describe. + + Returns: + One line per shared credit, or a single line saying there are none. + """ + if not ledger: + return "No shared credits found — every entry records its own workout." + lines = [f"{sum(len(v) for v in ledger.values())} shared credit(s):"] + lines.extend( + f" {date_str} {record['counted']} absorbs " + f"{', '.join(record['duplicate_of'])}" + for date_str, records in ledger.items() + for record in records + ) + return "\n".join(lines) + + +def _write(ledger: dict[str, list[CreditRecord]]) -> None: + """Back up any existing ledger and write the new one. + + Args: + ledger: The ledger to persist. + """ + if LEDGER_FILE.exists(): + stamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%SZ") + backup = LEDGER_FILE.with_suffix(f".json.bak-{stamp}") + shutil.copy2(LEDGER_FILE, backup) + _logger.info("Backed up existing ledger to %s", backup) + # indent=2 matches write_signed_entry, so both files diff the same way. + LEDGER_FILE.write_text(json.dumps(ledger, indent=2) + "\n", encoding="utf-8") + _logger.info("Wrote %s", LEDGER_FILE) + + +def main() -> int: + """Show, and optionally write, the duplicate-credit ledger. + + Returns: + Process exit code. + """ + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--apply", action="store_true", help="write the ledger (default: dry run)" + ) + args = parser.parse_args() + + handler = logging.StreamHandler(sys.stdout) + handler.setFormatter(logging.Formatter("%(message)s")) + logging.basicConfig(level=logging.INFO, handlers=[handler]) + + logs = load_workout_log(LOG_FILE) + if not logs: + sys.exit(f"No workout log to annotate at {LOG_FILE}") + + ledger = build_ledger(logs) + check_ledger_agrees(logs, ledger) + _logger.info("%s", _describe(ledger)) + + existing = ( + json.loads(LEDGER_FILE.read_text(encoding="utf-8")) + if LEDGER_FILE.exists() + else None + ) + if existing == ledger: + _logger.info("Ledger is already up to date; nothing to do.") + return 0 + if not args.apply: + _logger.info("Dry run — re-run with --apply to write %s", LEDGER_FILE) + return 0 + _write(ledger) + return 0 + + +if __name__ == "__main__": + sys.exit(main())