fix(wellbeing): split focus_mode_daemon to satisfy the 500-line limit

focus_mode_daemon.py has been 509 lines since 22e22c1, over the repo's
cap. It went unnoticed because the Python tests workflow only triggers on
python_pkg/** changes, so that commit never ran CI; the next commit to
touch python_pkg/ exposed it and turned main red.

Extract configuration, logging, notification, /proc process inspection and
the killers into _focus_mode_lib.py, leaving the daemon its policy: the
whitelist, the FocusMode state machine, and main(). log/notify move with
the process code that calls them, so the dependency runs one way and there
is no cycle. The helper is underscore-prefixed and registered in pylint's
init-hook path, mirroring the existing usage_report.py / _usage_report_*.py
sibling-import precedent.

The daemon runs from a symlink in /usr/local/bin, so a sibling import only
works if Python resolves that symlink before setting sys.path[0]; verified
that it does, and the installer symlinks rather than copies, so the sibling
travels with it. Confirmed against the running service: restarts clean and
resumes enforcing, process detection unchanged, status output matches the
pre-split version.

Structural only; no behaviour change.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RG3eEn8rwZbcXaJrKJxdv8
This commit is contained in:
Krzysztof Rudnicki 2026-07-17 17:36:30 +02:00
parent 69242e4bb3
commit e0f62d699a
6 changed files with 311 additions and 206 deletions

View File

@ -0,0 +1,17 @@
{
"title": "focus_mode_daemon: bring back under the 500-line limit",
"objective": "focus_mode_daemon.py has been 509 lines since commit 22e22c1, breaching the repo's 500-line cap. It went unnoticed because the Python-tests workflow only triggers on python_pkg/** changes, so that commit never ran CI; the first later commit touching python_pkg/ exposed it and main is red. Split the file along a real seam with no behaviour change, and keep the live focus-mode.service working throughout.",
"acceptance_criteria": [
"No tracked Python file exceeds 500 lines",
"The split follows a real responsibility boundary, with the dependency running one way and no import cycle",
"The daemon still runs through /usr/local/bin/focus-mode-daemon, the symlink systemd uses",
"focus-mode.service restarts cleanly and resumes enforcing Steam/browser mutual exclusion",
"Process detection behaviour is unchanged",
"No behaviour change of any kind: this is a structural split"
],
"out_of_scope": [
"Any change to the daemon's mutual-exclusion policy, whitelist behaviour, or poll interval",
"The brother_printer work in a12e696/69242e4"
],
"verifier": "ruff check; /usr/local/bin/focus-mode-daemon --help and status compared against the pre-split version from git; systemctl --user restart focus-mode.service plus journal inspection; wc -l over tracked *.py; CI (Python tests)"
}

View File

@ -0,0 +1,49 @@
{
"intent": "Get main's CI green by bringing focus_mode_daemon.py (509 lines) under the repo's 500-line limit, without changing the behaviour of a daemon that is actively running.",
"scope": [
"linux_configuration/scripts/periodic_background/digital_wellbeing/focus_mode_daemon.py: 509 -> 320 lines",
"linux_configuration/.../_focus_mode_lib.py: new, 227 lines",
"linux_configuration/.../__init__.py: new, following the existing system-maintenance/bin precedent",
"Non-goal: any behavioural change to focus mode"
],
"changes": [
"Extract configuration, logging, desktop notification, /proc process inspection, and the process killers into _focus_mode_lib.py. The daemon keeps the policy: whitelist handling, the FocusMode state machine, status output, and main(). The dependency runs one way (daemon imports lib), so there is no cycle - which is why log/notify moved with the process code that calls them rather than being left behind.",
"Add __init__.py to satisfy ruff INP001, matching the existing linux_configuration/scripts/periodic_background/system-maintenance/bin/__init__.py precedent."
],
"verification": [
{
"command": "Symlink resolution test before relying on a sibling import",
"result": "pass",
"evidence": "Ran a script through a symlink: sys.path[0] resolved to the real directory and the sibling import succeeded. This matters because systemd runs the daemon via /usr/local/bin/focus-mode-daemon -> repo path, and install_focus_mode_daemon.sh uses ln -sfn (not cp), so the sibling is always reachable on fresh installs too."
},
{
"command": "/usr/local/bin/focus-mode-daemon --help / status (the exact symlink systemd uses)",
"result": "pass",
"evidence": "Argument parsing and subcommands work through the symlink. 'status' output compared byte-for-byte against the pre-split version from git HEAD: both silent with no whitelist active, so the silence is pre-existing behaviour, not a regression."
},
{
"command": "focus_mode_lib.get_running_processes / is_steam_running / is_browser_running",
"result": "pass",
"evidence": "455 processes read from /proc; browser correctly detected as running, Steam as not - matching the machine's actual state."
},
{
"command": "systemctl --user restart focus-mode.service",
"result": "pass",
"evidence": "active (running); journal shows 'Focus Mode Daemon starting...' then 'Browser detected - entering BROWSING mode'. No import errors."
},
{
"command": "ruff check + wc -l over tracked *.py",
"result": "pass",
"evidence": "All checks passed. No tracked Python file exceeds 500 lines."
}
],
"risks": [
"The daemon now depends on a sibling module at runtime. Verified through the real symlink and a service restart; the installer symlinks rather than copies, so the sibling travels with it.",
"Structural split only - no logic changed, no function bodies edited. Behaviour compared against the pre-split version where observable.",
"This file has no unit tests, so verification is by running the real daemon rather than by test coverage."
],
"rollback": [
"git revert <commit>, then systemctl --user restart focus-mode.service.",
"After rollback, validate the service is active and the 500-line test fails again on focus_mode_daemon.py."
]
}

View File

@ -0,0 +1 @@
"""Focus mode daemon: Steam/browser mutual exclusion."""

View File

@ -0,0 +1,227 @@
"""Shared pieces of the focus-mode daemon: config, logging, process inspection.
Split out of focus_mode_daemon.py to keep both files within the repo's 500-line
limit. This half is everything the daemon observes or acts on - where state
lives, which processes count as Steam or a browser, how they are read from /proc
and how they are killed. The daemon half owns the policy: whitelist handling and
the mutual-exclusion decisions.
The dependency runs one way (daemon imports lib) so there is no import cycle.
Sibling imports resolve because Python resolves the symlink used to run the
daemon before setting sys.path[0].
"""
from __future__ import annotations
import contextlib
from datetime import datetime, timezone
import logging
from pathlib import Path
import shutil
import subprocess
import time
logger = logging.getLogger(__name__)
# Configuration
STATE_DIR = Path.home() / ".local" / "state" / "focus-mode"
LOG_FILE = STATE_DIR / "focus-mode.log"
WHITELIST_FILE = STATE_DIR / "whitelist"
POLL_INTERVAL = 2 # seconds between process checks
DEFAULT_WHITELIST_MINUTES = 5
# /proc/PID/stat state character for a dead-but-unreaped process.
ZOMBIE_STATE = "Z"
# Process patterns
STEAM_PATTERNS = frozenset(
[
"steam",
"steamwebhelper",
"steam_ocompati", # Proton compatibility tool
]
)
# Games often have steam_app_ prefix in process name
STEAM_GAME_PREFIX = "steam_app_"
BROWSER_PATTERNS = frozenset(
[
"firefox",
"firefox-esr",
"librewolf",
"chromium",
"chrome",
"google-chrome",
"brave",
"vivaldi",
"opera",
"microsoft-edge",
"ungoogled-chromium",
"thorium",
]
)
# Electron apps that should NOT be treated as browsers
# These use Chromium under the hood but are not web browsers
ELECTRON_IGNORE = frozenset(
[
"electron",
"code", # VS Code
"chrome_crashpad", # Crashpad handler used by all Electron apps
]
)
# Patterns to ignore (browser helpers that aren't the main browser)
IGNORE_PATTERNS = frozenset(
[
"crashhandler",
"update",
"helper",
"crashpad",
]
)
def log(message: str) -> None:
"""Log message with timestamp."""
timestamp = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
log_line = f"{timestamp} - {message}"
logger.info("%s", log_line)
with contextlib.suppress(OSError):
STATE_DIR.mkdir(parents=True, exist_ok=True)
with LOG_FILE.open("a") as f:
f.write(log_line + "\n")
def notify(title: str, message: str, urgency: str = "normal") -> None:
"""Send desktop notification."""
notify_send = shutil.which("notify-send")
if notify_send is None:
return
with contextlib.suppress(OSError, subprocess.SubprocessError):
subprocess.run(
[notify_send, "-u", urgency, title, message],
capture_output=True,
timeout=5,
check=False,
)
def _parse_stat(stat_line: str) -> tuple[str, str] | None:
"""Extract the (name, state) pair from a /proc/PID/stat line.
The stat format is ``pid (comm) state ...``. The comm field may itself
contain spaces and ')' characters, so the name is everything between the
first '(' and the *last* ')' - splitting on whitespace would corrupt any
process whose name embeds either.
Args:
stat_line: Raw contents of a /proc/PID/stat file.
Returns:
A (name, state) tuple, or None if the line is malformed.
"""
open_paren = stat_line.find("(")
before_close, sep, after_close = stat_line.rpartition(")")
if open_paren == -1 or not sep:
return None
state_fields = after_close.split()
if not state_fields:
return None
return before_close[open_paren + 1 :], state_fields[0]
def get_running_processes() -> set[str]:
"""Get set of currently running process names by reading /proc directly.
Reads /proc/*/stat rather than /proc/*/comm: both cost one read per PID
(so the poll loop stays fork-free), but stat also carries the process
state. That distinction matters because a dead process keeps its name
until the parent reaps it, so a crashed "steam" lingers as a zombie and
would otherwise be counted as a running Steam. Zombies are skipped.
Names are truncated to 15 characters here exactly as they are in
/proc/*/comm, so the module's process patterns match unchanged.
"""
processes: set[str] = set()
try:
for stat_path in Path("/proc").glob("*/stat"):
with contextlib.suppress(OSError):
parsed = _parse_stat(stat_path.read_text())
if parsed is None:
continue
proc_name, state = parsed
if state == ZOMBIE_STATE or not proc_name:
continue
processes.add(proc_name.lower())
except OSError as exc:
log(f"Error reading /proc: {exc}")
return processes
def is_steam_running(processes: set[str]) -> bool:
"""Check if Steam or any Steam game is running."""
for proc in processes:
if proc in STEAM_PATTERNS:
return True
if proc.startswith(STEAM_GAME_PREFIX):
return True
return False
def is_browser_running(processes: set[str]) -> bool:
"""Check if any browser is running."""
for proc in processes:
if proc in ELECTRON_IGNORE:
continue
if any(ign in proc for ign in IGNORE_PATTERNS):
continue
if proc in BROWSER_PATTERNS:
return True
return False
def _run_pkill(pattern: str, *, force: bool = False) -> None:
"""Run pkill with the given pattern."""
pkill_bin = shutil.which("pkill")
if pkill_bin is None:
return
cmd = [pkill_bin]
if force:
cmd.append("-9")
cmd.extend(["-f", pattern])
with contextlib.suppress(OSError, subprocess.SubprocessError):
subprocess.run(cmd, capture_output=True, timeout=5, check=False)
def kill_steam() -> None:
"""Kill all Steam-related processes."""
log("Killing Steam processes...")
notify(
"\U0001f3ae Gaming Blocked",
"Browser is active. Closing Steam.",
"critical",
)
_run_pkill("steam")
time.sleep(2)
_run_pkill("steam", force=True)
def kill_browsers() -> None:
"""Kill all browser processes."""
log("Killing browser processes...")
notify(
"\U0001f310 Browsers Blocked",
"Steam is active. Closing browsers.",
"critical",
)
for browser in BROWSER_PATTERNS:
_run_pkill(browser)
time.sleep(2)
for browser in BROWSER_PATTERNS:
_run_pkill(browser, force=True)

View File

@ -14,89 +14,33 @@ import argparse
import contextlib
from datetime import datetime, timedelta, timezone
import logging
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import time
from typing import TYPE_CHECKING
# Sibling module: resolves because Python resolves the symlink used to launch
# this daemon before setting sys.path[0].
from _focus_mode_lib import (
DEFAULT_WHITELIST_MINUTES,
POLL_INTERVAL,
STATE_DIR,
WHITELIST_FILE,
get_running_processes,
is_browser_running,
is_steam_running,
kill_browsers,
kill_steam,
log,
notify,
)
if TYPE_CHECKING:
from types import FrameType
logger = logging.getLogger(__name__)
# Configuration
STATE_DIR = Path.home() / ".local" / "state" / "focus-mode"
LOG_FILE = STATE_DIR / "focus-mode.log"
WHITELIST_FILE = STATE_DIR / "whitelist"
POLL_INTERVAL = 2 # seconds between process checks
DEFAULT_WHITELIST_MINUTES = 5
# /proc/PID/stat state character for a dead-but-unreaped process.
ZOMBIE_STATE = "Z"
# Process patterns
STEAM_PATTERNS = frozenset(
[
"steam",
"steamwebhelper",
"steam_ocompati", # Proton compatibility tool
]
)
# Games often have steam_app_ prefix in process name
STEAM_GAME_PREFIX = "steam_app_"
BROWSER_PATTERNS = frozenset(
[
"firefox",
"firefox-esr",
"librewolf",
"chromium",
"chrome",
"google-chrome",
"brave",
"vivaldi",
"opera",
"microsoft-edge",
"ungoogled-chromium",
"thorium",
]
)
# Electron apps that should NOT be treated as browsers
# These use Chromium under the hood but are not web browsers
ELECTRON_IGNORE = frozenset(
[
"electron",
"code", # VS Code
"chrome_crashpad", # Crashpad handler used by all Electron apps
]
)
# Patterns to ignore (browser helpers that aren't the main browser)
IGNORE_PATTERNS = frozenset(
[
"crashhandler",
"update",
"helper",
"crashpad",
]
)
def log(message: str) -> None:
"""Log message with timestamp."""
timestamp = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
log_line = f"{timestamp} - {message}"
logger.info("%s", log_line)
with contextlib.suppress(OSError):
STATE_DIR.mkdir(parents=True, exist_ok=True)
with LOG_FILE.open("a") as f:
f.write(log_line + "\n")
def is_whitelist_active() -> bool:
"""Check if the browser whitelist is currently active."""
@ -150,139 +94,6 @@ def cancel_whitelist() -> None:
log("No active whitelist to cancel")
def notify(title: str, message: str, urgency: str = "normal") -> None:
"""Send desktop notification."""
notify_send = shutil.which("notify-send")
if notify_send is None:
return
with contextlib.suppress(OSError, subprocess.SubprocessError):
subprocess.run(
[notify_send, "-u", urgency, title, message],
capture_output=True,
timeout=5,
check=False,
)
def _parse_stat(stat_line: str) -> tuple[str, str] | None:
"""Extract the (name, state) pair from a /proc/PID/stat line.
The stat format is ``pid (comm) state ...``. The comm field may itself
contain spaces and ')' characters, so the name is everything between the
first '(' and the *last* ')' - splitting on whitespace would corrupt any
process whose name embeds either.
Args:
stat_line: Raw contents of a /proc/PID/stat file.
Returns:
A (name, state) tuple, or None if the line is malformed.
"""
open_paren = stat_line.find("(")
before_close, sep, after_close = stat_line.rpartition(")")
if open_paren == -1 or not sep:
return None
state_fields = after_close.split()
if not state_fields:
return None
return before_close[open_paren + 1 :], state_fields[0]
def get_running_processes() -> set[str]:
"""Get set of currently running process names by reading /proc directly.
Reads /proc/*/stat rather than /proc/*/comm: both cost one read per PID
(so the poll loop stays fork-free), but stat also carries the process
state. That distinction matters because a dead process keeps its name
until the parent reaps it, so a crashed "steam" lingers as a zombie and
would otherwise be counted as a running Steam. Zombies are skipped.
Names are truncated to 15 characters here exactly as they are in
/proc/*/comm, so the module's process patterns match unchanged.
"""
processes: set[str] = set()
try:
for stat_path in Path("/proc").glob("*/stat"):
with contextlib.suppress(OSError):
parsed = _parse_stat(stat_path.read_text())
if parsed is None:
continue
proc_name, state = parsed
if state == ZOMBIE_STATE or not proc_name:
continue
processes.add(proc_name.lower())
except OSError as exc:
log(f"Error reading /proc: {exc}")
return processes
def is_steam_running(processes: set[str]) -> bool:
"""Check if Steam or any Steam game is running."""
for proc in processes:
if proc in STEAM_PATTERNS:
return True
if proc.startswith(STEAM_GAME_PREFIX):
return True
return False
def is_browser_running(processes: set[str]) -> bool:
"""Check if any browser is running."""
for proc in processes:
if proc in ELECTRON_IGNORE:
continue
if any(ign in proc for ign in IGNORE_PATTERNS):
continue
if proc in BROWSER_PATTERNS:
return True
return False
def _run_pkill(pattern: str, *, force: bool = False) -> None:
"""Run pkill with the given pattern."""
pkill_bin = shutil.which("pkill")
if pkill_bin is None:
return
cmd = [pkill_bin]
if force:
cmd.append("-9")
cmd.extend(["-f", pattern])
with contextlib.suppress(OSError, subprocess.SubprocessError):
subprocess.run(cmd, capture_output=True, timeout=5, check=False)
def kill_steam() -> None:
"""Kill all Steam-related processes."""
log("Killing Steam processes...")
notify(
"\U0001f3ae Gaming Blocked",
"Browser is active. Closing Steam.",
"critical",
)
_run_pkill("steam")
time.sleep(2)
_run_pkill("steam", force=True)
def kill_browsers() -> None:
"""Kill all browser processes."""
log("Killing browser processes...")
notify(
"\U0001f310 Browsers Blocked",
"Steam is active. Closing browsers.",
"critical",
)
for browser in BROWSER_PATTERNS:
_run_pkill(browser)
time.sleep(2)
for browser in BROWSER_PATTERNS:
_run_pkill(browser, force=True)
class FocusMode:
"""Tracks current focus mode and enforces mutual exclusion."""

View File

@ -170,7 +170,7 @@ disallow_untyped_decorators = false
# bare name (the same dirs linux_configuration/tests/conftest.py adds to
# sys.path at runtime) resolve under static analysis instead of raising E0401.
# Paths are relative to the repo root, which is pre-commit's working directory.
init-hook = "import sys; sys.path[:0] = ['meta/scripts', 'python_pkg', 'phone_focus_mode', 'phone_focus_mode/lib', 'linux_configuration/scripts/single_use/utils', 'linux_configuration/scripts/periodic_background/system-maintenance/bin', 'linux_configuration/scripts/single_use/misc/testsAndMisc-bash/tools']"
init-hook = "import sys; sys.path[:0] = ['meta/scripts', 'python_pkg', 'phone_focus_mode', 'phone_focus_mode/lib', 'linux_configuration/scripts/single_use/utils', 'linux_configuration/scripts/periodic_background/system-maintenance/bin', 'linux_configuration/scripts/periodic_background/digital_wellbeing', 'linux_configuration/scripts/single_use/misc/testsAndMisc-bash/tools']"
# Analyse import fallback blocks
analyse-fallback-blocks = true
# Pickle collected data for later comparisons