Name the sender behind every MPRIS call

control-events.log records that a [mpris] PlayPause arrived but not who sent
it — PyQt6 has no QDBusContext, so the adaptor can't see the caller. This tails
dbus-monitor, filters MPRIS method calls, and resolves each sender's unique bus
name to a pid and cmdline, which is the other half of the picture.

It has already earned its place: two weeks of capture show every
playback-affecting call into LinTunes coming from gsd-media-keys during waking
hours, and not one between midnight and 7am — so the phantom overnight audio
was not a stray Play, which is consistent with it being the paused-but-live
PipeWire stream fixed in v0.6.1. Read-only; touches no library or music file.

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

scripts/watch_mpris_senders.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""Attribute every D-Bus call into LinTunes' MPRIS interface to the sender.
+
+Run this in the background during normal use (and overnight) to catch the
+"phantom playback" incidents: lintunes' control-events.log shows those resumes
+arriving as [mpris] PlayPause / Play, but can't name *who* sent the call (PyQt6
+has no QDBusContext — see lintunes/mpris.py). This script does the other half:
+it tails `dbus-monitor --session`, filters for method calls on
+org.mpris.MediaPlayer2(.Player), and for each one resolves the sender's unique
+bus name (:1.NN) to a PID + command via the D-Bus daemon, writing a provenance
+line to ~/.cache/lintunes/mpris-senders.log (and stdout).
+
+Read-only: it never touches lintunes, the library, or any music file.
+
+Usage:
+    python3 scripts/watch_mpris_senders.py            # log + echo to stdout
+    python3 scripts/watch_mpris_senders.py --quiet    # log only
+    python3 scripts/watch_mpris_senders.py --follow  # also tail the log file
+
+Then keep it running. When you next hear the ~4 s of phantom audio, look at the
+last lines of ~/.cache/lintunes/mpris-senders.log — it will name the process
+(almost certainly gsd-media-keys fed by bluetoothd/wireplumber, but let's prove
+it rather than guess).
+"""
+import argparse
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+LOG_PATH = Path.home() / ".cache" / "lintunes" / "mpris-senders.log"
+
+# Members of the Player interface that can start/change playback. We log all
+# MPRIS calls actually, but flag these in the human log as "playback-affecting".
+PLAYBACK_MEMBERS = {
+    "Play", "PlayPause", "Pause", "Stop", "Next", "Previous",
+    "Seek", "SetPosition", "OpenUri",
+}
+
+
+def _service_pid(sender: str) -> tuple[int | None, str]:
+    """Resolve a D-Bus unique name (:1.NN) to (pid, comm). Returns
+    (None, reason) when the bus daemon can't (or won't) tell us."""
+    try:
+        out = subprocess.run(
+            ["busctl", "--user", "call",
+             "org.freedesktop.DBus", "/org/freedesktop/DBus",
+             "org.freedesktop.DBus", "GetConnectionUnixProcessID", "s", sender],
+            capture_output=True, text=True, timeout=5,
+        )
+    except (FileNotFoundError, subprocess.SubprocessError) as exc:
+        return None, f"busctl error: {exc}"
+    if out.returncode != 0:
+        return None, f"busctl rc={out.returncode}: {out.stderr.strip()}"
+    parts = out.stdout.split()
+    if len(parts) < 2 or parts[0] != "u":
+        return None, f"unparsed busctl output: {out.stdout.strip()!r}"
+    try:
+        pid = int(parts[1])
+    except ValueError:
+        return None, f"non-numeric pid: {parts[1]!r}"
+    # Get the human name of the process for the log. /proc/<pid>/comm is short
+    # (truncated to 15 chars by the kernel); cmdline gives the full invocation
+    # but is NUL-separated and may include the executable path — join and trim.
+    comm = f"pid={pid}"
+    try:
+        with open(f"/proc/{pid}/comm", encoding="utf-8", errors="replace") as f:
+            comm = f.read().strip() or comm
+        with open(f"/proc/{pid}/cmdline", "rb") as f:
+            args = f.read().replace(b"\x00", b" ").decode(
+                "utf-8", errors="replace").strip()
+        if args:
+            comm = f"{comm} :: {args[:160]}"
+    except OSError as exc:
+        comm = f"{comm} (comm unreadable: {exc})"
+    return pid, comm
+
+
+def _parse_method_call(line: str):
+    """Pull (sender, destination, member, interface) out of a dbus-monitor
+    "method call" line, or return None if it isn't one / isn't parseable."""
+    if not line.startswith("method call"):
+        return None
+    if "org.mpris.MediaPlayer2" not in line:
+        return None
+    fields = {}
+    # dbus-monitor writes "key=value" tokens separated by whitespace, and a
+    # trailing "path=...; interface=...; member=..." group. Splitting on ';' or
+    # whitespace and taking the last '=' segment of each token covers both.
+    for tok in line.replace(";", " ").split():
+        if "=" in tok:
+            k, _, v = tok.partition("=")
+            fields.setdefault(k, v)
+    if "member" not in fields or "sender" not in fields:
+        return None
+    return fields
+
+
+def _timestamp() -> str:
+    return time.strftime("%F %T")
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--quiet", action="store_true",
+                    help="don't echo to stdout (still writes the log file)")
+    ap.add_argument("--follow", action="store_true",
+                    help="also tail the log file to stdout as it's written")
+    args = ap.parse_args()
+
+    LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
+    # dbus-monitor emits two preamble lines naming the bus type; keep going.
+    proc = subprocess.Popen(
+        ["dbus-monitor", "--session"],
+        stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+        text=True, bufsize=1,
+    )
+    # Cache sender->(pid, comm) so repeat presses from the same process don't
+    # retry the D-Bus daemon call every line. Refresh if the pid ever changes.
+    pid_cache: dict[str, tuple[int | None, str]] = {}
+
+    print(f"[{time.strftime('%F %T')}] watching session D-Bus for MPRIS calls "
+          f"-> {LOG_PATH}", file=sys.stderr)
+    try:
+        for line in proc.stdout:            # dbus-monitor blocks here
+            line = line.rstrip()
+            parsed = _parse_method_call(line)
+            if parsed is None:
+                continue
+            sender = parsed["sender"]
+            member = parsed["member"]
+            dest = parsed.get("destination", "?")
+            iface = parsed.get("interface", "?")
+            pid_comm = pid_cache.get(sender)
+            if pid_comm is None:
+                pid, comm = _service_pid(sender)
+                pid_cache[sender] = (pid, comm)
+            pid, comm = pid_cache[sender]
+            marker = "<<" if member in PLAYBACK_MEMBERS else "  "
+            rec = (f"{_timestamp()} {marker} {member:10s} "
+                   f"sender={sender} dest={dest} iface={iface} {comm}")
+            with open(LOG_PATH, "a", encoding="utf-8") as f:
+                f.write(rec + "\n")
+            if not args.quiet:
+                print(rec)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        proc.terminate()
+        proc.wait()
+        if args.follow:
+            print(f"\n[{time.strftime('%F %T')}] stopped; log: {LOG_PATH}",
+                  file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()