Stream credits phone manual workouts instantly

The Firebase stream only ingested StrongLifts sessions, so a manual
workout logged on the phone waited for the 15-min workout-sync tick.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWVTJPDdbeF71W9Wd3q

screen_locker/_poke_credit.py

@@ -30,6 +30,7 @@ from screen_locker._day import today_str
 from screen_locker._earned import hhmm
 from screen_locker._extra_benefits import process_week_transition
 from screen_locker._grace_floor import tonight_minutes
+from screen_locker._manual_sync import ingest_manual_records
 from screen_locker._session_sync import ingest_session_records, validate_session
 from screen_locker._shutdown import read_shutdown_config
 from screen_locker._shutdown_base import reset_to_base_if_new_day
@@ -45,6 +46,7 @@ if TYPE_CHECKING:
 __all__ = [
     "BatchOutcome",
     "PokeOutcome",
+    "credit_manual_workouts",
     "credit_session",
     "credit_sessions",
     "current_gaming_minutes",
@@ -201,11 +203,40 @@ def credit_sessions(
         results.append(locker.credit_written(entry, prior_entries))
 
     ingested = ingest_session_records(locker.log_file, records, on_ingested=on_ingested)
+    return _outcome(ingested, results, "session")
+
+
+def credit_manual_workouts(
+    records: Iterable[tuple[str, Mapping[str, object]]],
+    *,
+    today: str | None = None,
+) -> BatchOutcome:
+    """``credit_sessions`` for manual workouts, via ``ingest_manual_records``."""
+    locker = fresh_locker()
+    locker.start_the_day()
+    results: list[WorkoutCreditResult] = []
+
+    def on_ingested(
+        entry: dict[str, str], prior_entries: list[dict[str, object]]
+    ) -> None:
+        results.append(locker.credit_written(entry, prior_entries))
+
+    ingested = ingest_manual_records(
+        locker.log_file, records, today=today, on_ingested=on_ingested
+    )
+    return _outcome(ingested, results, "manual workout")
+
+
+def _outcome(
+    ingested: list[str], results: list[WorkoutCreditResult], what: str
+) -> BatchOutcome:
+    """Count what earned time; warn about logged copies that earned none."""
     credited = sum(not result.already_counted_today for result in results)
     if len(results) > credited:
         _logger.warning(
-            "%d streamed session(s) were logged but earned no time: today's slot "
+            "%d streamed %s(s) were logged but earned no time: today's slot "
             "was already paid by another copy of the same workout",
             len(results) - credited,
+            what,
         )
     return BatchOutcome(ingested=tuple(ingested), credited=credited)

screen_locker/_stream_cache.py

@@ -8,7 +8,8 @@ tombstones ``_manual_push`` re-publishes every tick, and dropping it would
 let a retracted session be credited again -- but an event that touched only
 this PC's log never triggers a pass.
 
-Crediting is :func:`screen_locker._poke_credit.credit_sessions` under the
+Crediting is :func:`screen_locker._poke_credit.credit_sessions` (and
+``credit_manual_workouts`` for phone-logged manual workouts) under the
 same in-process lock as the LAN poke handler. ``write_signed_entry`` already
 dedups by workout id; the ``seen`` set here only spares the file I/O for
 records this daemon has already handed to the ingest.
@@ -23,11 +24,19 @@ from typing import TYPE_CHECKING, Final
 
 from screen_locker import _poke_credit
 from screen_locker._rtdb_stream import LOG_LEAF, device_updates
-from screen_locker._sync_records import _session_records, merge_device_texts
+from screen_locker._sync_records import (
+    _manual_records,
+    _session_records,
+    merge_device_texts,
+)
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
     import threading
 
+    from crdt_sync import Hlc
+
+    from screen_locker._poke_credit import BatchOutcome
     from screen_locker._rtdb_stream import SseEvent
 
 __all__ = ["DeviceLogCache", "EventResult"]
@@ -78,15 +87,17 @@ class DeviceLogCache:
                 self._texts[dev] = text
         return changed
 
-    def _pending(self) -> list[tuple[str, dict[str, object], str]]:
-        """Union records this daemon has not yet handed to the ingest."""
+    def _pending(
+        self, extract: Callable[[str], dict[str, tuple[dict, Hlc]]], what: str
+    ) -> list[tuple[str, dict[str, object], str]]:
+        """Union ``extract`` records this daemon has not yet handed to the ingest."""
         union = merge_device_texts(
             (
                 (f"{self._prefix}/{dev}/{LOG_LEAF}", text)
                 for dev, text in sorted(self._texts.items())
             ),
-            _session_records,
-            "sessions",
+            extract,
+            what,
         )
         return [
             (rid, payload, str(hlc))
@@ -115,36 +126,59 @@ class DeviceLogCache:
                 snapshot=False,
                 summary=f"only this PC's own log changed ({devices}); no credit",
             )
-        pending = self._pending()
-        if not pending:
+        sessions = self._pending(_session_records, "sessions")
+        manuals = self._pending(_manual_records, "manual workouts")
+        if not sessions and not manuals:
             return EventResult(
                 snapshot=full,
                 summary=f"{len(self._texts)} device log(s) cached, changed: {devices}; "
-                f"every session already handed to the ingest -- no-op, no file I/O "
-                f"({_ms(started)} ms)",
+                f"every session and manual workout already handed to the ingest "
+                f"-- no-op, no file I/O ({_ms(started)} ms)",
             )
         try:
             with self._gate:
-                outcome = _poke_credit.credit_sessions(
-                    [(rid, payload) for rid, payload, _hlc in pending]
+                done = [
+                    _credit(pending, credit, what)
+                    for pending, credit, what in (
+                        (sessions, _poke_credit.credit_sessions, "session"),
+                        (
+                            manuals,
+                            _poke_credit.credit_manual_workouts,
+                            "manual workout",
+                        ),
                     )
+                    if pending
+                ]
         except Exception:  # pylint: disable=broad-exception-caught  # one bad event must not drop the stream
             # Not a stream problem: keep the connection (a reconnect would
             # re-download the snapshot and fail the same way). Nothing is
             # marked seen, so the next event retries.
             _logger.exception(
-                "Crediting %d streamed session(s) failed; the next stream event "
-                "retries, and the 15-min workout-sync still covers them",
-                len(pending),
+                "Crediting %d streamed session(s) and %d manual workout(s) failed; "
+                "the next stream event retries, and the 15-min workout-sync still "
+                "covers them",
+                len(sessions),
+                len(manuals),
             )
             return EventResult(snapshot=full, summary="crediting failed (see above)")
-        self._seen.update((rid, hlc) for rid, _payload, hlc in pending)
+        self._seen.update((rid, hlc) for rid, _payload, hlc in [*sessions, *manuals])
         return EventResult(
             snapshot=full,
             summary=f"{len(self._texts)} device log(s) cached, changed: {devices}; "
-            f"{len(pending)} session(s) checked, {len(outcome.ingested)} newly "
+            f"{'; '.join(done)} ({_ms(started)} ms)",
+        )
+
+
+def _credit(
+    pending: list[tuple[str, dict[str, object], str]],
+    credit: Callable[[list[tuple[str, dict[str, object]]]], BatchOutcome],
+    what: str,
+) -> str:
+    """Run one batch credit step; describe what it did."""
+    outcome = credit([(rid, payload) for rid, payload, _hlc in pending])
+    return (
+        f"{len(pending)} {what}(s) checked, {len(outcome.ingested)} newly "
         f"logged {list(outcome.ingested)}, {outcome.credited} credited"
-            f"({_ms(started)} ms)",
     )
 
 

screen_locker/tests/test_stream_cache.py

@@ -154,7 +154,9 @@ class TestFailures:
         failures = [r for r in caplog.records if r.levelno >= logging.ERROR]
         assert failures
         assert failures[0].exc_info is not None
-        assert "Crediting 1 streamed session(s) failed" in failures[0].getMessage()
+        assert (
+            "Crediting 1 streamed session(s) and 0 manual" in failures[0].getMessage()
+        )
         credit.error = None
         cache.handle(snapshot({"phone": log_blob("s1")}))
         assert credit.batches == [["s1"]]

screen_locker/tests/test_stream_cache_manual.py

@@ -0,0 +1,148 @@
+"""The Firebase stream credits phone-logged manual workouts on arrival.
+
+Until 2026-11-02 the stream only handed StrongLifts sessions to the ingest, so
+a manual workout logged on the phone sat in Firebase until the next 15-minute
+``workout-sync`` tick -- long enough that it looked lost and was re-logged by
+hand on the PC.
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+from typing import TYPE_CHECKING
+from unittest.mock import MagicMock
+
+import pytest
+
+from screen_locker import _poke_credit, _stream_cache
+from screen_locker._poke_credit import BatchOutcome, PokeLocker
+from screen_locker._stream_cache import DeviceLogCache
+from screen_locker._workout_credit import WorkoutCreditResult
+from screen_locker.tests.test_manual_sync import _OTHER_DRAFT, _manual_record
+from screen_locker.tests.test_stream_cache import (
+    OWN,
+    Credit,
+    log_blob,
+    patch_event,
+    snapshot,
+)
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+_DATE = "2026-07-13"
+_RID, _PAYLOAD = _manual_record(_OTHER_DRAFT, _DATE)
+
+
+@pytest.fixture(name="steps")
+def steps_fixture(monkeypatch: pytest.MonkeyPatch) -> tuple[Credit, Credit]:
+    """Record both batch credit steps instead of building a locker."""
+    sessions, manuals = Credit(), Credit()
+    monkeypatch.setattr(_stream_cache._poke_credit, "credit_sessions", sessions)
+    monkeypatch.setattr(_stream_cache._poke_credit, "credit_manual_workouts", manuals)
+    return sessions, manuals
+
+
+@pytest.fixture(name="cache")
+def cache_fixture(steps: tuple[Credit, Credit]) -> DeviceLogCache:
+    """A cache whose own device dir is ``pc-own``, sharing one gate."""
+    gate = threading.Lock()
+    for credit in steps:
+        credit.gate = gate
+    return DeviceLogCache("sync/devices", frozenset({OWN}), gate)
+
+
+class TestStreamedManual:
+    """Which streamed manual records reach the manual ingest."""
+
+    def test_a_phone_manual_workout_is_credited_on_arrival(
+        self, cache: DeviceLogCache, steps: tuple[Credit, Credit]
+    ) -> None:
+        """A patch carrying a new manual record is handed over under the gate."""
+        sessions, manuals = steps
+        cache.handle(snapshot({"phone": log_blob("s1")}))
+        result = cache.handle(
+            patch_event({"phone": log_blob(_RID, wall=2, payload=_PAYLOAD)})
+        )
+        assert manuals.batches == [[_RID]]
+        assert manuals.gate_held == [True]
+        assert sessions.batches == [["s1"]]
+        assert "1 manual workout(s) checked, 1 newly logged" in result.summary
+
+    def test_already_handed_manual_records_are_a_no_op(
+        self, cache: DeviceLogCache, steps: tuple[Credit, Credit]
+    ) -> None:
+        """Re-streaming the same manual record costs no second ingest."""
+        _sessions, manuals = steps
+        blob = log_blob(_RID, payload=_PAYLOAD)
+        cache.handle(snapshot({"phone": blob}))
+        result = cache.handle(snapshot({"phone": blob}))
+        assert manuals.batches == [[_RID]]
+        assert "no-op, no file I/O" in result.summary
+
+    def test_tombstoned_manual_records_are_not_handed_over(
+        self, cache: DeviceLogCache, steps: tuple[Credit, Credit]
+    ) -> None:
+        """A deleted record must not be credited by the stream either."""
+        _sessions, manuals = steps
+        cache.handle(
+            snapshot({"phone": log_blob(_RID, deleted=True, payload=_PAYLOAD)})
+        )
+        assert manuals.batches == []
+
+    def test_a_session_only_event_does_not_call_the_manual_ingest(
+        self, cache: DeviceLogCache, steps: tuple[Credit, Credit]
+    ) -> None:
+        """Each kind goes only to its own ingest."""
+        sessions, manuals = steps
+        cache.handle(snapshot({"phone": log_blob("s1")}))
+        assert sessions.batches == [["s1"]]
+        assert manuals.batches == []
+
+
+@pytest.fixture(name="log_file")
+def log_file_fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """A fresh PokeLocker on a tmp log; day reset, local ledger and credit stubbed."""
+    log = tmp_path / "poke_log.json"
+
+    def headless(locker_cls: type[PokeLocker]) -> PokeLocker:
+        """Build a PokeLocker without a display, pointed at the temporary log."""
+        locker = object.__new__(locker_cls)
+        locker.log_file = log
+        locker.workout_data = {}
+        return locker
+
+    monkeypatch.setattr(_poke_credit, "_headless_locker", headless)
+    monkeypatch.setattr(PokeLocker, "start_the_day", lambda _self: None)
+    monkeypatch.setattr(
+        PokeLocker,
+        "_apply_credit_for_written_entry",
+        MagicMock(
+            return_value=WorkoutCreditResult(
+                shutdown_adjusted=True,
+                extra_bonus_delta=0,
+                weekly_count=1,
+                already_counted_today=False,
+            )
+        ),
+    )
+    return log
+
+
+class TestCreditManualWorkouts:
+    """The real manual ingest, run the way the stream runs it."""
+
+    def test_logs_and_credits_a_streamed_manual_workout(self, log_file: Path) -> None:
+        """The record lands in log.json under its own id and earns its credit."""
+        outcome = _poke_credit.credit_manual_workouts([(_RID, _PAYLOAD)], today=_DATE)
+        assert outcome == BatchOutcome(ingested=(_RID,), credited=1)
+        logged = json.loads(log_file.read_text())[_DATE]
+        assert [e["workout_id"] for e in logged] == [_RID]
+
+    def test_a_second_pass_is_a_duplicate(self, log_file: Path) -> None:
+        """Re-crediting the same record logs nothing new."""
+        _poke_credit.credit_manual_workouts([(_RID, _PAYLOAD)], today=_DATE)
+        outcome = _poke_credit.credit_manual_workouts([(_RID, _PAYLOAD)], today=_DATE)
+        assert outcome == BatchOutcome(ingested=(), credited=0)
+        assert len(json.loads(log_file.read_text())[_DATE]) == 1