feat(gig-log): k:'rec' — two lenses on Ardour actually recording

'Recording X had track Y playing at time Z' becomes answerable. Files lens =
ground truth (stems growing in interchange/, stat-only, 2s grace, RecPass state
machine); OSC lens = Ardour's own word (/set_surface feedback=8 handshake to
:3819, listens :3820, /rec_enable_toggle + transport — the manual's real name
for master arm, not /rec_enabled). Records on transitions + every tick while
recording; lens disagreement ships as conflict:true, never smoothed over.
44 new tests (155 total green), stdlib-only, --no-rec escape hatch.

tools/gig-log.py

@@ -23,6 +23,14 @@ the lag it was reporting. So:
   * NO per-sample subprocess spawn. Temps, freq, throttle, cpu and RSS all come
     from sysfs/procfs reads. Two long-lived children total (`pw-top -b` for xruns,
     `aseqdump` for MIDI), started once and drained by threads.
+  * THE OSC LENS IS A SOCKET, NOT A CHILD. It is the one place this file talks
+    network instead of sysfs/procfs — one non-blocking UDP socket on 3820,
+    drained by its own thread exactly like the two children above. Still zero
+    forks; the rule above is about forking, and a socket isn't one.
+  * THE FILES LENS NEVER OPENS AUDIO. It answers "is Ardour writing" by
+    stat()ing file sizes under interchange/<session>/audiofiles/ — never a
+    read() of the bytes. Same reasoning as "no audio capture": the truth we
+    need is in the metadata, not the waveform.
   * MIDI IS COALESCED, because a fader sweep is 100+ events/s and a logger that
     writes them all is a firehose that costs more than what it measures. One line
     per (port, channel, controller) per second, carrying count + first/last/min/max.
@@ -52,6 +60,13 @@ RECORD KINDS (the `k` field)
          cheap regex found any, which dN stream(s) the evaluated block
          defines. This is the honest boundary signal `track` isn't — an eval
          means the pattern changed, not just that the tab was in focus.
+  rec    Ardour's RECORDING state, from two independent lenses: `files`
+         (audio actually landing on disk — zero Ardour config, so it is
+         GROUND TRUTH) and `osc` (Ardour's own transport+arm feedback, over
+         its OSC surface). Written on every rec/arm TRANSITION, plus once
+         per tick WHILE recording, so the timeline has a spine — never while
+         idle. See "TWO LENSES ON RECORDING" below for what a disagreement
+         between them means and why it is reported, never resolved quietly.
   mark   a human annotation from `gig-log.py mark`
   end    once, at stop
 
@@ -62,6 +77,30 @@ telemetry and becomes an annotated performance score: which effect, on which orb
 at which second. That is the input for splitting a session into tracks, for telling
 practice apart from performance, and for finding the moment worth clipping.
 
