focus_project: minutes, ledger, harvester

Fail-closed: a failing source logs and adds 0 minutes;
a bad key, ledger or registry exits 1.

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

focus_project/focus_project/harvest.py

@@ -0,0 +1,62 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""The harvester: turn credited focus minutes into signed ledger rows.
+
+Run by ``focus-project-harvest.service`` (``.path`` on activity.jsonl plus a
+one-minute ``.timer``). Harvests yesterday too, so work recorded just before
+midnight is paid once the next run sees it. Failure modes, by path:
+
+* a source that errors: logged, contributes 0, the run continues (fail closed);
+* an unreadable registry, signing key or ledger: logged, exit status 1, so
+  the unit shows ``failed`` instead of silently crediting nothing.
+"""
+
+from __future__ import annotations
+
+from datetime import date, timedelta
+import logging
+import sys
+import time
+
+from focus_project import _api, ledger
+from focus_project._errors import RegistryError
+from focus_project.minutes import minutes_for
+
+_LOG = logging.getLogger("focus_project.harvest")
+
+
+def harvest_day(day: date, now: float) -> int:
+    """Credit ``day``; return the minutes added (0 with no project)."""
+    project = _api.active(day)
+    if project is None:
+        return 0
+    total = minutes_for(day)
+    added = ledger.credit_to(project, day, total, now=now)
+    if added:
+        _LOG.info(
+            "credited %d min to %s on %s (total %d)", added, project.id, day, total
+        )
+    return added
+
+
+def harvest(now: float | None = None) -> int:
+    """Harvest yesterday and today; 0 on success, 1 if any path failed."""
+    moment = time.time() if now is None else now
+    today = date.fromtimestamp(moment)  # noqa: DTZ012 (local day)
+    status = 0
+    for day in (today - timedelta(days=1), today):
+        try:
+            harvest_day(day, moment)
+        except (RegistryError, ledger.LedgerError, OSError) as exc:
+            _LOG.error("harvest %s failed: %s", day, exc)
+            status = 1
+    return status
+
+
+def main() -> None:
+    """Entry point of ``python -m focus_project.harvest``."""
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+    sys.exit(harvest())
+
+
+if __name__ == "__main__":
+    main()

focus_project/focus_project/ledger.py

@@ -0,0 +1,178 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""The signed credit ledger the ``focus`` earner reads.
+
+Row contract (earned_time's "tutor row contract"): ``detail.minutes`` is the
+positive increment the row pays, ``entry_id`` is
+``<project>-<YYYYMMDD>-m<n>`` with ``n`` the day's cumulative credited
+minutes AFTER the row (so ids never overlap and a retry rewrites the same
+id), ``day`` is the local date, ``detail.ended_at`` is when the last paid
+minute ended and never passes the day's end. The only writer is the
+harvester; every write is read back and verified.
+"""
+
+from __future__ import annotations
+
+import fcntl
+import json
+import logging
+import os
+from pathlib import Path
+import tempfile
+import time
+from typing import TYPE_CHECKING, Any, Final
+
+import earned_time
+
+from focus_project._paths import data_dir
+from focus_project.sources import day_bounds
+
+if TYPE_CHECKING:
+    from datetime import date
+
+    from focus_project._model import Project
+
+CREDIT: Final = "credit"
+KEY_ENV: Final = "FOCUS_PROJECT_KEY"
+_REAL_KEY: Final = Path("/etc/workout-locker/hmac.key")
+_LOG = logging.getLogger(__name__)
+
+
+class LedgerError(Exception):
+    """The ledger exists but is not a ledger: never silently emptied."""
+
+
+def key_path() -> Path:
+    """The signing key: ``$FOCUS_PROJECT_KEY`` or the shared root-owned key."""
+    override = os.environ.get(KEY_ENV)
+    return Path(override) if override else _REAL_KEY
+
+
+def ledger_path() -> Path:
+    """The ledger file inside the data dir."""
+    return data_dir() / "ledger.json"
+
+
+def read_key() -> bytes | None:
+    """The signing key, or None if unreadable or empty (logged)."""
+    try:
+        return key_path().read_bytes().strip() or None
+    except OSError as exc:
+        _LOG.warning("signing key %s unreadable: %s", key_path(), exc)
+        return None
+
+
+def read_rows(path: Path) -> list[Any]:
+    """The raw ``entries`` array; a missing file is empty.
+
+    Raises:
+        LedgerError: unreadable, not JSON, or no entries array.
+    """
+    try:
+        raw = json.loads(path.read_text(encoding="utf-8"))
+    except FileNotFoundError:
+        return []
+    except (OSError, ValueError) as exc:
+        msg = f"{path} is unreadable: {exc}"
+        raise LedgerError(msg) from exc
+    rows = raw.get("entries") if isinstance(raw, dict) else None
+    if not isinstance(rows, list):
+        msg = f"{path} has no entries array"
+        raise LedgerError(msg)
+    return rows
+
+
+def _write_rows(path: Path, rows: list[Any]) -> None:
+    """Atomically replace the ledger: readers see the old file or the new one."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile(
+        "w", encoding="utf-8", dir=path.parent, suffix=".tmp", delete=False
+    ) as handle:
+        json.dump({"version": 1, "entries": rows}, handle, indent=1)
+        handle.flush()
+        os.fsync(handle.fileno())
+    Path(handle.name).replace(path)
+
+
+def paid_minutes(rows: list[Any], key: bytes, project_id: str, day: date) -> int:
+    """Minutes already paid for ``project_id`` on ``day`` by verified rows.
+
+    A repeated ``entry_id`` pays once, at its smallest count.
+    """
+    by_id: dict[str, int] = {}
+    for row in rows:
+        if not (isinstance(row, dict) and row.get("kind") == CREDIT):
+            continue
+        detail = row.get("detail")
+        if not isinstance(detail, dict):
+            continue
+        minutes = detail.get("minutes")
+        if (
+            row.get("day") != day.isoformat()
+            or detail.get("project") != project_id
+            or not isinstance(minutes, int)
+            or isinstance(minutes, bool)
+            or minutes < 1
+            or not earned_time.verified(row, key)
+        ):
+            continue
+        eid = str(row.get("entry_id"))
+        by_id[eid] = min(by_id.get(eid, minutes), minutes)
+    return sum(by_id.values())
+
+
+def build_row(  # noqa: PLR0913
+    project_id: str, day: date, total: int, minutes: int, key: bytes, *, now: float
+) -> dict[str, Any]:
+    """The signed row paying ``minutes`` that bring the day's total to ``total``."""
+    ended = min(now, day_bounds(day)[1] - 1)
+    row: dict[str, Any] = {
+        "kind": CREDIT,
+        "entry_id": f"{project_id}-{day:%Y%m%d}-m{total}",
+        "day": day.isoformat(),
+        "created_at": float(now),
+        "detail": {
+            "project": project_id,
+            "minutes": int(minutes),
+            "total": int(total),
+            "ended_at": float(ended),
+        },
+    }
+    row["hmac"] = str(earned_time.entry_signature(row, key))
+    return row
+
+
+def credit_to(
+    project: Project, day: date, total: int, *, now: float | None = None
+) -> int:
+    """Pay the day up to ``total`` cumulative minutes; return minutes added.
+
+    Idempotent: nothing is written when the ledger already pays ``total`` or
+    more. Raises LedgerError (unreadable ledger, unreadable key, unverifiable
+    write) so the harvester can fail loudly instead of crediting nothing quietly.
+    """
+    key = read_key()
+    if key is None:
+        msg = f"signing key {key_path()} unreadable"
+        raise LedgerError(msg)
+    path = ledger_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.with_suffix(".lock").open("w") as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX)
+        rows = read_rows(path)
+        paid = paid_minutes(rows, key, project.id, day)
+        if total <= paid:
+            return 0
+        row = build_row(
+            project.id, day, total, total - paid, key, now=now or time.time()
+        )
+        if any(
+            isinstance(r, dict) and r.get("entry_id") == row["entry_id"] for r in rows
+        ):
+            msg = f"entry_id {row['entry_id']} already present but unverified"
+            raise LedgerError(msg)
+        _write_rows(path, [*rows, row])
+        stored = read_rows(path)
+    if paid_minutes(stored, key, project.id, day) != total:
+        msg = f"row {row['entry_id']} not found verified on read-back"
+        raise LedgerError(msg)
+    return total - paid

