Carry unspent Fri-Sun gaming time to the rest of the Fri-Mon block

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-10-02 21:07:49 +02:00
parent abf59d768b
commit 9586128163
14 changed files with 340 additions and 29 deletions

View File

@ -158,6 +158,8 @@ def build_rules(rules: PlaytimeRules) -> dict[str, Any]:
# is what would let this view report a budget the daemon is not
# enforcing. See _budget_resolve's "one seam".
"budget_reason": rules.budget_reason,
# Unspent Fri-Sun time inherited by today; already inside budget_seconds.
"carry_seconds": rules.carry_seconds,
"bonuses": {
"base": rules.base_seconds,
"workout": rules.workout_seconds,

View File

@ -14,6 +14,7 @@ import os
from typing import TYPE_CHECKING
from steam_backlog_enforcer._enforce_loop import do_enforce
from steam_backlog_enforcer._gaming_days import gaming_day_key
from steam_backlog_enforcer._playtime_block import (
block_targets,
mounted_targets,
@ -21,7 +22,6 @@ from steam_backlog_enforcer._playtime_block import (
)
from steam_backlog_enforcer._playtime_state import (
PlaytimeState,
gaming_day_key,
load_state,
rules_for,
save_state,

View File

@ -0,0 +1,147 @@
"""Gaming-day arithmetic: which day *now* is, and what it inherits.
A gaming "day" runs 06:00-05:59 local, so late-night sessions count against
the day they started in.
**Free-day carry-over.** Friday to Monday is one block. Whatever a Friday,
Saturday or Sunday leaves unspent is split evenly, with no strings attached,
across the days still left in that block:
Fri leftover -> Sat, Sun, Mon (a third each)
Sat leftover -> Sun, Mon (half each)
Sun leftover -> Mon
Monday's leftover, and any Tuesday-Thursday leftover, is dropped. A day's
leftover is measured against its whole budget *including* what it inherited,
so time carried into Saturday and not played moves on to Sunday and Monday
rather than being lost.
The ledger lives on ``PlaytimeState`` -- root-owned and immutable -- and never
in ``playtime_history.json``: that file is user-writable, and a stored budget
read from it would let anyone grant themselves unlimited time.
Everything here is pure: it works on primitives handed to it, so it imports
nothing from the playtime modules and cannot close an import cycle.
"""
from __future__ import annotations
from datetime import date, timedelta
from typing import TYPE_CHECKING, Final
if TYPE_CHECKING:
from datetime import datetime
from steam_backlog_enforcer._playtime_state import PlaytimeState
# The gaming day starts here, local time. 05:59 belongs to the previous day.
_DAY_BOUNDARY_HOURS: Final = 6
# Weekday (Monday == 0) whose leftover is passed on -> how many later days of
# the Friday-Monday block share it.
_SHARES: Final = {4: 3, 5: 2, 6: 1}
def gaming_day_key(now: datetime) -> str:
"""Return the ``YYYY-MM-DD`` gaming day containing *now*.
The boundary is 06:00 local, so 02:00 on the 5th belongs to the 4th.
Args:
now: Timezone-aware local timestamp.
Returns:
The gaming day key.
"""
return (now - timedelta(hours=_DAY_BOUNDARY_HOURS)).date().isoformat()
def pass_on(
carry: dict[str, float],
*,
outgoing_day: str,
budget: float,
spent: float,
new_day: str,
) -> dict[str, float]:
"""Return the carry ledger as it stands once *new_day* begins.
Shares owed to days before *new_day* are dropped, which both bounds the
ledger and means a day the daemon never saw simply forfeits its share.
Args:
carry: Day key to seconds owed, as held by the outgoing day.
outgoing_day: The gaming day that just ended; ``""`` if none.
budget: The outgoing day's budget, carry included.
spent: Seconds billed during the outgoing day.
new_day: The gaming day now starting.
Returns:
The ledger for *new_day*, the outgoing day's leftover added.
"""
owed = {day: seconds for day, seconds in carry.items() if day >= new_day}
if not outgoing_day:
return owed
ended = date.fromisoformat(outgoing_day)
shares = _SHARES.get(ended.weekday(), 0)
leftover = max(0.0, budget - spent)
for offset in range(1, shares + 1):
target = (ended + timedelta(days=offset)).isoformat()
if target >= new_day:
owed[target] = owed.get(target, 0.0) + leftover / shares
return owed
def carry_due(
carry: dict[str, float],
*,
stored_day: str,
budget: float,
spent: float,
today: str,
) -> float:
"""Return the seconds carried into *today*.
The stored record may still describe yesterday -- the first tick after
06:00 resolves the budget before rolling the state over -- so the ledger is
projected forward exactly as ``roll_over`` will do it.
Args:
carry: The stored ledger.
stored_day: The gaming day the stored record covers.
budget: That day's budget, carry included.
spent: Seconds billed during that day.
today: The current gaming day.
Returns:
Seconds inherited by *today*, or 0.
"""
owed = (
carry
if stored_day == today
else pass_on(
carry, outgoing_day=stored_day, budget=budget, spent=spent, new_day=today
)
)
return owed.get(today, 0.0)
def carry_into(stored: PlaytimeState | None, now: datetime) -> float:
"""Return the seconds carried into the gaming day containing *now*.
Args:
stored: The persisted production state, or ``None`` if unreadable.
now: Local timestamp for this tick.
Returns:
The carry owed to today; 0 when no state is recorded.
"""
if stored is None:
return 0.0
return carry_due(
stored.carry,
stored_day=stored.day_key,
budget=stored.budget_seconds,
spent=stored.seconds,
today=gaming_day_key(now),
)

View File

@ -12,13 +12,13 @@ import os
from typing import Any
from steam_backlog_enforcer._budget_view import build_budget_snapshot
from steam_backlog_enforcer._gaming_days import gaming_day_key
from steam_backlog_enforcer._mcp_server import (
logger,
mcp,
)
from steam_backlog_enforcer._playtime import (
PlaytimeState,
gaming_day_key,
save_state,
)
from steam_backlog_enforcer._playtime_block import release_block
@ -56,6 +56,9 @@ def get_gaming_time() -> dict[str, Any]:
"seconds_used": today["seconds_used"],
"budget_seconds": today["budget_seconds"],
"seconds_remaining": today["seconds_remaining"],
# What earned budget_seconds, Friday-Monday carry-over included.
"budget_reason": rules["budget_reason"],
"carry_seconds": rules["carry_seconds"],
"next_warning_seconds": today["next_warning_seconds"],
"blocked": today["blocked"],
"enforcement": rules["enforcement"],

View File

@ -33,6 +33,7 @@ import logging
import time
from typing import TYPE_CHECKING
from steam_backlog_enforcer._gaming_days import gaming_day_key
from steam_backlog_enforcer._playtime_block import (
mounted_targets,
reconcile,
@ -50,7 +51,6 @@ from steam_backlog_enforcer._playtime_procs import attributed_key, qualifying_pi
from steam_backlog_enforcer._playtime_state import (
PlaytimeRules,
PlaytimeState,
gaming_day_key,
load_state,
rules_for,
save_state,
@ -93,6 +93,13 @@ def playtime_tick(
now = datetime.now(UTC).astimezone()
rules = rules_for(config, demo=demo)
state = roll_over(_state_or_recover(rules, now=now), day_key=gaming_day_key(now))
# Recorded every tick, total-block branch included: it is what the next
# 06:00 roll-over measures this day's leftover against. A high-water mark,
# because earners can reset at midnight -- before the gaming day ends --
# and a 05:59 dip must not erase the evening's earned-but-unspent time.
state = replace(
state, budget_seconds=max(state.budget_seconds, rules.budget_seconds)
)
if is_total_block_active():
# The total block runs `pacman -R steam` every tick; our bind mounts

View File

@ -11,6 +11,7 @@ from dataclasses import replace
import logging
from typing import TYPE_CHECKING
from steam_backlog_enforcer._gaming_days import pass_on
from steam_backlog_enforcer._playtime_state import (
_DELTA_CLAMP_FACTOR,
PlaytimeRules,
@ -79,7 +80,9 @@ def roll_over(state: PlaytimeState, *, day_key: str) -> PlaytimeState:
"""Reset *state* if the gaming day has changed.
Pure. ``last_tick_at`` carries across the boundary so the first tick of the
new day measures a sane delta rather than restarting from zero.
new day measures a sane delta rather than restarting from zero, and the carry
ledger crosses it with the outgoing day's leftover added (see
``_gaming_days``).
Args:
state: Current accounting state.
@ -90,7 +93,17 @@ def roll_over(state: PlaytimeState, *, day_key: str) -> PlaytimeState:
"""
if state.day_key == day_key:
return state
return PlaytimeState(day_key=day_key, last_tick_at=state.last_tick_at)
return PlaytimeState(
day_key=day_key,
last_tick_at=state.last_tick_at,
carry=pass_on(
state.carry,
outgoing_day=state.day_key,
budget=state.budget_seconds,
spent=state.seconds,
new_day=day_key,
),
)
def pending_warning(state: PlaytimeState, rules: PlaytimeRules) -> int | None:

View File

@ -1,19 +1,20 @@
"""The daily playtime budget: its state, its rules, and its persistence.
A gaming "day" runs 06:00-05:59 local, so late-night sessions count against
the day they started in. State is written atomically and re-read every tick,
which is what lets the enforcer survive a restart mid-session.
Gaming days and the Friday-Monday carry-over live in ``_gaming_days``. State
is written atomically and re-read every tick, which is what lets the enforcer
survive a restart mid-session.
"""
from __future__ import annotations
from dataclasses import dataclass, field, fields
from datetime import datetime, timedelta
from datetime import datetime
import json
import logging
from typing import TYPE_CHECKING
from steam_backlog_enforcer._budget_resolve import resolve_budget
from steam_backlog_enforcer._gaming_days import carry_into
from steam_backlog_enforcer._whitelist_locking import (
_try_set_immutable,
unlock_for_write,
@ -45,9 +46,6 @@ _SCHEMA_VERSION = 1
# being rewritten.
STATE_MODE = 0o644
# The gaming day starts here, local time. 05:59 belongs to the previous day.
_DAY_BOUNDARY_HOURS = 6
# Each tick's measured delta is capped at this many tick intervals. Two is the
# smallest value that tolerates one skipped tick without inflating the count.
_DELTA_CLAMP_FACTOR = 2.0
@ -96,6 +94,13 @@ class PlaytimeState:
"""
warned_seconds: list[int] = field(default_factory=list)
"""Warning thresholds already fired during ``day_key``."""
budget_seconds: float = 0.0
"""Highest budget in force during ``day_key``, carry included; ``0.0`` if unset.
What ``roll_over`` measures the outgoing day's leftover against.
"""
carry: dict[str, float] = field(default_factory=dict)
"""Day key to seconds carried into it from earlier days of the block."""
def is_blocked(self) -> bool:
"""Whether the cutoff has engaged for this gaming day.
@ -129,6 +134,7 @@ class PlaytimeRules:
leetcode_seconds: float = 0.0
reading_seconds: float = 0.0
budget_reason: str = ""
carry_seconds: float = 0.0
def rules_for(config: Config, *, demo: bool) -> PlaytimeRules:
@ -151,8 +157,14 @@ def rules_for(config: Config, *, demo: bool) -> PlaytimeRules:
# life) and _budget_view (which reloads Config per HTTP request) from
# reporting different budgets.
resolved = None if demo else resolve_budget(config)
# Carry sits on top of the 8h cap: it was earned and not spent, so it
# comes with no strings attached.
now = datetime.now().astimezone()
carry = 0.0 if demo else carry_into(load_state(demo=False), now)
return PlaytimeRules(
budget_seconds=_DEMO_BUDGET_SECONDS if resolved is None else resolved.seconds,
budget_seconds=_DEMO_BUDGET_SECONDS
if resolved is None
else resolved.seconds + carry,
warn_at=_DEMO_WARN_AT if demo else _WARN_AT,
sigkill_after=(_DEMO_SIGKILL_AFTER_SECONDS if demo else _SIGKILL_AFTER_SECONDS),
count_launchers=config.count_launcher_processes,
@ -164,24 +176,16 @@ def rules_for(config: Config, *, demo: bool) -> PlaytimeRules:
workout_seconds=0.0 if resolved is None else resolved.workout_seconds,
leetcode_seconds=0.0 if resolved is None else resolved.leetcode_seconds,
reading_seconds=0.0 if resolved is None else resolved.reading_seconds,
budget_reason=("demo run" if resolved is None else resolved.reason),
budget_reason=(
"demo run"
if resolved is None
else resolved.reason
+ (f", +{carry / 60:.0f}m carried over" if carry > 0 else "")
),
carry_seconds=carry,
)
def gaming_day_key(now: datetime) -> str:
"""Return the ``YYYY-MM-DD`` gaming day containing *now*.
The boundary is 06:00 local, so 02:00 on the 5th belongs to the 4th.
Args:
now: Timezone-aware local timestamp.
Returns:
The gaming day key.
"""
return (now - timedelta(hours=_DAY_BOUNDARY_HOURS)).date().isoformat()
def state_path(*, demo: bool) -> Path:
"""Return the state file for a demo or production run.

View File

@ -234,5 +234,6 @@ class TestBuildBudgetSnapshot:
"demo",
"masked_launchers",
"budget_reason",
"carry_seconds",
"bonuses",
}

View File

@ -0,0 +1,118 @@
"""Tests for the Friday-Monday carry-over in _gaming_days."""
from __future__ import annotations
from dataclasses import replace
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
from steam_backlog_enforcer._gaming_days import carry_due, carry_into, pass_on
from steam_backlog_enforcer._playtime_budget import roll_over
from steam_backlog_enforcer._playtime_state import PlaytimeState, rules_for
from steam_backlog_enforcer.config import Config
HOUR = 3600.0
LOCAL = timezone(timedelta(hours=2))
# 2026-10-01 is a Thursday.
THU, FRI, SAT, SUN, MON, TUE = (f"2026-10-{d:02d}" for d in range(1, 7))
def _leftover(day: str, *, budget: float, spent: float, new_day: str) -> dict:
return pass_on({}, outgoing_day=day, budget=budget, spent=spent, new_day=new_day)
class TestPassOn:
"""Who inherits a day's unspent budget."""
def test_friday_splits_three_ways(self) -> None:
owed = _leftover(FRI, budget=8 * HOUR, spent=7 * HOUR, new_day=SAT)
assert owed == {SAT: 1200.0, SUN: 1200.0, MON: 1200.0}
def test_saturday_splits_two_ways(self) -> None:
owed = _leftover(SAT, budget=8 * HOUR, spent=7 * HOUR, new_day=SUN)
assert owed == {SUN: 1800.0, MON: 1800.0}
def test_sunday_goes_to_monday(self) -> None:
assert _leftover(SUN, budget=HOUR, spent=0.0, new_day=MON) == {MON: HOUR}
def test_monday_and_weekdays_drop_their_leftover(self) -> None:
assert _leftover(MON, budget=HOUR, spent=0.0, new_day=TUE) == {}
assert _leftover(THU, budget=HOUR, spent=0.0, new_day=FRI) == {}
def test_overspent_day_carries_nothing(self) -> None:
owed = _leftover(SUN, budget=HOUR, spent=2 * HOUR, new_day=MON)
assert owed == {MON: 0.0}
def test_no_outgoing_day_keeps_only_future_shares(self) -> None:
owed = pass_on(
{FRI: 5.0, SUN: 7.0}, outgoing_day="", budget=0.0, spent=0.0, new_day=SAT
)
assert owed == {SUN: 7.0}
def test_a_skipped_day_forfeits_its_share(self) -> None:
owed = _leftover(FRI, budget=3 * HOUR, spent=0.0, new_day=SUN)
assert owed == {SUN: HOUR, MON: HOUR}
def test_shares_accumulate_with_earlier_ones(self) -> None:
owed = pass_on(
{SUN: 600.0, MON: 600.0},
outgoing_day=SAT,
budget=HOUR,
spent=0.0,
new_day=SUN,
)
assert owed == {SUN: 2400.0, MON: 2400.0}
class TestCarryDue:
"""Today's share, whether or not the stored record has rolled over."""
def test_same_day_reads_the_ledger(self) -> None:
due = carry_due({SAT: 60.0}, stored_day=SAT, budget=0.0, spent=0.0, today=SAT)
assert due == 60.0
def test_stale_record_is_projected_forward(self) -> None:
due = carry_due({}, stored_day=FRI, budget=4 * HOUR, spent=HOUR, today=SAT)
assert due == HOUR
class TestCarryInto:
"""Reading the carry off a stored state."""
def test_no_state_means_no_carry(self) -> None:
assert carry_into(None, datetime(2026, 10, 3, 10, tzinfo=LOCAL)) == 0.0
def test_reads_the_projected_share(self) -> None:
stored = PlaytimeState(day_key=FRI, seconds=HOUR, budget_seconds=4 * HOUR)
assert carry_into(stored, datetime(2026, 10, 3, 10, tzinfo=LOCAL)) == HOUR
class TestRollOverCarry:
"""roll_over threads the ledger across the 06:00 boundary."""
def test_unspent_carry_moves_on(self) -> None:
sat = roll_over(
PlaytimeState(day_key=FRI, seconds=7 * HOUR, budget_seconds=8 * HOUR),
day_key=SAT,
)
sat = replace(sat, seconds=8 * HOUR, budget_seconds=8 * HOUR + 1200.0)
sun = roll_over(sat, day_key=SUN)
assert sun.carry == {SUN: 1800.0, MON: 1800.0}
class TestRulesCarry:
"""rules_for adds the carry above the cap and says so."""
def test_carry_lands_in_budget_and_reason(self) -> None:
with patch(
"steam_backlog_enforcer._playtime_state.carry_into", return_value=1200.0
):
rules = rules_for(Config(), demo=False)
assert rules.carry_seconds == 1200.0
assert rules.budget_reason.endswith(", +20m carried over")
def test_demo_never_carries(self) -> None:
with patch(
"steam_backlog_enforcer._playtime_state.carry_into", return_value=1200.0
):
assert rules_for(Config(), demo=True).carry_seconds == 0.0

View File

@ -9,9 +9,9 @@ from unittest.mock import patch
import pytest
from steam_backlog_enforcer._gaming_days import gaming_day_key
from steam_backlog_enforcer._playtime_state import (
PlaytimeState,
gaming_day_key,
load_state,
rules_for,
save_state,

View File

@ -12,6 +12,13 @@ describe('BudgetRulesCard', () => {
expect(screen.getByText(/1h 0m, 30m, 10m, 5m remaining/)).toBeInTheDocument()
expect(screen.queryByText(/demo/)).toBeNull()
expect(screen.queryByText('Masked now')).toBeNull()
expect(screen.queryByText('Carried over')).toBeNull()
})
it('shows time carried over from earlier free days', () => {
render(<BudgetRulesCard rules={makeBudgetRules({ carry_seconds: 1200 })} />)
expect(screen.getByText('Carried over')).toBeInTheDocument()
expect(screen.getByText(/\+20m unspent from earlier free days/)).toBeInTheDocument()
})
it('shows the demo mode and masked launchers when they apply', () => {

View File

@ -16,6 +16,12 @@ export function BudgetRulesCard({ rules }: Props) {
<dl className="budget-rules-list">
<dt>Daily budget</dt>
<dd>{fmtDuration(rules.budget_seconds)}</dd>
{rules.carry_seconds > 0 && (
<>
<dt>Carried over</dt>
<dd>+{fmtDuration(rules.carry_seconds)} unspent from earlier free days</dd>
</>
)}
<dt>Enforcement</dt>
<dd>{onOff(rules.enforcement)}</dd>
<dt>Counts launchers</dt>

View File

@ -142,6 +142,7 @@ export function makeBudgetRules(over: Partial<BudgetRules> = {}): BudgetRules {
warn_at: [3600, 1800, 600, 300],
demo: false,
masked_launchers: [],
carry_seconds: 0,
...over,
}
}

View File

@ -173,6 +173,8 @@ export interface BudgetRules {
warn_at: number[]
demo: boolean
masked_launchers: string[]
/** Unspent Fri-Sun time inherited by today; already inside budget_seconds. */
carry_seconds: number
}
export interface BudgetSnapshot {