+TWO LENSES ON RECORDING
+`rec` exists so a review can answer "recording X had track Y playing at time Z" —
+joining `k:"rec"` against `k:"track"`/`k:"eval"` on their timestamps. It has to be
+two lenses because neither one alone tells the truth:
+
+  * FILES is authoritative: it watches interchange/<session>/audiofiles/ grow, byte
+    by byte, on disk. Nothing about Ardour's config, arm state or transport display
+    can lie to it — either bytes landed or they didn't. But it needs a live,
+    resolvable session (see check-mix.py's 2026-07-28 lesson: audit the RUNNING
+    session, from the process's own argv, never a hardcoded/guessed path).
+  * OSC is Ardour's own opinion: master record-arm and transport-rolling, over its
+    control-surface protocol. It works even when the files lens can't resolve a
+    session, but it is exactly the kind of "config says X" signal that has burned
+    this rig before (`../Ardour/Tidal Live/` vs `Tidal Multi`, faders at -inf while
+    the editor showed nothing wrong) — so it is corroborating, not authoritative.
+
+They can disagree, and both directions matter: ARMED BUT NOT ROLLING is not
+recording (OSC alone would get this right; a naive "arm implies recording" reading
+would not). ROLLING+ARMED WITH NO FILE GROWTH is the alarming case — Ardour thinks
+it is recording and nothing is landing on disk. When both lenses are live and
+disagree on the recording bool, the record commits to an answer (whichever lens
+says true — silence is the wrong default for a possible false negative) but sets
+`"conflict": true`, so a reviewer sees it rather than a tool quietly picking a side.
+
 USAGE
   tools/gig-log.py record                    # 1 Hz recorder, Ctrl-C to stop
   tools/gig-log.py mark "vague de crime"     # annotate the running session
@@ -80,6 +119,8 @@ import os
 import re
 import shutil
 import signal
+import socket
+import struct
 import subprocess
 import sys
 import threading
@@ -589,6 +630,426 @@ class MidiReader(threading.Thread):
 
 
 # --------------------------------------------------------------------------- #
+# rec: two lenses on Ardour's RECORDING state (see "TWO LENSES ON RECORDING"
+# in the module header for WHY there are two and what a disagreement means)
+# --------------------------------------------------------------------------- #
+
+# ---- files lens: audio actually landing on disk, zero Ardour config ------- #
+
+FILES_RESCAN_S = 10.0   # how often to re-resolve the session (Ardour can (re)start)
+FILES_RECENT_S = 10.0   # a file untouched this long is not "being written now"
+FILES_GRACE_S = 2.0     # a flush/buffer stall this short must not end the pass
+
+
+def files_growth(prev_sizes: dict[str, int], entries: dict[str, tuple[float, int]],
+                 now: float, recent_s: float = FILES_RECENT_S
+                 ) -> tuple[int, dict[str, int]]:
+    """(bytes grown since last tick, sizes to keep for the NEXT tick).
+
+    Pure and filesystem-free on purpose — this is the arithmetic that decides
+    "is Ardour writing right now", so it has to be testable without a real
+    Ardour session. `entries` is {filename: (mtime, size)} for one os.scandir
+    snapshot; only files touched within `recent_s` are candidates (a file from
+    an hour ago sitting in the same folder must not count), and a file first
+    seen this tick (no `prev`) contributes no growth — its own next sample is
+    what proves it grew, exactly like XrunReader's warm-up needs a second
+    sighting before a delta means anything.
+    """
+    cutoff = now - recent_s
+    growth = 0
+    kept: dict[str, int] = {}
+    for name, (mtime, size) in entries.items():
+        if mtime < cutoff:
+            continue
+        prev = prev_sizes.get(name)
+        if prev is not None and size > prev:
+            growth += size - prev
+        kept[name] = size
+    return growth, kept
+
+
+class RecPass:
+    """The GRACE-window state machine: growth-per-tick in, (rec, elapsed,
+    bytes) out. No I/O — FilesLens does the stat()ing and feeds this the one
+    number that matters, so the grace/reset logic is testable on its own,
+    the same way XrunReader._baseline is tested without a real pw-top.
+
+    A recording PASS is continuous growth allowing FILES_GRACE_S of silence
+    (a flush, a buffer stall) without ending it. `elapsed` and `bytes` reset
+    to zero the moment a pass ends, because they describe THIS pass, not the
+    file's lifetime — the same "session, not history" rule as _pkg_base.
+    """
+
+    def __init__(self, grace_s: float = FILES_GRACE_S):
+        self.grace_s = grace_s
+        self._last_growth: float | None = None
+        self._start: float | None = None
+        self.bytes_total = 0
+
+    def tick(self, growth: int, now_mono: float) -> dict:
+        if growth > 0:
+            self._last_growth = now_mono
+        active = (self._last_growth is not None
+                 and now_mono - self._last_growth <= self.grace_s)
+        if not active:
+            self._start = None
+            self.bytes_total = 0
+            return {"rec": False, "el": 0.0, "by": 0}
+        if self._start is None:
+            self._start = self._last_growth   # the pass began at the growth, not now
+            self.bytes_total = 0
+        self.bytes_total += growth
+        return {"rec": True, "el": round(now_mono - self._start, 1),
+                "by": self.bytes_total}
+
+
+class FilesLens:
+    """"Is Ardour recording" answered from the one place that can't lie: bytes
+    landing on disk under interchange/<session>/audiofiles/.
+
+    Session resolution is COPIED from check-mix.py's `running_session()`
+    (2026-07-28: a stale/guessed session audited as if it were live produced
+    a confidently wrong report — pgrep also doesn't work here, the process is
+    `ArdourGUI`, not `ardour`). Keep the two in sync if that logic changes;
+    it is duplicated rather than imported because check-mix.py is a script,
+    not a library, and this file stays stdlib-only either way.
+
+    Cost per tick: one os.scandir() (one syscall) plus one stat() per entry
+    already returned by it, on ONE small directory — no subprocess, no read()
+    of audio data, ever. Re-resolves the session every FILES_RESCAN_S rather
+    than once at startup, for the same reason GearWatch re-scans pids: a
+    binding fixed once and never rechecked is this rig's most common failure.
+    """
+
+    def __init__(self):
+        self._audiofiles: Path | None = None
+        self._last_scan = 0.0
+        self._sizes: dict[str, int] = {}
+        self._pass = RecPass()
+        self._logged_reason: str | None = "unset"   # force one log at startup
+
+    @staticmethod
+    def resolve_audiofiles_dir() -> Path | None:
+        """<session>/interchange/<session name>/audiofiles/, or None.
+
+        Matches check-mix.py's running_session(): the live Ardour process's
+        OWN argv, not a hardcoded path and not `pgrep ardour` (the binary is
+        ArdourGUI; a pgrep match on "ardour" returns nothing).
+        """
+        try:
+            for proc in Path("/proc").iterdir():
+                if not proc.name.isdigit():
+                    continue
+                try:
+                    argv = (proc / "cmdline").read_bytes().split(b"\0")
+                except OSError:
+                    continue
+                if not argv or b"ardour" not in argv[0].lower():
+                    continue
+                for arg in argv[1:]:
+                    if arg.endswith(b".ardour"):
+                        p = Path(arg.decode(errors="replace"))
+                        if p.is_file():
+                            return p.parent / "interchange" / p.stem / "audiofiles"
+        except OSError:
+            pass
+        return None
+
+    def _log(self, reason: str | None) -> None:
+        if reason == self._logged_reason:
+            return
+        self._logged_reason = reason
+        if reason:
+            print(f"gig-log: files lens inactive — {reason}", file=sys.stderr)
+        else:
+            print(f"gig-log: files lens active — {self._audiofiles}", file=sys.stderr)
+
+    def sample(self) -> dict:
+        now_mono = time.monotonic()
+        if self._audiofiles is None or now_mono - self._last_scan >= FILES_RESCAN_S:
+            self._last_scan = now_mono
+            found = self.resolve_audiofiles_dir()
+            if found != self._audiofiles:
+                self._audiofiles = found
+                self._sizes = {}          # a session/track swap invalidates old sizes
+        if self._audiofiles is None:
+            self._log("Ardour not running, or no .ardour session resolvable "
+                      "from its argv")
+            return {"active": False, "rec": False, "el": 0.0, "by": 0}
+        try:
+            entries = {}
+            for e in os.scandir(self._audiofiles):
+                st = e.stat()
+                entries[e.name] = (st.st_mtime, st.st_size)
+        except OSError as exc:
+            self._log(f"cannot read {self._audiofiles}: {exc}")
+            return {"active": False, "rec": False, "el": 0.0, "by": 0}
+        self._log(None)
+        growth, self._sizes = files_growth(self._sizes, entries, time.time())
+        out = self._pass.tick(growth, now_mono)
+        out["active"] = True
+        return out
+
+
+# ---- osc lens: Ardour's own transport+arm feedback, over its OSC surface -- #
+
+# Wire format, minimal OSC 1.0 — stdlib struct+socket only, no python-osc dep.
+def _osc_pad(b: bytes) -> bytes:
+    return b + b"\0" * ((4 - len(b) % 4) % 4)
+
+
+def osc_encode(address: str, *args: bool | int | float | str) -> bytes:
+    """One OSC message: address, then a comma-prefixed type-tag string, then
+    the argument bytes. `bool` is checked before `int` — in Python `bool` IS
+    an `int` subclass, so the naive order would type every True/False as an
+    int32 0/1 instead of an OSC T/F (no bytes at all)."""
+    out_types = ","
+    out_args = b""
+    for a in args:
+        if isinstance(a, bool):
+            out_types += "T" if a else "F"
+        elif isinstance(a, int):
+            out_types += "i"
+            out_args += struct.pack(">i", a)
+        elif isinstance(a, float):
+            out_types += "f"
+            out_args += struct.pack(">f", a)
+        elif isinstance(a, str):
+            out_types += "s"
+            out_args += _osc_pad(a.encode() + b"\0")
+        else:
+            raise TypeError(f"unsupported OSC arg type: {type(a)!r}")
+    return (_osc_pad(address.encode() + b"\0")
+           + _osc_pad(out_types.encode()) + out_args)
+
+
+def _osc_read_str(data: bytes, off: int) -> tuple[str, int]:
+    end = data.index(b"\0", off)
+    s = data[off:end].decode("utf-8", errors="replace")
+    off = end + 1
+    return s, off + (4 - off % 4) % 4
+
+
+def osc_decode(data: bytes) -> tuple[str, list] | None:
+    """(address, args) from one OSC packet, or None if it can't be parsed.
+
+    Minimal on purpose: only the type tags Ardour's feedback actually uses
+    (i/f/s/T/F/N/b) plus a floor of "never raise" — a malformed or truncated
+    UDP datagram must drop this one packet, not take the recorder down, the
+    same contract _eval_record holds for a line from a different process.
+    Bundles (`#bundle...`) are not unpacked; Ardour's feedback here is single
+    messages, so a bundle is simply reported as unparseable.
+    """
+    try:
+        addr, off = _osc_read_str(data, 0)
+        if not addr.startswith("/"):
+            return None
+        if off >= len(data) or data[off:off + 1] != b",":
+            return addr, []
+        types, off = _osc_read_str(data, off)
+        args: list = []
+        for t in types[1:]:
+            if t == "i":
+                args.append(struct.unpack_from(">i", data, off)[0]); off += 4
+            elif t == "f":
+                args.append(struct.unpack_from(">f", data, off)[0]); off += 4
+            elif t == "s":
+                s, off = _osc_read_str(data, off)
+                args.append(s)
+            elif t == "T":
+                args.append(True)
+            elif t == "F":
+                args.append(False)
+            elif t == "N":
+                args.append(None)
+            elif t == "b":
+                blen = struct.unpack_from(">i", data, off)[0]; off += 4
+                args.append(data[off:off + blen])
+                off += blen + (4 - blen % 4) % 4
+            else:
+                break               # unknown tag: stop, return what parsed cleanly
+        return addr, args
+    except (struct.error, ValueError, IndexError):
+        return None
+
+
+OSC_SEND_PORT = 3819    # Ardour's OSC surface (fixed, per Ardour's own docs)
+OSC_RECV_PORT = 3820    # OUR reply port — must match Ardour's OSC setup dialog:
+                        # Preferences > Control Surfaces > OSC > Port Mode = Manual,
+                        # Reply Port = 3820. Manual mode makes Ardour send feedback
+                        # to this FIXED port regardless of the port our handshake
+                        # happened to go out from — the alternative ("auto" mode,
+                        # matching send/receive port) is exactly the kind of
+                        # implicit binding this rig has been burned by before
+                        # (feedback_stale_binding_pattern), so this lens insists on
+                        # the explicit, documented setting instead.
+OSC_RETRY_S = 30.0      # re-send the handshake at most this often while quiet
+OSC_STALE_S = 8.0       # /heartbeat is 1/s; silence this long means the link is dead
+
+# Feedback bitmask for /set_surface, from manual.ardour.org's "Calculating
+# Feedback and Strip-types Values": bit value 8 = "heartbeat to surface".
+# That is the ONLY bit gig-log asks for. Every other defined bit is about
+# STRIP feedback this lens does not want (button/fader state, meters, master-
+# section gain, bar/beat, timecode, selection...) — none of them gate
+# /transport_play, /transport_stop or /rec_enable_toggle, which the manual's
+# OSC-control page lists as plain surface feedback, not tied to any bit. So
+# heartbeat alone is the minimal request that also proves the link is alive.
+OSC_FEEDBACK = 8
+
+
+def combine_rec(files_active: bool, files_rec: bool,
+                osc_active: bool, osc_arm: bool | None, osc_roll: bool | None
+                ) -> tuple[bool, str | None, bool]:
+    """Fold the two lenses into (rec, src, conflict) — see "TWO LENSES ON
+    RECORDING" in the module header. OSC's own notion of "recording" is
+    arm AND rolling; a fader/monitor state OSC cannot see is exactly what
+    the files lens exists to catch, so on a DISAGREEMENT this does not pick
+    a side by trusting one lens more — it reports whichever says TRUE (an
+    unnoticed false negative is worse than a flagged uncertainty) and sets
+    conflict=True so a reviewer sees it instead of a quietly wrong "clean".
+    """
+    osc_rec = bool(osc_active and osc_arm and osc_roll)
+    if files_active and osc_active:
+        if files_rec == osc_rec:
+            return files_rec, "files+osc", False
+        return (files_rec or osc_rec), ("files" if files_rec else "osc"), True
+    if files_active:
+        return files_rec, "files", False
+    if osc_active:
+        return osc_rec, "osc", False
+    return False, None, False
+
+
+class OscLens(threading.Thread):
+    """Ardour's own transport+arm truth, over its OSC control surface.
+
+    UDP out to 3819 (the handshake), UDP in on 3820 (feedback) — a thread,
+    not a subprocess, draining recvfrom() in a loop exactly like XrunReader
+    and MidiReader drain their child's stdout. `/set_surface` asks for
+    heartbeat feedback only (see OSC_FEEDBACK); Ardour then also sends plain
+    transport/record state unprompted, per the OSC-control page. If nothing
+    answers at all, or feedback stops arriving, the lens goes `alive=False`
+    and the handshake is retried at most every OSC_RETRY_S — never more
+    often, so a genuinely absent Ardour doesn't turn into a busy-retry loop.
+    """
+
+    def __init__(self, host: str = "127.0.0.1",
+                send_port: int = OSC_SEND_PORT, recv_port: int = OSC_RECV_PORT):
+        super().__init__(daemon=True)
+        self.host = host
+        self.send_port = send_port
+        self.recv_port = recv_port
+        self.lock = threading.Lock()
+        self.alive = False
+        self.arm: bool | None = None
+        self.roll: bool | None = None
+        self._last_rx = 0.0
+        self._last_handshake = 0.0
+        self._logged_alive: bool | None = None
+        self._sock: socket.socket | None = None
+        self._stop = threading.Event()
+
+    def _handshake(self, sock: socket.socket) -> None:
+        # bank_size strip_types feedback fadermode send_page_size plugin_page_size
+        # port linkset linkid — per manual.ardour.org's osc-control page. strip_types
+        # and the page-size args are all 0: no per-strip feedback wanted at all.
+        # port=0 means "auto", which per that same page defers to the setup
+        # dialog's Manual/3820 config rather than overriding it.
+        try:
+            sock.sendto(osc_encode("/set_surface", 0, 0, OSC_FEEDBACK, 0, 0, 0, 0, 0, 0),
+                       (self.host, self.send_port))
+        except OSError:
+            pass
+        self._last_handshake = time.monotonic()
+
+    def _feed(self, data: bytes) -> None:
+        got = osc_decode(data)
+        if got is None:
+            return
+        addr, args = got
+        with self.lock:
+            self._last_rx = time.monotonic()
+            self.alive = True
+            if addr == "/transport_play" and args:
+                self.roll = bool(args[0])
+            elif addr == "/transport_stop" and args and bool(args[0]):
+                self.roll = False
+            elif addr == "/rec_enable_toggle" and args:
+                self.arm = bool(args[0])
+            # /heartbeat carries no state of its own — it already did its job
+            # by updating _last_rx/alive above.
+
+    def _log_if_changed(self) -> None:
+        if self.alive == self._logged_alive:
+            return
+        self._logged_alive = self.alive
+        print(f"gig-log: osc lens {'connected' if self.alive else 'lost feedback'} "
+             f"(udp {self.send_port}->{self.recv_port})", file=sys.stderr)
+
+    def run(self) -> None:
+        try:
+            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            sock.bind((self.host, self.recv_port))
+            sock.settimeout(1.0)
+        except OSError as exc:
+            print(f"gig-log: osc lens off — cannot bind udp {self.recv_port}: {exc}",
+                 file=sys.stderr)
+            return
+        self._sock = sock
+        self._handshake(sock)
+        while not self._stop.is_set():
+            try:
+                data, _ = sock.recvfrom(2048)
+            except socket.timeout:
+                pass
+            except OSError:
+                break
+            else:
+                self._feed(data)
+            now = time.monotonic()
+            with self.lock:
+                if self._last_rx and now - self._last_rx >= OSC_STALE_S:
+                    self.alive = False
+            self._log_if_changed()
+            if not self.alive and now - self._last_handshake >= OSC_RETRY_S:
+                self._handshake(sock)
+        try:
+            sock.close()
+        except OSError:
+            pass
+
+    def sample(self) -> dict:
+        with self.lock:
+            return {"active": self.alive, "arm": self.arm, "roll": self.roll}
+
+    def stop(self) -> None:
+        self._stop.set()
+        if self._sock:
+            try:
+                self._sock.close()
+            except OSError:
+                pass
+
+
+def rec_record(now: float, rec: bool, src: str | None, conflict: bool,
+              files: dict, osc: dict) -> dict:
+    """Build one `k:"rec"` line. Only carries a lens's fields when that lens
+    is actually live — an inactive lens has nothing honest to report."""
+    r: dict = {"t": round(now, 3), "k": "rec", "rec": rec}
+    if src:
+        r["src"] = src
+    if conflict:
+        r["conflict"] = True
+    if files.get("active"):
+        r["el"] = files.get("el")
+        r["by"] = files.get("by")
+    if osc.get("active"):
+        r["arm"] = osc.get("arm")
+        r["roll"] = osc.get("roll")
+    return r
+
+
+# --------------------------------------------------------------------------- #
 # the recorder
 # --------------------------------------------------------------------------- #
 
@@ -611,7 +1072,7 @@ def _package_throttle_sum() -> int:
 class Recorder:
     def __init__(self, out_dir: Path = LOG_DIR, hz: float = 1.0,
                  xruns: bool = True, midi: bool = True,
-                 midi_port: str | None = None):
+                 midi_port: str | None = None, rec: bool = True):
         self.dir = out_dir
         self.period = 1.0 / max(0.05, hz)
         self.th = perf.Thermals()
