Credit StrongLifts sessions from the headless sync, as pc_workout_verified

`sync_now()` -- what workout-sync.timer runs every 15 minutes -- only ever
ingested MANUAL records. The session reader `pull_synced_workout()` was
reachable only from `_verify_phone_workout`, i.e. the interactive lock/status
UI, so a finished session earned nothing until a lock screen happened to open.

That was survivable while the phone was the only device recording sessions:
the same lock screen that demanded the workout also verified it. The Linux
build finishes sessions with no lock screen involved at all, so the credit
would simply never arrive.

`pull_all_session_records()` is the session counterpart to the existing
`pull_all_manual_records()` -- `pull_synced_workout` returns only the single
newest session for the live unlock check, which cannot backfill a week.

New type `pc_workout_verified` rather than reusing `phone_verified`: it is
machine-checked in exactly the same way (same WorkoutSession payload, same
set/rep and duration validation), but there is no phone involved, and writing
"phone_verified" into an HMAC-signed log would make the log lie about its own
provenance.

Ingestion re-derives the verdict instead of trusting the record's `succeeded`
flag -- the record arrives over the network and this is the thing handing out
unlock credit. Every rejection logs at WARNING with an actionable reason, per
the repo's no-silent-failures rule.

One real leak fixed along the way: formatting the duration with `.0f` rounded
34m59s up to "35 min", which is precisely the hidden accept bar -- the message
would have told the user the real threshold while rejecting them at it. It now
truncates, so the advertised 40 stays the only number shown.

Verified against live data: ingests the 4 real synced sessions, and a second
run ingests 0 (the write chokepoint dedups by workout_id), so the repeating
timer cannot double-credit. Suite green at 100% branch coverage, 1046 tests.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DfJC1Wu1SEgJ6X49k5Jqtg

screen_locker/_session_sync.py