focus_project/focus_project/minutes.py

@@ -1,15 +1,42 @@
 # Copyright (c) 2026 Krzysztof Rudnicki
-"""Credited focus minutes per day. Owned by agent D: replace the body.
+"""Credited focus minutes per day: the union of every source's intervals.
 
 Contract: union of activity intervals on the local calendar day, capped at
-the active project's daily_minutes. This stub returns 0 until D lands.
+the active project's ``daily_minutes``; overlaps never double count; whole
+minutes (floor). An unreadable registry reads as 0 (logged): fail closed.
 """
 
 from __future__ import annotations
 
 from datetime import date  # noqa: TC003
+import logging
+
+from focus_project import _api, sources
+from focus_project._errors import RegistryError
+
+_LOG = logging.getLogger(__name__)
 
 
-def minutes_for(day: date) -> int:  # noqa: ARG001
-    """Minutes of focus work credited on ``day`` (stub: always 0)."""
-    return 0
+def union_seconds(spans: list[sources.Interval]) -> float:
+    """Total length of the union of ``spans``."""
+    total = 0.0
+    end = float("-inf")
+    for lo, hi in sorted(spans):
+        if hi <= end:
+            continue
+        total += hi - max(lo, end)
+        end = hi
+    return total
+
+
+def minutes_for(day: date) -> int:
+    """Minutes of focus work credited on ``day`` (0 with no project active)."""
+    try:
+        project = _api.active(day)
+    except (RegistryError, OSError) as exc:
+        _LOG.warning("focus_project registry unreadable, 0 minutes: %s", exc)
+        return 0
+    if project is None:
+        return 0
+    seconds = union_seconds(sources.collect(day, project))
+    return min(int(seconds // 60), project.daily_minutes)

focus_project/focus_project/sources/__init__.py

@@ -63,17 +63,13 @@ def collect_by_source(day: date, project: Project) -> dict[str, list[Interval]]:
         module_name = f"{__name__}.{name}"
         try:
             module = importlib.import_module(module_name)
-        except ModuleNotFoundError as exc:
-            if exc.name == module_name:
+        except Exception as exc:  # noqa: BLE001 (a broken adapter must add 0)
+            if isinstance(exc, ModuleNotFoundError) and exc.name == module_name:
                 _LOG.info("source %s: not installed (inactive)", name)
                 continue
             _LOG.exception("source %s: import failed, contributes 0", name)
             found[name] = []
             continue
-        except Exception:  # noqa: BLE001 (any broken adapter must contribute 0)
-            _LOG.exception("source %s: import failed, contributes 0", name)
-            found[name] = []
-            continue
         try:
             found[name] = clip(list(module.intervals(day, project)), bounds)
         except Exception:  # noqa: BLE001 (fail closed: this source adds 0)

focus_project/focus_project/sources/activity.py

@@ -0,0 +1,80 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""Adapter for ``activity.jsonl``: the timer and every heartbeat writer.
+
+``timer`` events form segments ``start -> checkin ... -> stop``. A span
+between two consecutive events counts only if the gap is at most
+:data:`MAX_GAP_S`: a missed check-in stops the segment, and a late check-in
+starts a fresh one. Time after the last check-in is not credited until the
+next event vouches for it. Every other source (tutor, sql_drills, ...) is a
+heartbeat covering the five minutes before it. Only events recorded for
+``project`` count.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import TYPE_CHECKING, Any
+
+from focus_project._paths import activity_path
+from focus_project.sources import Interval, heartbeat_interval
+
+if TYPE_CHECKING:
+    from datetime import date
+
+    from focus_project._model import Project
+
+MAX_GAP_S = 600.0
+_LOG = logging.getLogger(__name__)
+
+
+def read_events(project_id: str) -> list[dict[str, Any]]:
+    """Well-formed events of ``project_id`` in time order; junk lines skipped."""
+    try:
+        text = activity_path().read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return []
+    events: list[dict[str, Any]] = []
+    for line in text.splitlines():
+        try:
+            row = json.loads(line)
+            ok = (
+                isinstance(row, dict)
+                and row.get("project") == project_id
+                and isinstance(row["ts"], (int, float))
+                and not isinstance(row["ts"], bool)
+            )
+        except ValueError, KeyError:
+            ok = False
+        if ok:
+            events.append(row)
+        elif line.strip():
+            _LOG.debug("activity line skipped: %.80s", line)
+    events.sort(key=lambda e: e["ts"])
+    return events
+
+
+def timer_intervals(events: list[dict[str, Any]]) -> list[Interval]:
+    """Credited spans of the ``timer`` events (already time-ordered)."""
+    spans: list[Interval] = []
+    last: float | None = None  # the last event of the open segment
+    for event in events:
+        ts = float(event["ts"])
+        kind = event.get("kind")
+        if kind not in {"start", "checkin", "stop"}:
+            continue
+        if last is not None and ts - last <= MAX_GAP_S:
+            spans.append((last, ts))
+        last = None if kind == "stop" else ts
+    return spans
+
+
+def intervals(day: date, project: Project) -> list[Interval]:  # noqa: ARG001
+    """Timer segments plus heartbeats of ``project`` (the registry clips)."""
+    events = read_events(project.id)
+    timer = [e for e in events if e.get("source") == "timer"]
+    beats = [e for e in events if e.get("source") != "timer"]
+    return [
+        *timer_intervals(timer),
+        *(heartbeat_interval(float(e["ts"])) for e in beats),
+    ]

focus_project/focus_project/sources/leetcode.py

@@ -0,0 +1,67 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""Adapter for accepted LeetCode submissions (leetcode-guard's ledger).
+
+Read-only. Each accepted submission (``ac:<id>`` credit row, HMAC verified)
+is a heartbeat at ``detail.submitted_at``, so it covers the five minutes
+before it. The window is deliberately NOT longer: the time spent reading and
+thinking on the site is the browser adapter's to count, and a submission
+alone proves only that work happened, not for how long. Applies only to a
+project whose ``site_patterns`` name leetcode; any other project is not
+LeetCode practice. No readable signing key: contributes nothing (logged).
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import earned_time
+
+from focus_project import ledger
+from focus_project.sources import Interval, heartbeat_interval
+
+if TYPE_CHECKING:
+    from datetime import date
+
+    from focus_project._model import Project
+
+LEDGER_ENV = "FOCUS_PROJECT_LEETCODE_LEDGER"
+_LOG = logging.getLogger(__name__)
+
+
+def ledger_file() -> Path:
+    """leetcode-guard's ledger, or the test override."""
+    override = os.environ.get(LEDGER_ENV)
+    if override:
+        return Path(override)
+    return Path.home() / ".local" / "share" / "leetcode_guard" / "ledger.json"
+
+
+def _submitted_at(row: object, key: bytes) -> float | None:
+    """The submission time of a verified accepted-submission row, else None."""
+    if not (
+        isinstance(row, dict)
+        and row.get("kind") == "credit"
+        and str(row.get("entry_id", "")).startswith("ac:")
+        and earned_time.verified(row, key)
+    ):
+        return None
+    detail = row.get("detail")
+    try:
+        return float(detail["submitted_at"])  # type: ignore[index]
+    except KeyError, TypeError, ValueError:
+        return None
+
+
+def intervals(day: date, project: Project) -> list[Interval]:  # noqa: ARG001
+    """One heartbeat per accepted submission (the registry clips to the day)."""
+    if not any("leetcode" in pattern.lower() for pattern in project.site_patterns):
+        return []
+    key = ledger.read_key()
+    if key is None:
+        _LOG.warning("leetcode source: no signing key, contributes 0")
+        return []
+    stamps = (_submitted_at(row, key) for row in ledger.read_rows(ledger_file()))
+    return [heartbeat_interval(ts) for ts in stamps if ts is not None]