@@ -623,6 +1084,12 @@ class Recorder:
         # writes it lazily, same as current-track), and EvalTail degrades to
         # "nothing new" rather than needing an availability check.
         self.ev = EvalTail()
+        # The two `rec` lenses. FilesLens is always instantiated when enabled —
+        # it costs nothing until sampled (no thread, no socket) and degrades to
+        # "inactive" on its own if Ardour/the session can't be resolved.
+        self.rec_enabled = rec
+        self.files = FilesLens() if rec else None
+        self.osc = OscLens() if rec else None
         self.path = self._open()
         self.samples = 0
         self._stop = threading.Event()
@@ -630,6 +1097,18 @@ class Recorder:
         # value below, which means a session that never switches track still has
         # exactly one authoritative statement of what was loaded.
         self._track: str | None = None
+        # Previous combined-rec bool and OSC arm bool, so poll_rec() can tell a
+        # TRANSITION from steady state — see poll_rec()'s docstring. Seeded to
+        # the natural idle default (False), NOT None: gig-log almost always
+        # starts before the take, and seeding with None would make the very
+        # first real tick read as a spurious False->None "transition" even
+        # when nothing changed (unlike `_track`, which re-seeds from a live
+        # read in run() before the loop, an unarmed/not-recording default
+        # here costs nothing when it's right and is still HONEST when it's
+        # wrong — a recorder started mid-take legitimately reports "already
+        # recording" on its first tick, which is real information, not noise).
+        self._rec_prev: bool = False
+        self._arm_prev: bool = False
 
     def _open(self) -> Path:
         self.dir.mkdir(parents=True, exist_ok=True)