@@ -0,0 +1,130 @@
+"""Ingesting synced StrongLifts sessions into ``log.json``.
+
+The manual-workout half of sync (``_manual_sync``) has always run headlessly
+from ``workout-sync.timer``. Sessions did not: ``pull_synced_workout`` was
+reachable only from the interactive verification flow, so a finished session
+sat in Firebase uncredited until the locker next opened a window. That was
+survivable while the phone was the only device recording sessions -- the same
+lock screen that asked for the workout also verified it -- but the Linux build
+finishes sessions with no lock screen involved at all.
+
+Mirrors :mod:`screen_locker._manual_sync`: reconstruct, re-validate on the PC,
+then HMAC-sign and append under the session's own date. Never trusts the
+record's own claim that it succeeded.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from screen_locker._constants import (
+    MIN_WORKOUT_DURATION_MINUTES,
+    WORKOUT_DURATION_ACCEPT_MINUTES,
+)
+from screen_locker._log_mixin import write_signed_entry
+from screen_locker._weekly_check import PC_WORKOUT_TYPE
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Iterable, Mapping
+    from pathlib import Path
+
+    from screen_locker._log_mixin import RecordResult
+
+    OnIngestedCallback = Callable[[dict, "list[dict]"], None]
+
+_logger = logging.getLogger(__name__)
+
+
+def validate_session(payload: Mapping[str, object]) -> tuple[bool, str]:
+    """Check a synced session payload is a real, long-enough workout.
+
+    Deliberately re-runs the same content checks
+    ``PhoneVerificationMixin._validate_json_content`` applies, rather than
+    trusting ``succeeded``: a record arrives over the network and the PC is
+    the thing handing out unlock credit, so it re-derives the verdict.
+
+    Returns ``(ok, reason)``; ``reason`` always reads like a sentence so a
+    rejected session says why instead of silently not counting.
+    """
+    exercises = payload.get("exercises")
+    if not isinstance(exercises, list) or not exercises:
+        return False, "no exercises in the session payload"
+    raw_duration = payload.get("duration_seconds", 0)
+    if not isinstance(raw_duration, (int, float)):
+        return False, f"duration_seconds is not a number: {raw_duration!r}"
+    duration_min = raw_duration / 60.0
+    # Accept bar carries the hidden leeway; the message advertises the round
+    # number only (see _constants.py).
+    if duration_min < WORKOUT_DURATION_ACCEPT_MINUTES:
+        # Truncate rather than round: 34m59s rounds UP to "35 min", which is
+        # exactly the hidden accept bar and would tell the user the real
+        # threshold while rejecting them at it. The advertised number stays
+        # MIN_WORKOUT_DURATION_MINUTES (see _constants.py).
+        return False, (
+            f"only {int(duration_min)} min logged, "
+            f"need at least {MIN_WORKOUT_DURATION_MINUTES} min"
+        )
+    return True, f"{duration_min:.0f} min"
+
+
+def build_session_entry(payload: Mapping[str, object], detail: str) -> dict[str, str]:
+    """Build the ``workout_data`` for a verified synced session."""
+    flag = "all succeeded" if payload.get("succeeded") else "partial"
+    workout_type = payload.get("workout_type", "?")
+    return {
+        "type": PC_WORKOUT_TYPE,
+        "source": f"StrongLifts workout {workout_type} ({detail}, {flag})",
+        "duration_minutes": f"{float(payload.get('duration_seconds', 0)) / 60.0:.1f}",
+        "sync_record_id": str(payload.get("start_time", "")),
+    }
+
+
+def ingest_session_records(
+    log_file: Path,
+    records: Iterable[tuple[str, Mapping[str, object]]],
+    *,
+    on_ingested: OnIngestedCallback | None = None,
+) -> list[str]:
+    """Ingest synced StrongLifts sessions into ``log.json``.
+
+    For each ``(record_id, payload)``: re-validate on the PC, then HMAC-sign
+    and APPEND under the session's own ``date`` so the weekly count places it
+    in the right ISO week. Idempotent -- the write chokepoint dedups by
+    ``workout_id`` -- so re-syncing the same session every 15 minutes is a
+    no-op rather than repeated credit.
+
+    ``on_ingested(entry, prior_entries)`` fires for each newly-appended
+    session so the caller applies the identical live-workout reward.
+
+    Returns the record ids actually ingested.
+    """
+    ingested: list[str] = []
+    for record_id, payload in records:
+        date = payload.get("date")
+        if not isinstance(date, str) or not date:
+            _logger.warning(
+                "Skipping synced session %s: no usable date field (%r), so it "
+                "cannot be filed in a week and will never count.",
+                record_id,
+                date,
+            )
+            continue
+        ok, detail = validate_session(payload)
+        if not ok:
+            _logger.warning(
+                "Skipping synced session %s dated %s: %s — it does NOT count "
+                "toward the weekly minimum.",
+                record_id,
+                date,
+                detail,
+            )
+            continue
+        entry = build_session_entry(payload, detail)
+        result: RecordResult = write_signed_entry(log_file, date, entry)
+        if not result.appended:
+            continue
+        ingested.append(record_id)
+        if on_ingested is not None:
+            on_ingested(entry, result.prior_entries)
+    return ingested

screen_locker/_sync_mixin.py

@@ -14,11 +14,15 @@ from screen_locker._constants import EXTRA_BENEFITS_FILE
 from screen_locker._extra_benefits import weekly_shutdown_bonus_hours
 from screen_locker._manual_push import push_pc_workouts
 from screen_locker._manual_sync import ingest_manual_records
+from screen_locker._session_sync import ingest_session_records
 from screen_locker._weekly_check import (
     WEEKLY_WORKOUT_MINIMUM,
     count_weekly_workouts,
 )
-from screen_locker._workout_sync import pull_all_manual_records
+from screen_locker._workout_sync import (
+    pull_all_manual_records,
+    pull_all_session_records,
+)
 
 __all__ = ["SyncMixin"]
 
@@ -43,6 +47,7 @@ class SyncMixin:
         Repeating it every 15 minutes closes that gap.
         """
         self._ingest_synced_manual_workouts()
+        self._ingest_synced_sessions()
         self._auto_fill_week_runnerup_bonus()
 
     def _ingest_synced_manual_workouts(self) -> None:
@@ -64,6 +69,27 @@ class SyncMixin:
             _logger.info("Ingested synced manual workout: %s", record_id)
         self.workout_data = {}
 
+    def _ingest_synced_sessions(self) -> None:
+        """Ingest StrongLifts sessions other devices (or this PC) finished.
+
+        The desktop build of workout_app writes a session to its own device
+        log and nothing else; without this the record would only ever be seen
+        by the interactive verification flow, so a workout finished on the PC
+        earned nothing until a lock screen happened to open.
+
+        Each newly-ingested session earns the identical reward a live-logged
+        workout would, and re-running is a no-op (the write chokepoint dedups
+        by workout_id).
+        """
+        ingested = ingest_session_records(
+            self.log_file,
+            pull_all_session_records(),
+            on_ingested=self._credit_ingested_manual_workout,
+        )
+        for record_id in ingested:
+            _logger.info("Ingested synced StrongLifts session: %s", record_id)
+        self.workout_data = {}
+
     def _credit_ingested_manual_workout(
         self, entry: dict[str, str], prior_entries: list[dict]
     ) -> None:

screen_locker/_weekly_check.py

@@ -27,11 +27,19 @@ WEEKLY_WORKOUT_MINIMUM: int = 5
 # Python weekday(): Mon=0, Tue=1, Wed=2, Thu=3, Fri=4, Sat=5, Sun=6
 _RELAXED_WEEKDAYS: frozenset[int] = frozenset({1, 2, 3})  # Tue, Wed, Thu
 
+# A StrongLifts session performed on the PC itself (the Linux build of
+# workout_app). Machine-checked exactly like ``phone_verified`` -- same
+# WorkoutSession payload, same set/rep validation -- but named separately
+# because there is no phone involved: writing "phone_verified" into an
+# HMAC-signed log for a workout done on the desktop would make the log lie
+# about its own provenance.
+PC_WORKOUT_TYPE: str = "pc_workout_verified"
+
 # VERIFIED workouts are machine-checked (a real StrongLifts session or a real
 # RunnerUp run). They count toward the weekly total INDIVIDUALLY — multiple
 # per day are allowed — because they can't be faked.
 VERIFIED_WORKOUT_TYPES: frozenset[str] = frozenset(
-    {"phone_verified", "runnerup_verified"},
+    {"phone_verified", "runnerup_verified", PC_WORKOUT_TYPE},
 )
 # The single self-reported type. It counts toward the weekly total
 # INDIVIDUALLY, same as a verified workout — the manual-workout rate budget

screen_locker/_workout_credit.py

@@ -14,7 +14,11 @@ import logging
 from typing import TYPE_CHECKING
 
 from screen_locker import _sick_tracker
-from screen_locker._weekly_check import COUNTED_WORKOUT_TYPES, count_weekly_workouts
+from screen_locker._weekly_check import (
+    COUNTED_WORKOUT_TYPES,
+    PC_WORKOUT_TYPE,
+    count_weekly_workouts,
+)
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -62,6 +66,7 @@ class WorkoutCreditMixin:
         if self.workout_data.get("type") not in (
             "phone_verified",
             "runnerup_verified",
+            PC_WORKOUT_TYPE,
             "manual_workout",
         ):
             return None

screen_locker/_workout_sync.py

