leisure kill: a reason's start time pins the process (PID reuse)

When a now.json reason carries the start time it saw (leisure-watch
0.2.1 `start`, /proc/<pid>/stat field 22), only the live process with
that start time is signalled; a reused pid is spared. Absent (0.2.0) or
unusable, the old behaviour stands.

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

steam_backlog_enforcer/_leisure_kill.py

@@ -18,10 +18,12 @@ Steam cutoff (:mod:`_playtime_cutoff`). Which pids:
 Each process gets SIGTERM once, then SIGKILL every tick from 30 s after *its*
 SIGTERM -- timed per process, so an app opened an hour into the block still
 gets its 30 s. A process is its (pid, start time), so a reused pid starts
-over. Root/system processes, lockers and the enforcer itself are never
-signalled, nor -- whatever ``now.json`` says -- the desktop session
-(:mod:`_leisure_never_kill`; each refusal logged once). Every SIGTERM and the
-first SIGKILL go to ``budget.jsonl``.
+over; a reason carrying the start time it saw (leisure-watch 0.2.1 ``start``)
+names only that process, never a later one reusing the pid. Root/system
+processes, lockers and the enforcer itself are never signalled, nor --
+whatever ``now.json`` says -- the desktop session (:mod:`_leisure_never_kill`;
+each refusal logged once). Every SIGTERM and the first SIGKILL go to
+``budget.jsonl``.
 """
 
 from __future__ import annotations
@@ -43,7 +45,7 @@ from steam_backlog_enforcer._leisure_procs import (
 )
 
 if TYPE_CHECKING:
-    from leisure_watch.facts import Now
+    from leisure_watch.facts import Now, Reason
 
     from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
     from steam_backlog_enforcer._shadow import ShadowVerdict
@@ -69,9 +71,9 @@ class KillLog(Protocol):
         """Append one record."""
 
 
-def fact_targets(fact: Now) -> dict[int, str]:
-    """Pids a trusted *fact* says must end, to the reason key that named them."""
-    found: dict[int, str] = {}
+def _picks(fact: Now) -> dict[int, Reason]:
+    """Pids a trusted *fact* says must end, to the (first) reason naming them."""
+    found: dict[int, Reason] = {}
     for reason in fact.reasons:
         if reason.pid is None:
             continue
@@ -79,10 +81,28 @@ def fact_targets(fact: Now) -> dict[int, str]:
         if failclosed or (
             reason.source in _KILLED_SOURCES and not is_browser(reason.pid)
         ):
-            found.setdefault(reason.pid, reason.key)
+            found.setdefault(reason.pid, reason)
     return found
 
 
+def fact_targets(fact: Now) -> dict[int, str]:
+    """Pids a trusted *fact* says must end, to the reason key that named them."""
+    return {pid: reason.key for pid, reason in _picks(fact).items()}
+
+
+def _start(reason: Reason) -> int | None:
+    # Optional from leisure-watch 0.2.1; absent (0.2.0) or unusable = unknown.
+    start = getattr(reason, "start", None)
+    usable = isinstance(start, int) and not isinstance(start, bool)
+    return start if usable else None
+
+
+def fact_starts(fact: Now) -> dict[int, int]:
+    """Per target pid, the start time its reason saw (when it carries one)."""
+    starts = {pid: _start(reason) for pid, reason in _picks(fact).items()}
+    return {pid: start for pid, start in starts.items() if start is not None}
+
+
 class LeisureKiller:
     """The per-daemon kill state: when each process got its SIGTERM."""
 
@@ -143,13 +163,15 @@ class LeisureKiller:
             return
         if verdict.paused_by is not None:
             return
+        starts: dict[int, int] = {}
         if verdict.fact is None:
             if untrusted_for < UNTRUSTED_SWEEP_AFTER:
                 return
             targets, why = sweep_targets(), "now.json untrusted"
         else:
             targets, why = fact_targets(verdict.fact), "now.json"
-        live = self._identify(targets)
+            starts = fact_starts(verdict.fact)
+        live = self._identify(targets, starts)
         self._termed = {i: t for i, t in self._termed.items() if i in live}
         self._killed &= set(live)
         base = {
@@ -179,13 +201,17 @@ class LeisureKiller:
             self._untrusted_since = stamp
         return stamp - self._untrusted_since
 
-    def _identify(self, targets: dict[int, str]) -> dict[_Ident, str]:
+    def _identify(
+        self, targets: dict[int, str], starts: dict[int, int]
+    ) -> dict[_Ident, str]:
         chain = own_chain()
         live: dict[_Ident, str] = {}
         refused: set[_Ident] = set()
         for pid, key in targets.items():
             started = start_time(pid)
-            if started is None or is_protected(pid, chain):
+            if started is None or starts.get(pid, started) != started:
+                continue  # gone, or the pid now belongs to another process
+            if is_protected(pid, chain):
                 continue
             guard = never_kill(pid)
             if guard is None:

steam_backlog_enforcer/tests/test_leisure_kill_start.py

@@ -0,0 +1,89 @@
+"""PID reuse: a reason carrying its process start time kills only that process.
+
+leisure-watch 0.2.1 adds an optional ``start`` (``/proc/<pid>/stat`` field 22)
+to each reason; :class:`_R` stands in for it, so this runs on 0.2.0 too.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import os
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+from leisure_watch.facts import Focus, Now, Reason
+import pytest
+
+from steam_backlog_enforcer._leisure_kill import LeisureKiller, fact_starts
+from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
+from steam_backlog_enforcer._shadow import ShadowVerdict
+from steam_backlog_enforcer.tests._isolate_leisure import leisure_proc
+from steam_backlog_enforcer.tests._leisure_fake_proc import FakeProc, add_proc
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+_RULES = PlaytimeRules(
+    budget_seconds=100.0,
+    warn_at=(),
+    sigkill_after=30.0,
+    count_launchers=False,
+    enforcement=True,
+    demo=False,
+    leisure=True,
+)
+_SPENT = PlaytimeState(day_key="2026-11-02", seconds=100.0)
+
+
+@dataclass(frozen=True)
+class _R(Reason):
+    start: int | None = None
+
+
+def _killed(tmp_path: Path, *reasons: Reason) -> list[int]:
+    root = leisure_proc(tmp_path)
+    add_proc(root, FakeProc(10, exe="/usr/bin/mpv", start=500))
+    fact = Now(0.0, 1, "leisure", None, reasons, Focus("mpv", "", None, idle=False))
+    verdict = ShadowVerdict(paused_by=None, keys=("k",), fact=fact)
+    with patch.object(os, "kill") as kill:
+        LeisureKiller(None).enforce(_SPENT, _RULES, verdict=verdict, stamp=0.0)
+    return [c.args[0] for c in kill.call_args_list]
+
+
+def test_a_matching_start_is_killed(tmp_path: Path) -> None:
+    assert _killed(tmp_path, _R("media", "app:mpv", 10, "mpv", start=500)) == [10]
+
+
+def test_a_reused_pid_is_spared(tmp_path: Path) -> None:
+    assert _killed(tmp_path, _R("media", "app:mpv", 10, "mpv", start=499)) == []
+
+
+@pytest.mark.parametrize(
+    "reason",
+    [
+        Reason("media", "app:mpv", 10, "mpv"),
+        _R("media", "app:mpv", 10, "mpv", start=None),
+        _R("media", "app:mpv", 10, "mpv", start=True),
+    ],
+)
+def test_no_usable_start_keeps_the_old_behaviour(
+    tmp_path: Path, reason: Reason
+) -> None:
+    assert _killed(tmp_path, reason) == [10]
+
+
+def test_starts_come_from_the_reason_that_named_the_pid() -> None:
+    fact = Now(
+        0.0,
+        1,
+        "leisure",
+        None,
+        (
+            _R("game", "steam:1", 10, "x", start=1),
+            _R("media", "app:mpv", 10, "mpv", start=500),
+            _R("focus", "app:mpv", 10, "mpv", start=7),
+            _R("media", "app:vlc", None, "vlc", start=9),
+        ),
+        Focus("mpv", "", None, idle=False),
+    )
+    assert fact_starts(fact) == {10: 500}