@@ -672,6 +1151,7 @@ class Recorder:
             "base_pkg_throttle": self._pkg_base,
             "period": round(self.period, 3),
             "xruns": bool(self.xr), "midi": bool(self.mid),
+            "rec": self.rec_enabled,
             "track": _read_track(),
             "gear": {c: bool(self.gear._pids.get(c)) for c in GEAR},
         }
@@ -716,8 +1196,32 @@ class Recorder:
         return {"t": round(time.time(), 3), "k": "track",
                 "path": cur, "from": prev}
 
+    def poll_rec(self) -> dict | None:
+        """A `k:"rec"` record on every rec/arm TRANSITION, plus one every
+        tick WHILE recording — nothing while idle. See "TWO LENSES ON
+        RECORDING" in the module header for what `combine_rec` decides.
+
+        `self.files`/`self.osc` are None when `rec=False` was passed to the
+        constructor; sampling a disabled lens as "inactive" (rather than
+        skipping this method entirely) keeps combine_rec as the one place
+        that owns the emit-or-not logic.
+        """
+        f = self.files.sample() if self.files else {"active": False, "rec": False,
+                                                     "el": 0.0, "by": 0}
+        o = self.osc.sample() if self.osc else {"active": False, "arm": None,
+                                                "roll": None}
+        rec, src, conflict = combine_rec(f["active"], f["rec"],
+                                         o["active"], o["arm"], o["roll"])
+        arm = o["arm"] if o["active"] else None
+        emit = rec != self._rec_prev or rec or (o["active"] and arm != self._arm_prev)
+        self._rec_prev = rec
+        self._arm_prev = arm
+        if not emit:
+            return None
+        return rec_record(time.time(), rec, src, conflict, f, o)
+
     def run(self) -> int:
-        for t in (self.xr, self.mid):
+        for t in (self.xr, self.mid, self.osc):
             if t:
                 t.start()
         self._write(self.header())
@@ -726,6 +1230,7 @@ class Recorder:
         print(f"gig-log: recording -> {self.path}", file=sys.stderr)
         print(f"gig-log: xruns={'pw-top' if self.xr else 'off'} "
               f"midi={'aseqdump' if self.mid else 'off'} "
+              f"rec={'files+osc' if self.rec_enabled else 'off'} "
               f"period={self.period:.2f}s", file=sys.stderr)
         next_t = time.monotonic()
         try:
@@ -736,6 +1241,9 @@ class Recorder:
                 tr = self.poll_track()
                 if tr:
                     self._write(tr)
+                rc = self.poll_rec()
+                if rc:
+                    self._write(rc)
                 for raw in self.ev.poll():
                     rec = _eval_record(raw)
                     if rec:
@@ -755,7 +1263,7 @@ class Recorder:
                          "samples": self.samples,
                          "midi_events": self.mid.events if self.mid else 0,
                          "xrun": self.xr.total_delta() if self.xr else None})
-            for t in (self.xr, self.mid):
+            for t in (self.xr, self.mid, self.osc):
                 if t:
                     t.stop()
             self.f.close()
@@ -1578,6 +2086,8 @@ def main(argv: list[str] | None = None) -> int:
                      help="skip pw-top (no PipeWire client at all)")
     rec.add_argument("--no-midi", action="store_true", help="skip aseqdump")
     rec.add_argument("--midi-port", help="ALSA seq port (default: find the LCXL)")
