end-session: one click to work mode without losing a take

The tray's Quit only closed the tray, and music mode could not be left while
SuperDirt (a headless unit) or Ardour ran, so 'I quit everything' was false.
This stops the SuperDirt watchdog first (or it restarts the sound), then
SuperDirt (the black box stops with it), then hands over to music-work off.
It refuses while a take is rolling, never closes Ardour (it may hold unsaved
work; a second run finishes once Ardour is closed), and reports every outcome
as a desktop notification.

tools/end_session.py

+#!/usr/bin/env python3
+"""end-session — one click from "the rig is up" to "work mode", without losing a take.
+
+WHY (2026-10-11): the tray's Quit only closed the tray, the ParVagues tile could not
+switch music mode off because `music-work off` refuses while SuperCollider or Ardour
+runs, and SuperDirt is a HEADLESS unit with no window to close. So a user who had "quit
+everything" was still in music mode with the rig up, and got alarms about a rig they
+believed was down. There was no single honest way to say "I am done".
+
+WHAT IT DOES, in this order, and why the order matters
+  1. Refuses if a take is rolling (Ardour is writing stem files). A session ended under
+     a rolling take is the one mistake this tool must not make.
+  2. Stops the SuperDirt watchdog FIRST. Its whole job is to restart SuperDirt when
+     scsynth dies, so stopping SuperDirt first just makes it come back.
+  3. Stops SuperDirt. The black box (gig-record) is PartOf it and stops with it.
+  4. If Ardour is closed: hands over to `music-work off`, which stops the helpers and
+     switches music mode off. The helper list lives in ONE place (music-work).
+     If Ardour is still open: it does NOT close it. Ardour may hold unsaved work and the
+     launcher deliberately cannot close it. It says so; click End session again after
+     you have saved and closed Ardour, and the second run finishes the job. Idempotent,
+     no background waiter to forget about (an "on next X do Y" watcher is how loaded
+     guns are made).
+
+EXIT CODES  0 = work mode;  2 = refused (a take is rolling);  3 = SuperDirt stopped but
+Ardour is still open, so music mode stays on;  1 = something failed.
+"""
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+HOME = Path.home()
+ARDOUR_SESSION_DIR = Path(os.environ.get(
+    "PARVAGUES_ARDOUR_SESSION_DIR", HOME / "Work/Sound/Ardour/Tidal Live"))
+MUSIC_WORK = Path(os.environ.get("MUSIC_WORK", HOME / ".local/bin/music-work"))
+SC_PROCS = ("sclang", "scsynth", "supernova")
+ARDOUR_PROCS = ("ArdourGUI", "ardour")
+ROLLING_WINDOW_S = 6.0     # a stem file touched this recently means Ardour is recording
+
+
+def comm_running(names, proc_root: Path = Path("/proc")) -> bool:
+    """Exact process-name match from /proc, no pgrep (never identify a process by a
+    pattern that can match the tool doing the asking)."""
+    for p in proc_root.glob("[0-9]*/comm"):
+        try:
+            if p.read_text().strip() in names:
+                return True
+        except OSError:
+            continue
+    return False
+
+
+def take_rolling(session_dir: Path = ARDOUR_SESSION_DIR, now=time.time) -> bool:
+    """Ardour writes one capture file per armed track while recording, so a stem file
+    modified in the last few seconds is a take in progress (the same lens take-rec and
+    gig-log use: the files, never a UI's opinion)."""
+    newest = 0.0
+    for wav in session_dir.glob("interchange/*/audiofiles/*.wav"):
+        try:
+            newest = max(newest, wav.stat().st_mtime)
+        except OSError:
+            continue
+    return bool(newest) and (now() - newest) <= ROLLING_WINDOW_S
+
+
+def sh(*argv) -> subprocess.CompletedProcess:
+    return subprocess.run(argv, capture_output=True, text=True)
+
+
+def say(msg: str) -> None:
+    print(msg, flush=True)
+
+
+def desktop_notify(msg: str) -> None:
+    """The outcome, where the person is looking. The caller may be the tray (which this
+    very script stops at the end), the shell tile, or a terminal: a notification works for
+    all three, a return value does not."""
+    import shutil
+    if shutil.which("notify-send") is None:
+        return
+    try:
+        subprocess.Popen(["notify-send", "-a", "ParVagues", "-t", "9000", "End session", msg],
+                         stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    except OSError:
+        pass
+
+
+def end(*, force: bool = False, running=comm_running, rolling=take_rolling, run=sh,
+        wait_s: float = 20.0, sleep=time.sleep, notify=lambda m: None) -> int:
+    def done(code: int, msg: str) -> int:
+        say(msg)
+        notify(msg)
+        return code
+
+    if rolling() and not force:
+        return done(2, "A take is rolling (Ardour is writing stems). Stop the take first; nothing was touched.")
+
+    say("Stopping the SuperDirt watchdog (it would restart the sound)...")
+    run("systemctl", "--user", "stop", "parvagues-sc-watchdog.service")
+    say("Stopping SuperDirt (the black box stops with it)...")
+    r = run("systemctl", "--user", "stop", "parvagues-sc.service")
+    if r.returncode != 0:
+        return done(1, f"Could not stop SuperDirt: {(r.stderr or '').strip()[:160]}")
+    waited = 0.0
+    while running(SC_PROCS) and waited < wait_s:
+        sleep(0.5)
+        waited += 0.5
+    if running(SC_PROCS):
+        return done(1, "SuperCollider is still running after the unit stopped; leaving music mode on.")
+
+    if running(ARDOUR_PROCS):
+        return done(3, "SuperDirt is stopped. Ardour is still open, so music mode stays on: save and "
+                       "close Ardour, then click End session again to finish.")
+
+    say("Stopping the helpers and leaving music mode...")
+    r = run(str(MUSIC_WORK), "off")
+    if r.returncode != 0:
+        return done(1, (r.stderr or r.stdout or "music-work off failed").strip().splitlines()[-1])
+    return done(0, "Session ended: work mode.")
+
+
+def dry_run(*, force: bool = False, running=comm_running, rolling=take_rolling) -> int:
+    """Say what `end` would do, change nothing, return the code it would return."""
+    if rolling() and not force:
+        say("would refuse: a take is rolling (Ardour is writing stems).")
+        return 2
+    say("would stop: parvagues-sc-watchdog, then parvagues-sc (the black box stops with it)")
+    if running(ARDOUR_PROCS):
+        say("would then STOP SHORT: Ardour is open, so music mode stays on "
+            "(save and close Ardour, then run it again).")
+        return 3
+    say("would then run `music-work off` (helpers stop, music mode off): work mode.")
+    return 0
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = sys.argv[1:] if argv is None else argv
+    if "--dry-run" in args:
+        return dry_run(force="--force" in args)
+    return end(force="--force" in args, notify=desktop_notify)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tools/tests/test_end_session.py

+"""End session: one click to work mode, never at the cost of a take or of Ardour's work.
+
+The branches that matter: a rolling take refuses, the watchdog dies BEFORE SuperDirt (or
+it restarts the sound), Ardour is never closed by this tool (it may hold unsaved work),
+and a run that could only half finish says so and finishes on the second click.
+"""
+from __future__ import annotations
+
+import os
+import pathlib
+import subprocess
+import sys
+import time
+
+TOOLS = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(TOOLS))
+import end_session as es  # noqa: E402
+
+
+class Fake:
+    def __init__(self, sc=False, ardour=False, rolling=False, stop_rc=0, off_rc=0):
+        self.sc, self.ardour, self.rolling = sc, ardour, rolling
+        self.stop_rc, self.off_rc = stop_rc, off_rc
+        self.calls: list[tuple] = []
+
+    def running(self, names):
+        if set(names) & set(es.SC_PROCS):
+            return self.sc
+        return self.ardour
+
+    def run(self, *argv):
+        self.calls.append(argv)
+        if argv[:3] == ("systemctl", "--user", "stop") and argv[3] == "parvagues-sc.service":
+            if self.stop_rc == 0:
+                self.sc = False
+            return subprocess.CompletedProcess(argv, self.stop_rc, "", "boom" if self.stop_rc else "")
+        if argv[-1:] == ("off",):
+            return subprocess.CompletedProcess(argv, self.off_rc, "", "refused" if self.off_rc else "")
+        return subprocess.CompletedProcess(argv, 0, "", "")
+
+    def go(self, **kw):
+        return es.end(running=self.running, rolling=lambda: self.rolling, run=self.run,
+                      sleep=lambda s: None, wait_s=1.0, **kw)
+
+    def stops(self):
+        return [c[3] for c in self.calls if c[:3] == ("systemctl", "--user", "stop")]
+
+
+def test_a_rolling_take_refuses_and_touches_nothing(capsys):
+    f = Fake(sc=True, rolling=True)
+    assert f.go() == 2
+    assert f.calls == []
+    assert "take is rolling" in capsys.readouterr().out
+
+
+def test_force_overrides_a_rolling_take():
+    f = Fake(sc=True, rolling=True)
+    assert f.go(force=True) == 0
+    assert "parvagues-sc.service" in f.stops()
+
+
+def test_the_watchdog_is_stopped_before_superdirt():
+    """The watchdog restarts SuperDirt when scsynth dies; stop SuperDirt first and it comes back."""
+    f = Fake(sc=True)
+    f.go()
+    s = f.stops()
+    assert s.index("parvagues-sc-watchdog.service") < s.index("parvagues-sc.service")
+
+
+def test_everything_closed_ends_in_work_mode_via_music_work_last(capsys):
+    f = Fake(sc=True)
+    assert f.go() == 0
+    assert f.calls[-1][-1] == "off" and str(es.MUSIC_WORK) in f.calls[-1][0]
+    assert "work mode" in capsys.readouterr().out
+
+
+def test_ardour_is_never_closed_by_this_tool_and_music_mode_stays_on(capsys):
+    """Ardour may hold unsaved work, and the launcher deliberately cannot close it."""
+    f = Fake(sc=True, ardour=True)
+    assert f.go() == 3
+    assert not any(c[-1:] == ("off",) for c in f.calls)               # music mode untouched
+    assert not any("ardour" in " ".join(c).lower() and c[2] == "stop" for c in f.calls)
+    assert not any(c[0] == "kill" or "TERM" in c for c in f.calls)
+    out = capsys.readouterr().out
+    assert "Ardour is still open" in out and "click End session again" in out
+
+
+def test_the_second_click_finishes_once_ardour_is_closed():
+    f = Fake(sc=False, ardour=False)          # state after SuperDirt stopped and Ardour was closed
+    assert f.go() == 0
+    assert f.calls[-1][-1] == "off"
+
+
+def test_a_failed_superdirt_stop_does_not_declare_work_mode():
+    f = Fake(sc=True, stop_rc=1)
+    assert f.go() == 1
+    assert not any(c[-1:] == ("off",) for c in f.calls)
+
+
+def test_superdirt_that_will_not_die_leaves_music_mode_on(capsys):
+    f = Fake(sc=True)
+    f.run = lambda *a: (f.calls.append(a), subprocess.CompletedProcess(a, 0, "", ""))[1]   # stop "works", sc stays up
+    assert f.go() == 1
+    assert "still running" in capsys.readouterr().out
+
+
+def test_music_work_refusing_is_reported_not_swallowed(capsys):
+    f = Fake(sc=True, off_rc=1)
+    assert f.go() == 1
+    assert "refused" in capsys.readouterr().out
+
+
+def test_a_recently_written_stem_file_is_a_rolling_take(tmp_path):
+    wav = tmp_path / "interchange" / "Tidal Live" / "audiofiles" / "Tidal 01.wav"
+    wav.parent.mkdir(parents=True)
+    wav.write_bytes(b"x")
+    now = time.time()
+    assert es.take_rolling(tmp_path, now=lambda: now + 2) is True
+    assert es.take_rolling(tmp_path, now=lambda: now + 600) is False
+    assert es.take_rolling(tmp_path / "nowhere") is False
+
+
+def test_comm_running_matches_the_exact_name_only(tmp_path):
+    for pid, name in (("100", "sclang"), ("200", "sclang-helper"), ("300", "bash")):
+        d = tmp_path / pid
+        d.mkdir()
+        (d / "comm").write_text(name + "\n")
+    assert es.comm_running(("sclang",), tmp_path) is True
+    assert es.comm_running(("scsynth",), tmp_path) is False       # not a prefix match
+    assert es.comm_running(("sclang-hel",), tmp_path) is False
+
+
+def test_dry_run_changes_nothing_and_predicts_the_exit_code(capsys):
+    assert es.dry_run(running=lambda n: False, rolling=lambda: False) == 0
+    assert es.dry_run(running=lambda n: set(n) == set(es.ARDOUR_PROCS), rolling=lambda: False) == 3
+    assert es.dry_run(running=lambda n: False, rolling=lambda: True) == 2
+    assert es.dry_run(running=lambda n: False, rolling=lambda: True, force=True) == 0
+    assert "would" in capsys.readouterr().out
+
+
+def test_the_outcome_is_announced_for_every_exit_path():
+    """The caller may be the tray, which this script stops at the end, so a return value
+    is not enough: every outcome must reach the desktop."""
+    for kw, code in ((dict(rolling=True), 2), (dict(sc=True, ardour=True), 3), (dict(sc=True), 0),
+                     (dict(sc=True, stop_rc=1), 1)):
+        said = []
+        f = Fake(**kw)
+        got = es.end(running=f.running, rolling=lambda: f.rolling, run=f.run,
+                     sleep=lambda s: None, wait_s=1.0, notify=said.append)
+        assert got == code and len(said) == 1, (kw, got, said)