focus_project/focus_project/tests/conftest.py

@@ -19,6 +19,10 @@ if TYPE_CHECKING:
 def _redirect_state(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
     """Point the data directory at ``tmp_path`` before any test runs."""
     monkeypatch.setenv("FOCUS_PROJECT_DATA", str(tmp_path / "data"))
+    # Read-only evidence files and the signing key: never the real ones.
+    monkeypatch.setenv("FOCUS_PROJECT_LEETCODE_LEDGER", str(tmp_path / "lc.json"))
+    monkeypatch.setenv("FOCUS_PROJECT_KT_DB", str(tmp_path / "kt.db"))
+    monkeypatch.setenv("FOCUS_PROJECT_KEY", str(tmp_path / "hmac.key"))
 
 
 @pytest.fixture

focus_project/focus_project/tests/test_ledger_harvest.py

@@ -0,0 +1,181 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""Signed rows, idempotency, harvester paths."""
+
+from __future__ import annotations
+
+from datetime import date, timedelta
+import json
+import logging
+from pathlib import Path  # noqa: TC003
+
+import earned_time
+import pytest
+
+import focus_project as fp
+from focus_project import harvest, ledger, minutes
+from focus_project.tests.test_minutes import D1, at, events, project
+
+KEY = b"k" * 32
+
+
+@pytest.fixture(autouse=True)
+def _key(tmp_path: Path) -> None:
+    (tmp_path / "hmac.key").write_bytes(KEY)
+
+
+def rows() -> list[dict]:
+    return json.loads(ledger.ledger_path().read_text())["entries"]
+
+
+def beats(*times: float) -> None:
+    events(*(("tutor", "heartbeat", t) for t in times))
+
+
+def test_credit_writes_signed_row_with_contract_fields() -> None:
+    project()
+    beats(at(9, 30))
+    assert harvest.harvest(now=at(9, 31)) == 0
+    (row,) = rows()
+    assert row["entry_id"] == "hr-20261102-m5"
+    assert row["day"] == "2026-11-02"
+    assert row["detail"]["minutes"] == 5
+    assert row["detail"]["project"] == "hr"
+    assert row["detail"]["ended_at"] == at(9, 31)
+    assert earned_time.verified(row, KEY)
+
+
+def test_second_run_is_idempotent() -> None:
+    project()
+    beats(at(9, 30))
+    harvest.harvest(now=at(9, 31))
+    harvest.harvest(now=at(9, 32))
+    assert len(rows()) == 1
+
+
+def test_growth_writes_increment_rows_that_sum_to_total() -> None:
+    project()
+    beats(at(9, 30))
+    harvest.harvest(now=at(9, 31))
+    beats(at(9, 38))  # union: 9:25..9:30 + 9:33..9:38 = 10
+    harvest.harvest(now=at(9, 39))
+    got = rows()
+    assert [r["entry_id"] for r in got] == ["hr-20261102-m5", "hr-20261102-m10"]
+    assert [r["detail"]["minutes"] for r in got] == [5, 5]
+    assert ledger.paid_minutes(got, KEY, "hr", D1) == minutes.minutes_for(D1) == 10
+
+
+def test_yesterday_is_harvested_after_midnight() -> None:
+    project()
+    beats(at(23, 58))
+    harvest.harvest(now=at(0, 1, day=D1 + timedelta(days=1)))
+    (row,) = rows()
+    assert row["day"] == "2026-11-02"
+    assert row["detail"]["ended_at"] < at(0, 0, day=D1 + timedelta(days=1))
+
+
+def test_no_project_writes_nothing() -> None:
+    beats(at(9, 30))
+    assert harvest.harvest(now=at(9, 31)) == 0
+    assert not ledger.ledger_path().exists()
+
+
+def test_unreadable_key_fails_loudly_and_writes_nothing(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    (tmp_path / "hmac.key").unlink()
+    project()
+    beats(at(9, 30))
+    with caplog.at_level(logging.ERROR):
+        assert harvest.harvest(now=at(9, 31)) == 1
+    assert "unreadable" in caplog.text
+    assert not ledger.ledger_path().exists()
+
+
+def test_corrupt_ledger_is_not_emptied() -> None:
+    project()
+    beats(at(9, 30))
+    ledger.ledger_path().parent.mkdir(parents=True, exist_ok=True)
+    ledger.ledger_path().write_text("{oops")
+    assert harvest.harvest(now=at(9, 31)) == 1
+    assert ledger.ledger_path().read_text() == "{oops"
+
+
+def test_ledger_without_entries_array_is_an_error() -> None:
+    ledger.ledger_path().parent.mkdir(parents=True, exist_ok=True)
+    ledger.ledger_path().write_text('{"entries": 3}')
+    with pytest.raises(ledger.LedgerError):
+        ledger.read_rows(ledger.ledger_path())
+
+
+def test_broken_registry_fails_loudly(monkeypatch: pytest.MonkeyPatch) -> None:
+    def boom(_day: date) -> None:
+        msg = "corrupt"
+        raise fp.RegistryError(msg)
+
+    monkeypatch.setattr("focus_project._api.active", boom)
+    assert harvest.harvest(now=at(9, 31)) == 1
+
+
+def test_empty_key_file_counts_as_unreadable(tmp_path: Path) -> None:
+    (tmp_path / "hmac.key").write_bytes(b"\n")
+    assert ledger.read_key() is None
+
+
+def test_forged_and_malformed_rows_do_not_count() -> None:
+    good = ledger.build_row("hr", D1, 5, 5, KEY, now=at(9))
+    forged = {
+        **good,
+        "entry_id": "hr-20261102-m50",
+        "detail": {**good["detail"], "minutes": 50},
+    }
+    other_key = ledger.build_row("hr", D1, 9, 9, b"x" * 32, now=at(9))
+    junk: list = [
+        "str",
+        {"kind": "charge"},
+        {**good, "detail": "no"},
+        {**good, "day": "2026-11-03"},
+        {**good, "detail": {**good["detail"], "project": "other"}},
+        {**good, "detail": {**good["detail"], "minutes": True}},
+        {**good, "detail": {**good["detail"], "minutes": 0}},
+        forged,
+        other_key,
+    ]
+    assert ledger.paid_minutes([good, *junk], KEY, "hr", D1) == 5
+
+
+def test_repeated_entry_id_pays_once_at_smallest() -> None:
+    a = ledger.build_row("hr", D1, 10, 10, KEY, now=at(9))
+    b = ledger.build_row("hr", D1, 10, 4, KEY, now=at(9))
+    assert ledger.paid_minutes([a, b], KEY, "hr", D1) == 4
+
+
+def test_unverified_same_id_row_blocks_write() -> None:
+    p = project_obj()
+    row = ledger.build_row("hr", D1, 5, 5, KEY, now=at(9))
+    row["hmac"] = "bad"
+    ledger.ledger_path().parent.mkdir(parents=True, exist_ok=True)
+    ledger.ledger_path().write_text(json.dumps({"entries": [row]}))
+    with pytest.raises(ledger.LedgerError, match="unverified"):
+        ledger.credit_to(p, D1, 5, now=at(9))
+
+
+def test_readback_failure_is_reported(monkeypatch: pytest.MonkeyPatch) -> None:
+    p = project_obj()
+    real = ledger._write_rows
+    monkeypatch.setattr(ledger, "_write_rows", lambda path, _rows: real(path, []))
+    with pytest.raises(ledger.LedgerError, match="read-back"):
+        ledger.credit_to(p, D1, 5, now=at(9))
+
+
+def project_obj() -> fp.Project:
+    project()
+    found = fp.active(D1)
+    assert found is not None
+    return found
+
+
+def test_main_exits_with_status(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(harvest, "harvest", lambda: 7)
+    with pytest.raises(SystemExit) as info:
+        harvest.main()
+    assert info.value.code == 7

focus_project/focus_project/tests/test_leetcode_source.py

@@ -0,0 +1,129 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""The accepted-submission adapter, incl. read-only replay of the real ledger."""
+
+from __future__ import annotations
+
+from datetime import date
+import json
+from pathlib import Path
+from typing import Any
+
+import earned_time
+import pytest
+
+import focus_project as fp
+from focus_project import minutes, sources
+from focus_project.sources import leetcode
+from focus_project.tests.test_minutes import D1, at, project
+
+KEY = b"k" * 32
+
+
+def ac_row(ts: float, eid: str = "ac:1", key: bytes = KEY) -> dict:
+    row: dict[str, Any] = {
+        "entry_id": eid,
+        "kind": "credit",
+        "day": "2026-11-02",
+        "amount": 1,
+        "detail": {"title_slug": "x", "submitted_at": str(int(ts))},
+    }
+    row["hmac"] = earned_time.entry_signature(row, key)
+    return row
+
+
+def write(tmp_path: Path, *rows: object) -> None:
+    (tmp_path / "hmac.key").write_bytes(KEY)
+    (tmp_path / "lc.json").write_text(json.dumps({"version": 1, "entries": list(rows)}))
+
+
+def test_submission_is_a_five_minute_heartbeat(tmp_path: Path) -> None:
+    project()
+    write(tmp_path, ac_row(at(10)), ac_row(at(10, 2), "ac:2"))
+    assert minutes.minutes_for(D1) == 7  # union 9:55..10:02
+
+
+def test_other_rows_and_forgeries_ignored(tmp_path: Path) -> None:
+    project()
+    charge = {**ac_row(at(10), "charge:2026-11-02"), "kind": "charge"}
+    bad_detail = ac_row(at(10), "ac:3")
+    bad_detail["detail"] = {"submitted_at": "soon"}
+    bad_detail["hmac"] = earned_time.entry_signature(bad_detail, KEY)
+    no_detail = ac_row(at(10), "ac:4")
+    no_detail["detail"] = None
+    no_detail["hmac"] = earned_time.entry_signature(no_detail, KEY)
+    write(
+        tmp_path,
+        charge,
+        ac_row(at(10), "ac:5", key=b"x" * 32),
+        bad_detail,
+        no_detail,
+        "junk",
+    )
+    assert minutes.minutes_for(D1) == 0
+
+
+def test_project_without_leetcode_pattern_ignores_ledger(tmp_path: Path) -> None:
+    fp.create_project(
+        "sql",
+        name="SQL",
+        start=D1,
+        end=D1,
+        replaces=["reading"],
+        daily_minutes=60,
+        unlock_minutes=10,
+        site_patterns=["sqlzoo.net"],
+    )
+    write(tmp_path, ac_row(at(10)))
+    assert minutes.minutes_for(D1) == 0
+
+
+def test_no_key_contributes_zero(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    project()
+    write(tmp_path, ac_row(at(10)))
+    (tmp_path / "hmac.key").unlink()
+    assert minutes.minutes_for(D1) == 0
+    assert "no signing key" in caplog.text
+
+
+def test_missing_or_corrupt_ledger_contributes_zero(tmp_path: Path) -> None:
+    project()
+    (tmp_path / "hmac.key").write_bytes(KEY)
+    assert minutes.minutes_for(D1) == 0
+    (tmp_path / "lc.json").write_text("{")
+    assert minutes.minutes_for(D1) == 0  # LedgerError caught by collect
+
+
+def test_default_ledger_path(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    monkeypatch.delenv(leetcode.LEDGER_ENV)
+    monkeypatch.setenv("HOME", str(tmp_path))
+    assert (
+        leetcode.ledger_file() == tmp_path / ".local/share/leetcode_guard/ledger.json"
+    )
+
+
+REAL = Path.home() / ".local/share/leetcode_guard/ledger.json"
+REAL_KEY = Path("/etc/workout-locker/hmac.key")
+
+
+@pytest.mark.skipif(not (REAL.exists() and REAL_KEY.exists()), reason="no live ledger")
+def test_read_only_replay_of_real_ledger(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Every real accepted submission yields a heartbeat; nothing is written."""
+    before = REAL.stat().st_mtime_ns
+    monkeypatch.setenv(leetcode.LEDGER_ENV, str(REAL))
+    monkeypatch.setenv("FOCUS_PROJECT_KEY", str(REAL_KEY))
+    proj = fp.Project(
+        id="x",
+        name="x",
+        start=date(2026, 1, 1),
+        end=date(2026, 12, 31),
+        replaces=("leetcode",),
+        daily_minutes=120,
+        unlock_minutes=30,
+        site_patterns=("leetcode.com",),
+    )
+    got = leetcode.intervals(D1, proj)
+    assert got
+    assert all(b - a == sources.HEARTBEAT_WINDOW_S for a, b in got)
+    assert REAL.stat().st_mtime_ns == before

focus_project/focus_project/tests/test_minutes.py

@@ -0,0 +1,202 @@
+# Copyright (c) 2026 Krzysztof Rudnicki
+"""minutes_for: union of intervals, gap rule, cap, project filter."""
+
+from __future__ import annotations
+
+from datetime import date, datetime
+import json
+from typing import TYPE_CHECKING
+
+import focus_project as fp
+from focus_project import minutes
+from focus_project._paths import activity_path
+
+if TYPE_CHECKING:
+    import pytest
+
+D1 = date(2026, 11, 2)
+
+
+def at(hour: int, minute: int = 0, second: int = 0, day: date = D1) -> float:
+    """Local unix time on ``day``."""
+    return (
+        datetime(day.year, day.month, day.day, hour, minute, second)
+        .astimezone()
+        .timestamp()
+    )
+
+
+def project(daily: int = 120) -> None:
+    fp.create_project(
+        "hr",
+        name="HR prep",
+        start=D1,
+        end=date(2026, 11, 5),
+        replaces=["leetcode"],
+        daily_minutes=daily,
+        unlock_minutes=min(30, daily),
+        site_patterns=["leetcode.com"],
+    )
+
+
+def events(*rows: tuple[str, str, float]) -> None:
+    activity_path().parent.mkdir(parents=True, exist_ok=True)
+    with activity_path().open("a", encoding="utf-8") as handle:
+        for source, kind, ts in rows:
+            row = {
+                "ts": ts,
+                "source": source,
+                "kind": kind,
+                "project": "hr",
+                "dev": "pc",
+            }
+            handle.write(json.dumps(row) + "\n")
+
+
+def test_no_project_is_zero() -> None:
+    events(("timer", "start", at(9)), ("timer", "stop", at(9, 5)))
+    assert minutes.minutes_for(D1) == 0
+
+
+def test_timer_start_checkin_stop() -> None:
+    project()
+    events(
+        ("timer", "start", at(9)),
+        ("timer", "checkin", at(9, 5)),
+        ("timer", "checkin", at(9, 10)),
+        ("timer", "stop", at(9, 12)),
+    )
+    assert minutes.minutes_for(D1) == 12
+
+
+def test_missed_checkin_stops_the_segment() -> None:
+    project()
+    events(
+        ("timer", "start", at(9)),
+        ("timer", "checkin", at(9, 5)),
+        ("timer", "checkin", at(9, 30)),  # 25 min gap > 600 s: not credited
+        ("timer", "stop", at(9, 33)),
+    )
+    assert minutes.minutes_for(D1) == 5 + 3
+
+
+def test_gap_of_exactly_600s_counts() -> None:
+    project()
+    events(("timer", "start", at(9)), ("timer", "checkin", at(9, 10)))
+    assert minutes.minutes_for(D1) == 10
+
+
+def test_open_timer_credits_only_up_to_last_checkin() -> None:
+    project()
+    events(("timer", "start", at(9)), ("timer", "checkin", at(9, 4)))
+    assert minutes.minutes_for(D1) == 4
+
+
+def test_stop_without_open_segment_ignored() -> None:
+    project()
+    events(
+        ("timer", "stop", at(9)),
+        ("timer", "start", at(10)),
+        ("timer", "stop", at(10, 2)),
+    )
+    assert minutes.minutes_for(D1) == 2
+
+
+def test_start_while_open_acts_as_checkin() -> None:
+    project()
+    events(
+        ("timer", "start", at(9)),
+        ("timer", "start", at(9, 3)),
+        ("timer", "stop", at(9, 5)),
+    )
+    assert minutes.minutes_for(D1) == 5
+
+
+def test_heartbeat_covers_five_minutes() -> None:
+    project()
+    events(("sql_drills", "heartbeat", at(11)))
+    assert minutes.minutes_for(D1) == 5
+
+
+def test_overlaps_never_double_count() -> None:
+    project()
+    events(
+        ("timer", "start", at(9)),
+        ("timer", "stop", at(9, 6)),
+        ("tutor", "heartbeat", at(9, 5)),  # [9:00, 9:05] inside the timer span
+        ("sql_drills", "heartbeat", at(9, 8)),  # [9:03, 9:08] extends it by 2
+    )
+    assert minutes.minutes_for(D1) == 8
+
+
+def test_capped_at_daily_minutes() -> None:
+    project(daily=10)
+    events(
+        ("timer", "start", at(9)),
+        ("timer", "stop", at(9, 8)),
+        ("tutor", "heartbeat", at(12)),
+    )
+    assert minutes.minutes_for(D1) == 10
+
+
+def test_other_day_and_other_project_events_ignored() -> None:
+    project()
+    events(("tutor", "heartbeat", at(11, day=date(2026, 11, 3))))
+    with activity_path().open("a", encoding="utf-8") as handle:
+        row = {"ts": at(11), "source": "tutor", "kind": "heartbeat", "project": "x"}
+        handle.write(json.dumps(row) + "\n")
+        row = {"ts": at(12), "source": "tutor", "kind": "heartbeat", "project": None}
+        handle.write(json.dumps(row) + "\n")
+    assert minutes.minutes_for(D1) == 0
+
+
+def test_garbage_lines_skipped() -> None:
+    project()
+    events(("tutor", "heartbeat", at(11)))
+    with activity_path().open("a", encoding="utf-8") as handle:
+        handle.write(
+            "not json\n[1]\n" + json.dumps({"ts": "x", "project": "hr"}) + "\n\n"
+        )
+        handle.write(
+            json.dumps({"ts": at(12), "source": "tutor", "project": "hr"}) + "\n"
+        )
+    assert minutes.minutes_for(D1) == 5 + 5
+
+
+def test_interval_spanning_midnight_is_clipped() -> None:
+    project()
+    events(("tutor", "heartbeat", at(0, 2)))  # covers 23:57..00:02 of D1
+    assert minutes.minutes_for(D1) == 2
+    assert minutes.minutes_for(date(2026, 11, 1)) == 0  # no project there
+
+
+def test_broken_registry_reads_zero(monkeypatch: pytest.MonkeyPatch) -> None:
+    from focus_project import _api, _errors
+
+    def boom(_day: date) -> None:
+        msg = "corrupt"
+        raise _errors.RegistryError(msg)
+
+    monkeypatch.setattr(_api, "active", boom)
+    assert minutes.minutes_for(D1) == 0
+
+
+def test_missing_activity_file_is_zero() -> None:
+    project()
+    assert minutes.minutes_for(D1) == 0
+
+
+def test_contained_heartbeat_adds_nothing() -> None:
+    project()
+    events(
+        ("timer", "start", at(9)),
+        ("timer", "stop", at(9, 10)),
+        ("tutor", "heartbeat", at(9, 8)),
+    )
+    assert minutes.minutes_for(D1) == 10
+
+
+def test_non_timer_kinds_from_timer_source_ignored() -> None:
+    project()
+    events(("timer", "heartbeat", at(9)))
+    assert minutes.minutes_for(D1) == 0

focus_project/focus_project/tests/test_sources.py

@@ -9,14 +9,14 @@ import sys
 import types
 from typing import TYPE_CHECKING
 
-import pytest
-
 from focus_project import sources
 from focus_project._model import Project
 
 if TYPE_CHECKING:
     from collections.abc import Callable
 
+    import pytest
+
 DAY = date(2026, 11, 2)
 PROJECT = Project(
     id="p",
@@ -92,3 +92,17 @@ def test_missing_adapter_is_inactive_not_an_error(
     _fake(monkeypatch, "activity", _none)
     monkeypatch.setattr(sources, "ADAPTERS", ("activity", "nonexistent"))
     assert "nonexistent" not in sources.collect_by_source(DAY, PROJECT)
+
+
+def test_broken_adapter_import_contributes_zero(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    def broken(name: str) -> None:
+        raise ImportError(name)
+
+    monkeypatch.setattr(sources.importlib, "import_module", broken)
+    with caplog.at_level(logging.ERROR):
+        result = sources.collect_by_source(DAY, PROJECT)
+    assert all(spans == [] for spans in result.values())
+    assert set(result) == set(sources.ADAPTERS)
+    assert "import failed" in caplog.text

focus_project/install_harvest.sh

@@ -0,0 +1,33 @@
+#!/bin/bash
+# Install + enable the focus_project harvester units (systemd --user).
+# Run as the desktop user (never sudo). Idempotent. Needs focus_project
+# importable by /usr/bin/python3 (see install.sh).
+set -euo pipefail
+
+REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+readonly REPO_DIR
+readonly UNIT_DIR="${XDG_CONFIG_HOME:-$HOME/.config}/systemd/user"
+readonly DATA_DIR="${FOCUS_PROJECT_DATA:-$HOME/.local/share/focus_project}"
+readonly UNITS=(focus-project-harvest.service focus-project-harvest.path focus-project-harvest.timer)
+
+if [[ "$(id -u)" -eq 0 ]]; then
+    echo "run as your own user, not root" >&2
+    exit 1
+fi
+(cd / && /usr/bin/python3 -c 'import focus_project') ||
+    {
+        echo "focus_project not importable: run $REPO_DIR/install.sh first" >&2
+        exit 1
+    }
+
+# The .path unit watches files inside this directory.
+mkdir -p "$DATA_DIR" "$UNIT_DIR"
+for unit in "${UNITS[@]}"; do
+    install -m 0644 "$REPO_DIR/systemd/$unit" "$UNIT_DIR/$unit"
+done
+systemctl --user daemon-reload
+systemctl --user enable --now focus-project-harvest.path focus-project-harvest.timer
+for unit in "${UNITS[@]}"; do
+    diff -q "$REPO_DIR/systemd/$unit" "$UNIT_DIR/$unit"
+done
+echo "focus-project-harvest units installed and enabled"

focus_project/systemd/focus-project-harvest.path

@@ -0,0 +1,10 @@
+[Unit]
+Description=focus_project: harvest the moment activity or a LeetCode accept lands
+
+[Path]
+PathChanged=%h/.local/share/focus_project/activity.jsonl
+PathChanged=%h/.local/share/leetcode_guard/ledger.json
+Unit=focus-project-harvest.service
+
+[Install]
+WantedBy=default.target

focus_project/systemd/focus-project-harvest.service

@@ -0,0 +1,10 @@
+[Unit]
+Description=focus_project: credit focus minutes to the signed ledger
+# The .path unit fires on every activity line; bursts must not trip the limit.
+StartLimitIntervalSec=0
+
+[Service]
+Type=oneshot
+WorkingDirectory=/
+ExecStart=/usr/bin/python3 -m focus_project.harvest
+Nice=10

focus_project/systemd/focus-project-harvest.timer

@@ -0,0 +1,11 @@
+[Unit]
+Description=focus_project: harvest every minute (covers sources with no file event)
+
+[Timer]
+OnBootSec=1min
+OnUnitInactiveSec=1min
+AccuracySec=5s
+Unit=focus-project-harvest.service
+
+[Install]
+WantedBy=timers.target