+    rec.add_argument("--no-rec", action="store_true",
+                     help="skip the files+osc recording-state lenses")
     rec.add_argument("--seconds", type=float, help="stop after N seconds")
 
     mk = sub.add_parser("mark", help="annotate the running session")
@@ -1632,7 +2142,8 @@ def main(argv: list[str] | None = None) -> int:
         r = Recorder(out_dir=a.dir, hz=hz,
                      xruns=not getattr(a, "no_xruns", False),
                      midi=not getattr(a, "no_midi", False),
-                     midi_port=getattr(a, "midi_port", None))
+                     midi_port=getattr(a, "midi_port", None),
+                     rec=not getattr(a, "no_rec", False))
         signal.signal(signal.SIGTERM, r.stop)
         signal.signal(signal.SIGINT, r.stop)
         if getattr(a, "seconds", None):

tools/tests/test_gig_log.py

@@ -972,3 +972,319 @@ def test_track_and_eval_marks_coexist_the_new_lens_does_not_replace_the_old(tmp_
     _, _, _, marks, _ = gl.load(r.path)
     kinds = [m["k"] for m in marks]
     assert kinds == ["track", "eval"]
+
+
+# --------------------------------------------------------------------------- #
+# k:"rec" — two lenses on Ardour's RECORDING state
+#
+# `files` is ground truth (bytes landing on disk); `osc` is Ardour's own
+# transport+arm feedback. Neither is trusted blindly: files needs a
+# resolvable session (check-mix.py's 2026-07-28 lesson — audit the RUNNING
+# session, from argv, never a guess), and osc is exactly the kind of
+# "config/state says X" signal that has been wrong before. These tests pin
+# the pure arithmetic (growth, the grace-window pass, and how the two lenses
+# combine) without touching a real Ardour session or socket.
+# --------------------------------------------------------------------------- #
+
+def test_files_growth_counts_only_positive_deltas_on_recent_files():
+    prev = {"Tidal01.wav": 1000}
+    entries = {"Tidal01.wav": (100.0, 1500)}          # grew 500 bytes
+    growth, kept = gl.files_growth(prev, entries, now=100.0, recent_s=10.0)
+    assert growth == 500
+    assert kept == {"Tidal01.wav": 1500}
+
+
+def test_files_growth_ignores_files_outside_the_recent_window():
+    """A file from an hour ago sitting in the same audiofiles/ dir (an old
+    take) must not count, or a stale directory would read as 'recording'
+    forever."""
+    prev = {}
+    entries = {"old_take.wav": (0.0, 999999)}
+    growth, kept = gl.files_growth(prev, entries, now=100.0, recent_s=10.0)
+    assert growth == 0 and kept == {}
+
+
+def test_files_growth_a_brand_new_file_contributes_no_growth_yet():
+    """The tick a file first appears has no `prev` to diff against — its
+    own NEXT sample is what proves it grew, the same 'second sighting'
+    discipline as XrunReader's warm-up."""
+    growth, kept = gl.files_growth({}, {"new.wav": (100.0, 4096)}, now=100.0)
+    assert growth == 0
+    assert kept == {"new.wav": 4096}
+
+
+def test_files_growth_a_shrinking_file_is_not_negative_growth():
+    """A rotated/truncated file must not produce a negative delta that could
+    cancel real growth elsewhere — same 'never go negative' rule as
+    XrunReader.deltas()."""
+    growth, kept = gl.files_growth({"a.wav": 5000}, {"a.wav": (100.0, 100)}, now=100.0)
+    assert growth == 0
+    assert kept == {"a.wav": 100}
+
+
+def test_files_growth_sums_across_several_tracked_files():
+    prev = {"o1.wav": 100, "o2.wav": 200}
+    entries = {"o1.wav": (100.0, 150), "o2.wav": (100.0, 260)}
+    growth, _ = gl.files_growth(prev, entries, now=100.0)
+    assert growth == 110
+
+
+def test_rec_pass_is_idle_with_no_growth():
+    p = gl.RecPass()
+    assert p.tick(0, now_mono=1000.0) == {"rec": False, "el": 0.0, "by": 0}
+
+
+def test_rec_pass_starts_a_pass_on_the_first_growth():
+    p = gl.RecPass()
+    out = p.tick(4096, now_mono=1000.0)
+    assert out == {"rec": True, "el": 0.0, "by": 4096}
+
+
+def test_rec_pass_accumulates_elapsed_and_bytes_across_continuous_growth():
+    p = gl.RecPass()
+    p.tick(1000, now_mono=0.0)
+    p.tick(1000, now_mono=1.0)
+    out = p.tick(1000, now_mono=2.0)
+    assert out == {"rec": True, "el": 2.0, "by": 3000}
+
+
+def test_rec_pass_survives_a_short_gap_without_ending_the_pass():
+    """A flush/buffer stall shorter than FILES_GRACE_S must not read as
+    'stopped recording' — that would fragment one real take into many."""
+    p = gl.RecPass(grace_s=2.0)
+    p.tick(1000, now_mono=0.0)
+    stalled = p.tick(0, now_mono=1.5)                 # no growth, still within grace
+    assert stalled["rec"] is True and stalled["by"] == 1000
+    resumed = p.tick(500, now_mono=1.8)
+    assert resumed["rec"] is True and resumed["by"] == 1500
+    assert resumed["el"] == 1.8                        # elapsed from the ORIGINAL start
+
+
+def test_rec_pass_ends_after_the_grace_window_and_the_next_growth_starts_fresh():
+    p = gl.RecPass(grace_s=2.0)
+    p.tick(1000, now_mono=0.0)
+    idle = p.tick(0, now_mono=5.0)                     # well past grace
+    assert idle == {"rec": False, "el": 0.0, "by": 0}
+    fresh = p.tick(500, now_mono=10.0)
+    assert fresh == {"rec": True, "el": 0.0, "by": 500}, "byte/elapsed must reset, not carry over"
+
+
+@pytest.mark.parametrize("files_active,files_rec,osc_active,osc_arm,osc_roll,want", [
+    # both live and AGREE
+    (True, True, True, True, True, (True, "files+osc", False)),
+    (True, False, True, False, False, (False, "files+osc", False)),
+    (True, False, True, True, False, (False, "files+osc", False)),   # armed, not rolling
+    # both live and DISAGREE — the record still answers, but flags it
+    (True, True, True, False, False, (True, "files", True)),
+    (True, False, True, True, True, (True, "osc", True)),   # the ALARMING case: rolling+
+    # armed but nothing landing on disk
+    # only one lens live
+    (True, True, False, None, None, (True, "files", False)),
+    (True, False, False, None, None, (False, "files", False)),
+    (False, False, True, True, True, (True, "osc", False)),
+    (False, False, True, True, False, (False, "osc", False)),
+    (False, False, True, None, None, (False, "osc", False)),
+    # neither live: quietly nothing
+    (False, False, False, None, None, (False, None, False)),
+])
+def test_combine_rec_table(files_active, files_rec, osc_active, osc_arm, osc_roll, want):
+    assert gl.combine_rec(files_active, files_rec, osc_active, osc_arm, osc_roll) == want
+
+
+def test_rec_record_only_carries_fields_from_LIVE_lenses():
+    """An inactive lens has nothing honest to report — its fields must be
+    absent, not null-filled, so a reviewer can't mistake 'lens was off' for
+    'lens measured false/zero'."""
+    files_off = {"active": False, "rec": False, "el": 0.0, "by": 0}
+    osc_off = {"active": False, "arm": None, "roll": None}
+    r = gl.rec_record(100.0, False, None, False, files_off, osc_off)
+    assert r == {"t": 100.0, "k": "rec", "rec": False}
+
+    files_on = {"active": True, "rec": True, "el": 3.2, "by": 4096}
+    osc_on = {"active": True, "arm": True, "roll": True}
+    r2 = gl.rec_record(100.0, True, "files+osc", False, files_on, osc_on)
+    assert r2 == {"t": 100.0, "k": "rec", "rec": True, "src": "files+osc",
+                 "el": 3.2, "by": 4096, "arm": True, "roll": True}
+
+
+def test_rec_record_sets_conflict_only_when_true():
+    files_on = {"active": True, "rec": True, "el": 0.0, "by": 0}
+    osc_on = {"active": True, "arm": False, "roll": False}
+    r = gl.rec_record(1.0, True, "files", True, files_on, osc_on)
+    assert r["conflict"] is True
+    r2 = gl.rec_record(1.0, True, "files+osc", False, files_on, osc_on)
+    assert "conflict" not in r2
+
+
+# ---- OSC wire format: minimal encode/decode, stdlib struct+socket only --- #
+
+def test_osc_encode_decode_round_trips_ints_and_strings():
+    msg = gl.osc_encode("/select/n_inputs", 8, "master")
+    addr, args = gl.osc_decode(msg)
+    assert addr == "/select/n_inputs"
+    assert args == [8, "master"]
+
+
+def test_osc_encode_uses_TF_tags_for_bool_not_int32():
+    """bool is an int subclass in Python — the naive isinstance order would
+    silently encode True/False as an int32 0/1 instead of an OSC T/F (which
+    carries NO argument bytes at all), desyncing every arg after it."""
+    msg = gl.osc_encode("/transport_play", True)
+    addr, args = gl.osc_decode(msg)
+    assert addr == "/transport_play"
+    assert args == [True]
+
+
+def test_osc_encode_pads_address_and_types_to_4_byte_boundaries():
+    # "/x" (2 bytes) + NUL = 3, padded to 4; the type-tag string is padded too.
+    msg = gl.osc_encode("/x", 1)
+    assert len(msg) % 4 == 0
+
+
+def test_osc_decode_of_the_set_surface_handshake_round_trips():
+    """The actual handshake this file sends: bank_size strip_types feedback
+    fadermode send_page_size plugin_page_size port linkset linkid, all 0
+    except feedback=8 (heartbeat only) — manual.ardour.org's osc-control and
+    calculating-feedback-and-strip-types-values pages."""
+    msg = gl.osc_encode("/set_surface", 0, 0, gl.OSC_FEEDBACK, 0, 0, 0, 0, 0, 0)
+    addr, args = gl.osc_decode(msg)
+    assert addr == "/set_surface"
+    assert args == [0, 0, 8, 0, 0, 0, 0, 0, 0]
+
+
+@pytest.mark.parametrize("junk", [b"", b"\x00", b"not-osc-at-all", b"/no/typetag\x00\x00\x00\x00garbage"])
+def test_osc_decode_never_raises_on_garbage(junk):
+    gl.osc_decode(junk)          # must not raise; None or a best-effort parse are both fine
+
+
+def test_osc_decode_rejects_a_bundle_rather_than_misreading_it_as_an_address():
+    assert gl.osc_decode(b"#bundle\x00\x00\x00\x00\x00\x00\x00\x00") is None
+
+
+def test_osc_decode_a_message_with_no_type_tag_returns_empty_args():
+    addr, args = gl.osc_decode(b"/heartbeat\x00\x00")
+    assert addr == "/heartbeat" and args == []
+
+
+# ---- OscLens: feeding real Ardour-shaped feedback, no socket involved ---- #
+
+def test_osc_lens_feed_updates_roll_from_transport_play_and_stop():
+    lens = gl.OscLens()
+    lens._feed(gl.osc_encode("/transport_play", True))
+    assert lens.alive is True and lens.roll is True
+    lens._feed(gl.osc_encode("/transport_stop", True))
+    assert lens.roll is False
+
+
+def test_osc_lens_feed_updates_arm_from_rec_enable_toggle():
+    lens = gl.OscLens()
+    lens._feed(gl.osc_encode("/rec_enable_toggle", True))
+    assert lens.arm is True
+    lens._feed(gl.osc_encode("/rec_enable_toggle", False))
+    assert lens.arm is False
+
+
+def test_osc_lens_feed_ignores_unrelated_addresses_without_crashing():
+    lens = gl.OscLens()
+    lens._feed(gl.osc_encode("/strip/gain", 1, -6.0))
+    assert lens.arm is None and lens.roll is None
+    assert lens.alive is True, "any parseable feedback still proves the link is alive"
+
+
+def test_osc_lens_feed_degrades_silently_on_unparseable_data():
+    lens = gl.OscLens()
+    lens._feed(b"garbage, not osc at all")
+    assert lens.alive is False
+
+
+def test_osc_lens_sample_reports_inactive_before_anything_arrives():
+    lens = gl.OscLens()
+    assert lens.sample() == {"active": False, "arm": None, "roll": None}
+
+
+# ---- FilesLens: session resolution must never crash off-stage ------------ #
+
+def test_resolve_audiofiles_dir_never_raises_when_no_ardour_is_running():
+    """Same contract as check-mix.py's running_session(): absence of Ardour
+    degrades to None, never an exception — this runs in CI with no /proc
+    entry for Ardour at all."""
+    got = gl.FilesLens.resolve_audiofiles_dir()
+    assert got is None or isinstance(got, Path)
+
+
+def test_files_lens_sample_is_inactive_and_silent_by_default_in_tests(capsys):
+    """No Ardour running in the test sandbox: the lens must report inactive
+    (not raise, not claim a false 'recording'), logging the reason ONCE."""
+    lens = gl.FilesLens()
+    first = lens.sample()
+    assert first == {"active": False, "rec": False, "el": 0.0, "by": 0}
+    second = lens.sample()
+    assert second == first
+    # logged once, not per call — the module header's "log the reason ONCE" rule
+    err_lines = [ln for ln in capsys.readouterr().err.splitlines() if "files lens" in ln]
+    assert len(err_lines) <= 1
+
+
+# ---- Recorder.poll_rec(): the emit-on-transition-or-while-recording rule - #
+
+class _FixedLens:
+    """A stand-in for FilesLens/OscLens that returns a canned sequence of
+    `.sample()` results — the same substitution trick used elsewhere in this
+    file (e.g. `r.ev = gl.EvalTail(eval_file)`), so poll_rec()'s emit logic
+    can be driven tick-by-tick without a real Ardour session or socket."""
+
+    def __init__(self, *outs):
+        self._outs = list(outs)
+
+    def sample(self):
+        return self._outs.pop(0)
+
+
+def test_poll_rec_emits_nothing_while_idle_then_marks_the_start_and_stop(tmp_path):
+    r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False, rec=False)
+    idle = {"active": True, "rec": False, "el": 0.0, "by": 0}
+    recording = {"active": True, "rec": True, "el": 1.0, "by": 4096}
+    osc_off = {"active": False, "arm": None, "roll": None}
+    r.files = _FixedLens(idle, idle, recording, recording, recording, idle, idle)
+    r.osc = _FixedLens(*([osc_off] * 7))
+    got = [r.poll_rec() for _ in range(7)]
+    rec_flags = [g["rec"] if g else None for g in got]
+    assert rec_flags == [None, None, True, True, True, False, None]
+    assert got[2]["src"] == "files" and got[2]["by"] == 4096
+    r.f.close()
+
+
+def test_poll_rec_emits_on_an_arm_change_even_when_rec_stays_false(tmp_path):
+    """OSC can arm without rolling — not a recording (combine_rec agrees),
+    but PLN arming the transport IS a state change worth a line in the
+    timeline, independent of whether files ever sees growth."""
+    r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False, rec=False)
+    files_off = {"active": False, "rec": False, "el": 0.0, "by": 0}
+    unarmed = {"active": True, "arm": False, "roll": False}
+    armed = {"active": True, "arm": True, "roll": False}
+    r.files = _FixedLens(*([files_off] * 4))
+    r.osc = _FixedLens(unarmed, armed, armed, unarmed)
+    got = [r.poll_rec() for _ in range(4)]
+    assert [g is not None for g in got] == [False, True, False, True]
+    assert got[1]["rec"] is False and got[1]["arm"] is True
+    r.f.close()
+
+
+def test_poll_rec_flags_conflict_when_the_lenses_disagree(tmp_path):
+    """The alarming case: OSC says rolling+armed, files sees nothing land.
+    Must not resolve silently — conflict=True, and rec reports the lens that
+    said TRUE (a missed problem is worse than a flagged uncertainty)."""
+    r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False, rec=False)
+    r.files = _FixedLens({"active": True, "rec": False, "el": 0.0, "by": 0})
+    r.osc = _FixedLens({"active": True, "arm": True, "roll": True})
+    got = r.poll_rec()
+    assert got["rec"] is True and got["conflict"] is True and got["src"] == "osc"
+    r.f.close()
+
+
+def test_recorder_with_rec_disabled_never_touches_the_lenses(tmp_path):
+    r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False, rec=False)
+    assert r.files is None and r.osc is None
+    assert r.header()["rec"] is False
+    assert r.poll_rec() is None      # both lenses report inactive -> no record
+    r.f.close()