@@ -52,6 +52,7 @@ __all__ = [
     "_records_matching",
     "_session_records",
     "pull_all_manual_records",
+    "pull_all_session_records",
     "pull_synced_workout",
     "read_sync_token",
     "remote_client",
@@ -198,6 +199,30 @@ def pull_synced_workout() -> tuple[dict | None, str | None]:
     return payload, None
 
 
+def pull_all_session_records() -> list[tuple[str, dict]]:
+    """Return all synced StrongLifts sessions across every device log.
+
+    The session counterpart to :func:`pull_all_manual_records`, and the reason
+    a workout finished on the PC reaches ``log.json`` without a lock screen
+    being open: ``pull_synced_workout`` returns only the single newest session
+    for the live unlock check, which cannot backfill a week.
+
+    Best-effort in the same way -- an unconfigured token or a corrupt device
+    log yields fewer records rather than raising.
+    """
+    client = sync_client()
+    if client is None:
+        _logger.warning(
+            "No sync client (neither a GitHub token nor a Firebase config) — "
+            "synced StrongLifts sessions cannot be pulled, so a workout done "
+            "on another device will NOT count toward the weekly minimum.",
+        )
+        return []
+
+    merged = _merge_device_records(client, _session_records, "sessions")
+    return [(rid, payload) for rid, (payload, _hlc) in merged.items()]
+
+
 def pull_all_manual_records() -> list[tuple[str, dict]]:
     """Return all synced manual-workout records across every device log.
 

screen_locker/tests/test_screen_lock_coverage_part5.py

@@ -131,3 +131,68 @@ class TestIngestSyncedManualWorkouts:
             locker._credit_ingested_manual_workout(entry, prior)
         assert locker.workout_data == entry
         apply_credit.assert_called_once_with(prior)
+
+
+class TestIngestSyncedSessions:
+    """Tests for _ingest_synced_sessions (StrongLifts-session wiring).
+
+    The headless ``--sync-only`` path had no session ingest at all, so a
+    workout finished on the PC earned nothing until a lock screen opened.
+    """
+
+    def test_logs_each_ingested_session(
+        self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
+    ) -> None:
+        """Logs each ingested session."""
+        locker = create_locker(mock_tk, tmp_path)
+        with (
+            patch(
+                "screen_locker._sync_mixin.pull_all_session_records",
+                return_value=[("2026-08-21T10:39:24", {})],
+            ),
+            patch(
+                "screen_locker._sync_mixin.ingest_session_records",
+                return_value=["2026-08-21T10:39:24"],
+            ) as ingest,
+        ):
+            locker._ingest_synced_sessions()
+        ingest.assert_called_once()
+
+    def test_no_sessions_ingests_nothing(
+        self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
+    ) -> None:
+        """No sessions ingests nothing."""
+        locker = create_locker(mock_tk, tmp_path)
+        with (
+            patch(
+                "screen_locker._sync_mixin.pull_all_session_records",
+                return_value=[],
+            ),
+            patch(
+                "screen_locker._sync_mixin.ingest_session_records",
+                return_value=[],
+            ),
+        ):
+            locker._ingest_synced_sessions()
+
+    def test_credits_a_session_like_a_live_workout(
+        self, mock_tk: MagicMock, mock_sys_exit: MagicMock, tmp_path: Path
+    ) -> None:
+        """A synced session earns the same reward a live workout would."""
+        locker = create_locker(mock_tk, tmp_path)
+        with (
+            patch(
+                "screen_locker._sync_mixin.pull_all_session_records",
+                return_value=[],
+            ),
+            patch(
+                "screen_locker._sync_mixin.ingest_session_records",
+                return_value=[],
+            ) as ingest,
+        ):
+            locker._ingest_synced_sessions()
+        assert (
+            ingest.call_args.kwargs["on_ingested"]
+            == locker._credit_ingested_manual_workout
+        )
+        assert locker.workout_data == {}

screen_locker/tests/test_session_sync.py

@@ -0,0 +1,177 @@
+"""Tests for ingesting synced StrongLifts sessions into log.json."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+from screen_locker._session_sync import (
+    build_session_entry,
+    ingest_session_records,
+    validate_session,
+)
+from screen_locker._weekly_check import PC_WORKOUT_TYPE, count_weekly_workouts
+from screen_locker._workout_sync import pull_all_session_records
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+_DATE = "2026-08-21"
+# The ISO week _DATE falls in, for count_weekly_workouts.
+_TODAY = datetime(2026, 8, 21, tzinfo=timezone.utc)
+
+
+def _session(**overrides: object) -> dict:
+    """A session payload shaped like the app's WorkoutSession.toJson()."""
+    payload: dict = {
+        "workout_type": "B",
+        "date": _DATE,
+        "start_time": f"{_DATE}T10:39:24.531831",
+        "duration_seconds": 6474,
+        "succeeded": True,
+        "exercises": [{"name": "Dumbbell Row", "succeeded": True}],
+    }
+    payload.update(overrides)
+    return payload
+
+
+class TestValidateSession:
+    """The PC re-derives the verdict rather than trusting the record."""
+
+    def test_accepts_a_long_enough_session(self) -> None:
+        ok, detail = validate_session(_session())
+        assert ok
+        assert "108 min" in detail
+
+    def test_rejects_a_session_with_no_exercises(self) -> None:
+        ok, detail = validate_session(_session(exercises=[]))
+        assert not ok
+        assert "no exercises" in detail
+
+    def test_rejects_a_non_list_exercises_field(self) -> None:
+        ok, detail = validate_session(_session(exercises="Dumbbell Row"))
+        assert not ok
+        assert "no exercises" in detail
+
+    def test_rejects_a_short_session(self) -> None:
+        ok, detail = validate_session(_session(duration_seconds=10 * 60))
+        assert not ok
+        assert "10 min" in detail
+
+    def test_rejects_a_non_numeric_duration(self) -> None:
+        ok, detail = validate_session(_session(duration_seconds="lots"))
+        assert not ok
+        assert "not a number" in detail
+
+    def test_accepts_exactly_the_hidden_accept_threshold(self) -> None:
+        """35 min passes; the message still advertises the round 40."""
+        ok, _ = validate_session(_session(duration_seconds=35 * 60))
+        assert ok
+
+    def test_rejects_just_below_the_accept_threshold(self) -> None:
+        ok, detail = validate_session(_session(duration_seconds=34 * 60 + 59))
+        assert not ok
+        # The real bar (35) must never leak into a user-visible string.
+        assert "35" not in detail
+        assert "40 min" in detail
+
+
+class TestBuildSessionEntry:
+    """The entry names the PC as the source, never the phone."""
+
+    def test_marks_the_workout_as_pc_verified(self) -> None:
+        entry = build_session_entry(_session(), "108 min")
+        assert entry["type"] == PC_WORKOUT_TYPE
+        assert entry["type"] != "phone_verified"
+
+    def test_describes_the_workout_and_its_outcome(self) -> None:
+        entry = build_session_entry(_session(), "108 min")
+        assert "workout B" in entry["source"]
+        assert "all succeeded" in entry["source"]
+
+    def test_reports_a_partial_workout_as_partial(self) -> None:
+        entry = build_session_entry(_session(succeeded=False), "107 min")
+        assert "partial" in entry["source"]
+
+    def test_carries_the_duration_and_record_id(self) -> None:
+        entry = build_session_entry(_session(), "108 min")
+        assert entry["duration_minutes"] == "107.9"
+        assert entry["sync_record_id"] == f"{_DATE}T10:39:24.531831"
+
+
+class TestIngestSessionRecords:
+    """Ingestion is idempotent and credits only real workouts."""
+
+    def test_appends_a_verified_session_under_its_own_date(
+        self, tmp_path: Path
+    ) -> None:
+        log = tmp_path / "log.json"
+        ingested = ingest_session_records(log, [("r1", _session())])
+        assert ingested == ["r1"]
+        assert count_weekly_workouts(log, today=_TODAY) == 1
+
+    def test_is_idempotent_across_repeated_syncs(self, tmp_path: Path) -> None:
+        """The 15-minute timer re-sees every record; it must not re-credit."""
+        log = tmp_path / "log.json"
+        records = [("r1", _session())]
+        assert ingest_session_records(log, records) == ["r1"]
+        assert ingest_session_records(log, records) == []
+        assert count_weekly_workouts(log, today=_TODAY) == 1
+
+    def test_skips_a_session_with_no_date(self, tmp_path: Path) -> None:
+        payload = _session()
+        del payload["date"]
+        assert ingest_session_records(tmp_path / "log.json", [("r1", payload)]) == []
+
+    def test_skips_a_session_whose_date_is_not_a_string(self, tmp_path: Path) -> None:
+        assert (
+            ingest_session_records(tmp_path / "log.json", [("r1", _session(date=7))])
+            == []
+        )
+
+    def test_skips_a_session_that_fails_validation(self, tmp_path: Path) -> None:
+        log = tmp_path / "log.json"
+        short = _session(duration_seconds=60)
+        assert ingest_session_records(log, [("r1", short)]) == []
+        assert count_weekly_workouts(log, today=_TODAY) == 0
+
+    def test_calls_back_for_each_newly_ingested_session(self, tmp_path: Path) -> None:
+        seen: list[tuple[dict, list]] = []
+        ingest_session_records(
+            tmp_path / "log.json",
+            [("r1", _session())],
+            on_ingested=lambda entry, prior: seen.append((entry, prior)),
+        )
+        assert len(seen) == 1
+        assert seen[0][0]["type"] == PC_WORKOUT_TYPE
+        assert seen[0][1] == []
+
+    def test_does_not_call_back_for_a_duplicate(self, tmp_path: Path) -> None:
+        log = tmp_path / "log.json"
+        records = [("r1", _session())]
+        ingest_session_records(log, records)
+        seen: list[dict] = []
+        ingest_session_records(
+            log, records, on_ingested=lambda entry, _prior: seen.append(entry)
+        )
+        assert seen == []
+
+
+class TestPullAllSessionRecords:
+    """Sessions come from every device log, or from none at all."""
+
+    def test_returns_nothing_without_a_sync_client(self) -> None:
+        with patch("screen_locker._workout_sync.sync_client", return_value=None):
+            assert pull_all_session_records() == []
+
+    def test_returns_the_merged_records(self) -> None:
+        merged = {"r1": (_session(), object())}
+        with (
+            patch("screen_locker._workout_sync.sync_client", return_value=object()),
+            patch(
+                "screen_locker._workout_sync._merge_device_records",
+                return_value=merged,
+            ),
+        ):
+            assert pull_all_session_records() == [("r1", _session())]