Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,22 @@

All notable changes to the Lager platform are documented here. For detailed release notes, see [docs.lagerdata.com](https://docs.lagerdata.com).

## [0.31.9] - 2026-07-13

### Fixed
- **UART reconnect can no longer land on a look-alike adapter with a clone
serial.** Many USB-serial adapters ship with a non-unique programmed serial
(e.g. several CP210x units all reading "0001"). If such a device dropped
mid-session, the v0.31.5 reconnect could match a sibling adapter with the
same serial while the real device was still off the bus — attaching the
session to the wrong hardware. Identity resolution now treats a serial
shared by multiple live devices as untrusted (the physical port must match,
and it keeps retrying until the real device returns), and new identity
snapshots record a bus-duplicated serial as null so the net is pinned to
its physical port outright. Nets on clone-serial adapters that were
enriched under v0.31.5 pick up the corrected snapshot on their next
re-save.

## [0.31.8] - 2026-07-13

`lager uninstall` now actually removes what the modern `lager install` creates.
Expand Down
64 changes: 54 additions & 10 deletions box/lager/devices/serial_id.py
Original file line number Diff line number Diff line change
Expand Up @@ -165,13 +165,45 @@ def _interface_for_tty(tty_dev: Path) -> Optional[int]:
return None


def _serial_is_unique(vid: str, pid: str, serial: str, own_dir_name: str) -> bool:
"""False when another live USB device shares vid/pid/serial.

Clone adapters (e.g. CP210x units all programmed with serial "0001") make
the serial worthless as identity. Sibling interfaces of one multi-port
chip share a device dir and are not counted as duplicates.
"""
seen = set()
try:
for tty_dev in _SYS_TTY.iterdir():
if not tty_dev.name.startswith(("ttyUSB", "ttyACM")):
continue
usb_dir = _usb_device_dir_for_tty(tty_dev)
if usb_dir is None or usb_dir.name == own_dir_name or usb_dir.name in seen:
continue
seen.add(usb_dir.name)
if ((_read_sysfs_text(usb_dir / "idVendor") or "").lower() == vid
and (_read_sysfs_text(usb_dir / "idProduct") or "").lower() == pid
and _read_sysfs_text(usb_dir / "serial") == serial):
return False
except OSError:
# Unreadable sysfs mid-walk: assume unique (degrades to trusting the
# serial, same as the other walkers' missing-_SYS_TTY guards).
pass
return True
Comment on lines +175 to +192

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

In environments where /sys/class/tty is missing, restricted, or inaccessible (such as non-Linux development environments or highly sandboxed containers), calling _SYS_TTY.iterdir() can raise OSError (e.g., FileNotFoundError or PermissionError). Wrapping the directory iteration in a try-except OSError block ensures the function fails gracefully and returns True (assuming the serial is unique) rather than crashing the application.

Suggested change
seen = set()
for tty_dev in _SYS_TTY.iterdir():
if not tty_dev.name.startswith(("ttyUSB", "ttyACM")):
continue
usb_dir = _usb_device_dir_for_tty(tty_dev)
if usb_dir is None or usb_dir.name == own_dir_name or usb_dir.name in seen:
continue
seen.add(usb_dir.name)
if ((_read_sysfs_text(usb_dir / "idVendor") or "").lower() == vid
and (_read_sysfs_text(usb_dir / "idProduct") or "").lower() == pid
and _read_sysfs_text(usb_dir / "serial") == serial):
return False
return True
seen = set()
try:
for tty_dev in _SYS_TTY.iterdir():
if not tty_dev.name.startswith(("ttyUSB", "ttyACM")):
continue
usb_dir = _usb_device_dir_for_tty(tty_dev)
if usb_dir is None or usb_dir.name == own_dir_name or usb_dir.name in seen:
continue
seen.add(usb_dir.name)
if ((_read_sysfs_text(usb_dir / "idVendor") or "").lower() == vid
and (_read_sysfs_text(usb_dir / "idProduct") or "").lower() == pid
and _read_sysfs_text(usb_dir / "serial") == serial):
return False
except OSError:
pass
return True



def identity_for_tty(tty: str) -> Optional[Dict[str, Optional[str]]]:
"""Snapshot the durable USB identity of a live tty node.

Accepts ``/dev/ttyUSB*`` / ``/dev/ttyACM*`` or a symlink to one (e.g. a
``/dev/serial/by-id/...`` path). Returns ``{"vid", "pid", "serial",
"port_path", "interface"}`` or None when the tty is not backed by a USB
device (or is gone).

A serial shared with another live device (clone serials) is recorded as
None: it cannot identify the device, so the snapshot pins to the physical
port instead. Otherwise a later resolution while this device is briefly
off the bus could match a look-alike sibling.
"""
if not tty or not isinstance(tty, str):
return None
Expand All @@ -187,10 +219,13 @@ def identity_for_tty(tty: str) -> Optional[Dict[str, Optional[str]]]:
pid = (_read_sysfs_text(usb_dir / "idProduct") or "").lower()
if not vid or not pid:
return None
serial = _read_sysfs_text(usb_dir / "serial")
if serial and not _serial_is_unique(vid, pid, serial, usb_dir.name):
serial = None
return {
"vid": vid,
"pid": pid,
"serial": _read_sysfs_text(usb_dir / "serial"),
"serial": serial,
"port_path": usb_dir.name,
"interface": _interface_for_tty(tty_dev),
}
Expand All @@ -200,11 +235,13 @@ def resolve_identity(ident) -> Optional[str]:
"""Resolve an ``identity_for_tty`` snapshot back to the live tty node.

Match rules: vid/pid always; interface when both sides know it; then USB
serial when the snapshot has one (falling back to the port path to break
ties between clone adapters sharing a serial), else the physical port
path. Returns ``/dev/tty...`` or None if the device is not (yet) back.
Tolerates arbitrary garbage input — a malformed snapshot resolves to None
rather than raising.
serial when the snapshot has one. A serial that several live devices
share (clone adapters) proves nothing, so the physical port is then
REQUIRED — with the true device absent this returns None so callers keep
retrying instead of grabbing a look-alike sibling. Serial-less snapshots
resolve by physical port path. Returns ``/dev/tty...`` or None if the
device is not (yet) back. Tolerates arbitrary garbage input — a malformed
snapshot resolves to None rather than raising.
"""
if not isinstance(ident, dict):
return None
Expand Down Expand Up @@ -245,10 +282,17 @@ def resolve_identity(ident) -> Optional[str]:
matches = [c for c in candidates if c[2] == serial]
if not matches:
return None
if len(matches) > 1 and port_path:
ported = [c for c in matches if c[1] == port_path]
if ported:
matches = ported
if len(matches) > 1:
# Several live devices share this serial (clone serials): the
# serial proves nothing, so the physical port is REQUIRED. When
# our port's device is absent, fail — the caller keeps retrying —
# rather than grabbing a look-alike sibling mid-re-enumeration.
if not port_path:
return None
for c in matches:
if c[1] == port_path:
return f"/dev/{c[0]}"
return None
return f"/dev/{matches[0][0]}"
if port_path:
for c in candidates:
Expand Down
2 changes: 1 addition & 1 deletion cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,4 +7,4 @@
A Command Line Interface for Lager Data
"""

__version__ = '0.31.8'
__version__ = '0.31.9'
1 change: 1 addition & 0 deletions docs/docs.json
Original file line number Diff line number Diff line change
Expand Up @@ -211,6 +211,7 @@
{
"group": "Version History",
"pages": [
"source/release-notes/v0.31.9",
"source/release-notes/v0.31.8",
"source/release-notes/v0.31.7",
"source/release-notes/v0.31.6",
Expand Down
26 changes: 26 additions & 0 deletions docs/source/release-notes/v0.31.9.mdx
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
---
title: "Version 0.31.9"
description: "July 13, 2026"
---

## <u>Bug Fixes</u>

- **UART reconnect can no longer land on a look-alike adapter with a clone serial.** Many USB-serial adapters ship with a non-unique programmed serial (e.g. several CP210x units all reading "0001"). If such a device dropped mid-session, the v0.31.5 reconnect could match a sibling adapter with the same serial while the real device was still off the bus — attaching the session to the wrong hardware. Identity resolution now treats a serial shared by multiple live devices as untrusted: the physical port must match, and reconnection keeps retrying until the real device returns. New identity snapshots record a bus-duplicated serial as null, pinning the net to its physical port outright. Nets on clone-serial adapters that were enriched under v0.31.5 pick up the corrected snapshot on their next re-save — re-save only while every adapter sits on the tty its net expects, since enrichment snapshots whatever device the stored pin currently points at.

## <u>Installation</u>

To install this version:

```bash
pip install lager-cli==0.31.9
```

To upgrade from a previous version:

```bash
pip install --upgrade lager-cli
```

## Resources

[View Release on PyPI](https://pypi.org/project/lager-cli/0.31.9/)
62 changes: 62 additions & 0 deletions test/unit/box/test_serial_id_cables.py
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,68 @@ def test_resolve_identity_clone_serials_port_tiebreak(self):
ident = serial_id.identity_for_tty("/dev/ttyUSB1")
self.assertEqual(serial_id.resolve_identity(ident), "/dev/ttyUSB1")

# ---- clone-serial regression (v0.31.5 reconnected to a sibling) ---------

def test_identity_for_tty_demotes_clone_serial(self):
# Two live devices share vid/pid/serial: the serial is not identity.
self._add_cable("ttyUSB0", "1-1.2", serial="0001")
self._add_cable("ttyUSB1", "1-1.3", serial="0001")
ident = serial_id.identity_for_tty("/dev/ttyUSB0")
self.assertIsNone(ident["serial"])
self.assertEqual(ident["port_path"], "1-1.2")

def test_identity_for_tty_keeps_unique_serial(self):
self._add_cable("ttyUSB0", "1-1.2", serial="UNIQ-A")
self._add_cable("ttyUSB1", "1-1.3", serial="UNIQ-B")
self.assertEqual(serial_id.identity_for_tty("/dev/ttyUSB0")["serial"],
"UNIQ-A")

def test_identity_for_tty_multi_interface_keeps_serial(self):
# Four ttys of ONE multi-port chip share the device dir; that is not
# a clone serial.
for n in range(4):
self._add_cable(f"ttyUSB{n}", "1-1.3", serial="QUAD", iface=n)
self.assertEqual(serial_id.identity_for_tty("/dev/ttyUSB2")["serial"],
"QUAD")

def test_resolve_identity_clone_absent_never_matches_sibling(self):
# The exact v0.31.5 field failure: a legacy snapshot carrying the
# clone serial, resolved while the true device is off the bus, must
# return None (keep retrying) — NOT a look-alike sibling.
self._add_cable("ttyUSB0", "1-1.2", serial="0001")
self._add_cable("ttyUSB1", "1-1.3", serial="0001")
self._add_cable("ttyUSB2", "1-1.4", serial="0001")
legacy_ident = {"vid": VID, "pid": PID, "serial": "0001",
"port_path": "1-1.4", "interface": 0}
shutil.rmtree(self._sys_tty / "ttyUSB2") # our device drops
self.assertIsNone(serial_id.resolve_identity(legacy_ident))
self._add_cable("ttyUSB5", "1-1.4", serial="0001") # it returns
self.assertEqual(serial_id.resolve_identity(legacy_ident), "/dev/ttyUSB5")

def test_serial_is_unique_tolerates_unreadable_sysfs(self):
# Best-effort walk: an inaccessible /sys/class/tty must not raise and
# assumes the serial is unique (degrades to trusting it).
serial_id._SYS_TTY = Path(self._tmp) / "does-not-exist"
self.assertTrue(serial_id._serial_is_unique(VID, PID, "0001", "1-1.2"))

def test_resolve_identity_clone_serial_without_port_is_unresolvable(self):
self._add_cable("ttyUSB0", "1-1.2", serial="0001")
self._add_cable("ttyUSB1", "1-1.3", serial="0001")
ident = {"vid": VID, "pid": PID, "serial": "0001"}
self.assertIsNone(serial_id.resolve_identity(ident))

def test_reconnect_snapshot_cycle_with_clones(self):
# End-to-end shape of the JUL-16 heal: snapshot (serial demoted),
# device drops (None while absent), returns renumbered on the same
# port (resolved by port).
self._add_cable("ttyUSB0", "1-1.2", serial="0001")
self._add_cable("ttyUSB1", "1-1.3", serial="0001")
ident = serial_id.identity_for_tty("/dev/ttyUSB0")
shutil.rmtree(self._sys_tty / "ttyUSB0")
self.assertIsNone(serial_id.resolve_identity(ident))
self._add_cable("ttyUSB4", "1-1.2", serial="0001")
self.assertEqual(serial_id.resolve_identity(ident), "/dev/ttyUSB4")

def test_resolve_identity_multi_interface_picks_channel(self):
# FT4232H: one USB device (no serial), four ttys on interfaces 0-3.
for n in range(4):
Expand Down
Loading