From c62fb8308b0b5bba48d4b1ddb4674327fd53a97f Mon Sep 17 00:00:00 2001 From: Sungman Cho Date: Sat, 15 Aug 2026 01:50:52 +0900 Subject: [PATCH 1/6] feat(oglo): add minimal USB incident flight logs --- src/syncfield/adapters/oglo/stream.py | 227 +++++++++++++++++- .../unit/adapters/test_oglo_usb_reconnect.py | 115 ++++++++- 2 files changed, 331 insertions(+), 11 deletions(-) diff --git a/src/syncfield/adapters/oglo/stream.py b/src/syncfield/adapters/oglo/stream.py index 3b4603f..6534913 100644 --- a/src/syncfield/adapters/oglo/stream.py +++ b/src/syncfield/adapters/oglo/stream.py @@ -28,6 +28,7 @@ from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Optional +from uuid import uuid4 try: import bleak # type: ignore[import-not-found] @@ -112,6 +113,8 @@ # healthy blip is never reset, while a wedge self-heals in ~4 s end-to-end. _RECONNECT_USB_RESET_AFTER_S = 3.0 _USBDEVFS_RESET = (ord("U") << 8) | 20 +_USB_EVIDENCE_PREFIX = "OGLO_USB_EVIDENCE " +_USB_INCIDENT_HISTORY_LIMIT = 16 def _usb_device_reset(serial_port: str, *, stream_id: str = "") -> bool: @@ -314,6 +317,12 @@ def __init__( # Drop detection across packets (seq_base continuity). self._next_expected_seq: dict[str, int] = {} + # Metadata-only flight recorder. It is emitted to the normal process + # log when an outage occurs; sensor and raw USB bytes never enter it. + self._usb_read_history: deque[tuple[int, int, str, int | None]] = deque( + maxlen=_USB_INCIDENT_HISTORY_LIMIT + ) + # ------------------------------------------------------------------ # Stream SPI — 4-phase lifecycle # ------------------------------------------------------------------ @@ -789,6 +798,74 @@ def _apply_manifest(self, manifest: OgloDeviceManifest) -> None: if manifest.side in ("left", "right"): self._hand = manifest.side + def _log_usb_evidence( + self, + event_type: str, + *, + level: int = logging.INFO, + **fields: Any, + ) -> None: + """Write one metadata-only JSON record to the existing process log.""" + event: dict[str, Any] = { + "event_type": event_type, + "stream_id": self.id, + "serial_port": self._serial_port, + **fields, + } + try: + logger.log( + level, + "%s%s", + _USB_EVIDENCE_PREFIX, + json.dumps(event, sort_keys=True, separators=(",", ":")), + ) + except Exception: # noqa: BLE001 - diagnostics must never affect capture + logger.debug("[%s] failed to encode USB evidence log", self.id, exc_info=True) + + def _record_usb_read( + self, + *, + started_ns: int, + byte_count: int, + diagnostic: str, + exception: BaseException | None = None, + ) -> None: + errno = getattr(exception, "errno", None) if exception is not None else None + if not isinstance(errno, int) or isinstance(errno, bool): + errno = None + duration_us = max(0, (time.monotonic_ns() - started_ns) // 1_000) + self._usb_read_history.append( + (duration_us, byte_count, diagnostic, errno) + ) + + @staticmethod + def _classify_usb_read( + chunk: bytes, + *, + valid_frame_observed: bool, + remainder: bytes, + ) -> str: + if valid_frame_observed: + return "valid_frame" + if remainder.startswith(b"\xA5\x5A"): + return "partial_frame" + if b"\xA5\x5A" in chunk: + return "malformed_header_or_frame" + return "unframed_traffic" + + def _usb_read_history_projection(self) -> list[dict[str, Any]]: + projection: list[dict[str, Any]] = [] + for duration_us, byte_count, classification, errno in self._usb_read_history: + item: dict[str, Any] = { + "duration_us": duration_us, + "byte_count": byte_count, + "classification": classification, + } + if errno is not None: + item["errno"] = errno + projection.append(item) + return projection + # ------------------------------------------------------------------ # Payload decoding (unit-testable without asyncio / bleak) # ------------------------------------------------------------------ @@ -883,7 +960,18 @@ def recover_or_die(reason: str) -> None: spinning quietly. """ nonlocal ser, buffer, last_packet_at - reconnected = self._usb_reconnect(reason=reason) + outage_id = str(uuid4()) + self._log_usb_evidence( + "outage_observed", + level=logging.WARNING, + outage_id=outage_id, + reason=reason, + read_spans=self._usb_read_history_projection(), + ) + reconnected = self._usb_reconnect( + reason=reason, + outage_id=outage_id, + ) if reconnected is None: raise OgloProtocolError( f"{reason} and reconnect did not succeed " @@ -901,9 +989,16 @@ def recover_or_die(reason: str) -> None: try: while not self._stop_event.is_set(): + read_started_ns = time.monotonic_ns() try: chunk = ser.read(4096) - except Exception: + except Exception as exc: + self._record_usb_read( + started_ns=read_started_ns, + byte_count=0, + diagnostic="read_exception", + exception=exc, + ) if not ready: raise OgloProtocolError( "OGLO TAG stream stopped before the first packet" @@ -934,8 +1029,23 @@ def recover_or_die(reason: str) -> None: # heartbeat text forever, so that made the watchdog # permanently blind (ogpi-005 bench, 2026-08-12). last_packet_at = time.monotonic() - elif not ready and time.monotonic() >= ready_deadline: - raise OgloProtocolError("OGLO TAG stream produced no valid packet") + self._record_usb_read( + started_ns=read_started_ns, + byte_count=len(chunk), + diagnostic=self._classify_usb_read( + chunk, + valid_frame_observed=handled > 0, + remainder=buffer, + ), + ) + else: + self._record_usb_read( + started_ns=read_started_ns, + byte_count=0, + diagnostic="zero_byte_read", + ) + if not ready and time.monotonic() >= ready_deadline: + raise OgloProtocolError("OGLO TAG stream produced no valid packet") silent_s = time.monotonic() - last_packet_at if ready and silent_s > _STREAM_SILENCE_TIMEOUT_S: logger.warning( @@ -1005,14 +1115,26 @@ def _try_adopt_reconnected(self, ser: Any) -> bytes | None: ser.flush() except Exception: return None + read_started_ns = time.monotonic_ns() try: chunk = ser.read(4096) - except Exception: + except Exception as exc: + self._record_usb_read( + started_ns=read_started_ns, + byte_count=0, + diagnostic="read_exception", + exception=exc, + ) return None if not chunk: + self._record_usb_read( + started_ns=read_started_ns, + byte_count=0, + diagnostic="zero_byte_read", + ) continue buffer += chunk - packets, _remainder = iter_usb_packets(buffer) + packets, remainder = iter_usb_packets(buffer) # `next(..., None)` — NOT `if packets:`. iter_usb_packets returns # an ITERATOR, and an empty iterator is truthy, so the old check # adopted any handle that produced a single byte. A stopped glove @@ -1020,13 +1142,28 @@ def _try_adopt_reconnected(self, ser: Any) -> bytes | None: # success on a link carrying no samples at all: that is how # ogpi-007 logged "USB link reconnected after 8443 ms" and then # recorded 36 minutes of nothing (2026-08-12). - if next(packets, None) is not None: + packet_found = next(packets, None) is not None + self._record_usb_read( + started_ns=read_started_ns, + byte_count=len(chunk), + diagnostic=self._classify_usb_read( + chunk, + valid_frame_observed=packet_found, + remainder=remainder, + ), + ) + if packet_found: return buffer if len(buffer) > 65_536: return None # a flood that never frames is not a TAG stream return None - def _usb_reconnect(self, reason: str = "USB link lost") -> "tuple[Any, bytes] | None": + def _usb_reconnect( + self, + reason: str = "USB link lost", + *, + outage_id: str, + ) -> "tuple[Any, bytes] | None": """Reopen the stable serial path after a link drop and resume TAG mode. Returns ``(handle, seed_bytes)`` on success, or ``None`` once the @@ -1054,6 +1191,24 @@ def _usb_reconnect(self, reason: str = "USB link lost") -> "tuple[Any, bytes] | import serial as serial_module reset_attempted = False + reset_outcome = "not_attempted" + attempt_no = 0 + open_failures = 0 + adoption_failures = 0 + last_errno: int | None = None + + def reconnect_summary() -> dict[str, Any]: + summary: dict[str, Any] = { + "attempts": attempt_no, + "open_failures": open_failures, + "adoption_failures": adoption_failures, + "reset_attempted": reset_attempted, + "reset_outcome": reset_outcome, + } + if last_errno is not None: + summary["last_errno"] = last_errno + return summary + deadline = outage_started + _RECONNECT_WINDOW_S while not self._stop_event.is_set() and time.monotonic() < deadline: if ( @@ -1062,11 +1217,31 @@ def _usb_reconnect(self, reason: str = "USB link lost") -> "tuple[Any, bytes] | and os.path.exists(self._serial_port) ): reset_attempted = True - _usb_device_reset(self._serial_port, stream_id=self.id) + self._log_usb_evidence( + "usb_reset_requested", + level=logging.WARNING, + outage_id=outage_id, + policy_rule="existing_usb_reset_after_3s", + ) + reset_succeeded = _usb_device_reset( + self._serial_port, stream_id=self.id + ) + reset_outcome = "succeeded" if reset_succeeded else "failed" + self._log_usb_evidence( + "usb_reset_result", + level=logging.WARNING, + outage_id=outage_id, + outcome=reset_outcome, + ) + attempt_no += 1 ser = None try: ser = _open_usb_cdc(serial_module, self._serial_port, timeout=0.1) - except Exception: + except Exception as exc: + open_failures += 1 + errno = getattr(exc, "errno", None) + if isinstance(errno, int) and not isinstance(errno, bool): + last_errno = errno if ser is not None: try: ser.close() @@ -1076,6 +1251,7 @@ def _usb_reconnect(self, reason: str = "USB link lost") -> "tuple[Any, bytes] | continue seed = self._try_adopt_reconnected(ser) if seed is None: + adoption_failures += 1 try: ser.close() except Exception: @@ -1089,6 +1265,7 @@ def _usb_reconnect(self, reason: str = "USB link lost") -> "tuple[Any, bytes] | self._next_expected_seq = {} recording = self._recording outage_ms = int((time.monotonic() - outage_started) * 1000) + loss_fields: dict[str, Any] = {} logger.warning( "[%s] USB link reconnected after %d ms", self.id, outage_ms ) @@ -1117,7 +1294,37 @@ def _usb_reconnect(self, reason: str = "USB link lost") -> "tuple[Any, bytes] | "modality": "tactile", }, )) + loss_fields = { + "estimated_loss_count": estimated, + "loss_estimate_method": "outage_duration_x_nominal_rate", + "loss_modality": "tactile", + } + self._log_usb_evidence( + "recovery_result", + level=logging.WARNING, + outage_id=outage_id, + outcome="recovered", + recovery_path=( + "usb_reset_then_reconnect" + if reset_attempted + else "logical_reconnect" + ), + outage_ms=outage_ms, + reconnect_summary=reconnect_summary(), + **loss_fields, + ) return ser, seed + self._log_usb_evidence( + "recovery_result", + level=logging.ERROR, + outage_id=outage_id, + outcome="aborted" if self._stop_event.is_set() else "not_recovered", + recovery_path=( + "usb_reset_then_reconnect" if reset_attempted else "logical_reconnect" + ), + outage_ms=int((time.monotonic() - outage_started) * 1000), + reconnect_summary=reconnect_summary(), + ) return None def _handle_usb_packet(self, packet: UsbTaggedPacket) -> None: diff --git a/tests/unit/adapters/test_oglo_usb_reconnect.py b/tests/unit/adapters/test_oglo_usb_reconnect.py index 5831a6e..da16a89 100644 --- a/tests/unit/adapters/test_oglo_usb_reconnect.py +++ b/tests/unit/adapters/test_oglo_usb_reconnect.py @@ -13,6 +13,7 @@ """ import importlib +import json import struct import sys import threading @@ -144,6 +145,15 @@ def _wait(predicate, timeout_s=4.0): assert predicate(), "condition not met in time" +def _usb_evidence_records(messages, module): + prefix = module.stream._USB_EVIDENCE_PREFIX + return [ + json.loads(message[len(prefix):]) + for message in messages + if message.startswith(prefix) + ] + + def _connected_recording_stream(module, holder, tmp_path, first): first.feed(tag(TAG_TYPE_TACTILE, 0, 1_000)) stream = module.OgloTactileStream( @@ -277,7 +287,9 @@ def dead_ctor(port=None, baudrate=115200, timeout=0.1, dsrdtr=False, rtscts=Fals stream.disconnect() -def test_present_but_mute_device_escalates_to_one_usb_reset(oglo_usb, tmp_path, monkeypatch): +def test_present_but_mute_device_escalates_to_one_usb_reset( + oglo_usb, tmp_path, monkeypatch +): """The wedge case: enumerated, port opens, firmware mute. Three real occurrences on ogpi-005 (2026-08-10); a kernel USBDEVFS_RESET revived the glove every time where logical replugging did not — so the reconnect loop @@ -294,6 +306,12 @@ def test_present_but_mute_device_escalates_to_one_usb_reset(oglo_usb, tmp_path, module.stream, "_usb_device_reset", lambda port, stream_id="": resets.append(port) or True, ) + messages = [] + monkeypatch.setattr( + module.stream.logger, + "log", + lambda _level, template, *args: messages.append(template % args), + ) first = _FakeSerial("p") first.feed(tag(TAG_TYPE_TACTILE, 0, 1_000)) @@ -310,4 +328,99 @@ def test_present_but_mute_device_escalates_to_one_usb_reset(oglo_usb, tmp_path, assert resets == ["/dev/serial/by-id/oglo-left"], ( "exactly one USB reset per outage" ) + records = _usb_evidence_records(messages, module) + event_types = [event["event_type"] for event in records] + assert "usb_reset_requested" in event_types + assert "usb_reset_result" in event_types + assert "recovery_result" in event_types + assert not any(event_type.startswith("reader_") for event_type in event_types) + incident_records = [event for event in records if "outage_id" in event] + assert len({event["outage_id"] for event in incident_records}) == 1 + assert all("parent_event_id" not in event for event in records) + recovery = next( + event for event in records if event["event_type"] == "recovery_result" + ) + assert recovery["reconnect_summary"]["attempts"] >= 1 + assert recovery["reconnect_summary"]["adoption_failures"] >= 1 + assert recovery["reconnect_summary"]["reset_attempted"] is True + assert recovery["reconnect_summary"]["reset_outcome"] == "succeeded" + assert "reconnect_spans" not in recovery stream.disconnect() + + +def test_usb_flight_log_is_bounded_metadata_only(oglo_usb, tmp_path, monkeypatch): + module, _holder = oglo_usb + stream = module.OgloTactileStream( + "tactile_left", + serial_port="/dev/serial/by-id/oglo-left", + hand="left", + output_dir=tmp_path, + ) + + for index in range(70): + stream._record_usb_read( + started_ns=index, + byte_count=index, + diagnostic="valid_frame", + ) + + assert len(stream._usb_read_history) == 16 + assert all(isinstance(entry, tuple) for entry in stream._usb_read_history) + projection = stream._usb_read_history_projection() + assert len(projection) == 16 + assert projection[0]["byte_count"] == 54 + assert set(projection[0]) == {"duration_us", "byte_count", "classification"} + + messages = [] + monkeypatch.setattr( + module.stream.logger, + "log", + lambda _level, template, *args: messages.append(template % args), + ) + stream._log_usb_evidence( + "outage_observed", + outage_id="outage-test", + read_spans=projection, + ) + + records = _usb_evidence_records(messages, module) + assert records[-1]["event_type"] == "outage_observed" + assert records[-1]["outage_id"] == "outage-test" + for redundant in ( + "event_id", + "origin_seq", + "source_monotonic_ns", + "source_realtime_ns", + "host_boot_id", + "invocation_id", + "parent_event_id", + "schema", + "raw_payload", + ): + assert redundant not in records[-1] + + +def test_usb_read_history_keeps_only_real_errno(oglo_usb, tmp_path): + module, _holder = oglo_usb + stream = module.OgloTactileStream( + "tactile_left", + serial_port="/dev/serial/by-id/oglo-left", + hand="left", + output_dir=tmp_path, + ) + stream._record_usb_read( + started_ns=1, + byte_count=0, + diagnostic="read_exception", + exception=OSError(5, "x" * 800), + ) + stream._record_usb_read( + started_ns=2, + byte_count=0, + diagnostic="zero_byte_read", + ) + + projection = stream._usb_read_history_projection() + assert projection[0]["errno"] == 5 + assert "errno" not in projection[1] + assert all("exception_message" not in item for item in projection) From a27cb126c52ae4af10df8f7486c768838308bb49 Mon Sep 17 00:00:00 2001 From: Sungman Cho Date: Sat, 15 Aug 2026 03:25:47 +0900 Subject: [PATCH 2/6] feat(oglo): correlate MCU identity across USB outages --- src/syncfield/adapters/oglo/stream.py | 243 +++++++++++++++--- .../adapters/test_oglo_silence_watchdog.py | 2 +- .../unit/adapters/test_oglo_usb_reconnect.py | 199 +++++++++++++- 3 files changed, 401 insertions(+), 43 deletions(-) diff --git a/src/syncfield/adapters/oglo/stream.py b/src/syncfield/adapters/oglo/stream.py index 6534913..db0f076 100644 --- a/src/syncfield/adapters/oglo/stream.py +++ b/src/syncfield/adapters/oglo/stream.py @@ -7,9 +7,10 @@ Design highlights: -The connect handshake stops every legacy stream mode, reads ``GET CONFIG``, -requires firmware >=0.9.3/schema 6 and the expected hand, then waits for a valid -TAG packet before reporting ready. Per-modality sequence gaps are health events. +The connect handshake stops every legacy stream mode, reads ``GET CONFIG`` and +``GET IDENT``, requires firmware >=0.9.3/schema 6 and the expected hand, then +waits for a valid TAG packet before reporting ready. Identity failure is logged +but does not gate capture; per-modality sequence gaps are health events. Requires the optional ``ble`` extra:: @@ -85,12 +86,10 @@ # several enumeration cycles plus one USB-reset escalation, short enough that # a genuinely severed cable still protective-stops the recording quickly. # -# The reconnect hot path deliberately skips the GET CONFIG round-trip: the -# by-id path embeds the USB serial, so the device answering on that path IS -# this glove, and every avoided handshake second is 250 lost tactile samples. -# Adoption is proven by actual TAG packets instead: a glove whose MCU stayed -# up is still streaming and talks the moment the port opens; a rebooted one -# sits idle until nudged with STREAM TAG ON after a short silence. +# The reconnect hot path still skips the large GET CONFIG round-trip, but a +# bounded GET IDENT probe records the MCU and flash-journal boot identity before +# TAG mode resumes. Adoption remains gated on an actual TAG packet: identity is +# evidence about the board that answered, not proof that its sensor stream works. _RECONNECT_WINDOW_S = 15.0 _RECONNECT_RETRY_INTERVAL_S = 0.15 # A ready TAG stream delivers 250 Hz tactile plus 500 Hz IMU, so silence this @@ -115,6 +114,83 @@ _USBDEVFS_RESET = (ord("U") << 8) | 20 _USB_EVIDENCE_PREFIX = "OGLO_USB_EVIDENCE " _USB_INCIDENT_HISTORY_LIMIT = 16 +_USB_IDENT_COMMAND = b"GET IDENT\n" +_USB_IDENT_PREFIX = b"#IDENT " +_USB_IDENT_PROBE_TIMEOUT_S = 0.75 +_USB_IDENT_MAX_LINE_BYTES = 4096 + + +@dataclass(frozen=True) +class _UsbIdentityProbe: + identity: dict[str, Any] | None + failure_class: str | None = None + + +@dataclass(frozen=True) +class _UsbAdoption: + seed: bytes | None + identity_probe: _UsbIdentityProbe + + +def _parse_usb_identity(raw: bytes) -> dict[str, Any]: + """Validate and allowlist one firmware ``#IDENT`` JSON object.""" + + try: + decoded = json.loads(raw) + except (UnicodeDecodeError, ValueError) as exc: + raise OgloProtocolError("malformed OGLO USB identity JSON") from exc + if not isinstance(decoded, dict): + raise OgloProtocolError("OGLO USB identity must be a JSON object") + + def required_string(field: str, *, max_chars: int = 256) -> str: + value = decoded.get(field) + if not isinstance(value, str) or not value or len(value) > max_chars: + raise OgloProtocolError(f"OGLO USB identity has invalid {field}") + return value + + def required_int(field: str) -> int: + value = decoded.get(field) + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise OgloProtocolError(f"OGLO USB identity has invalid {field}") + return value + + def required_bool(field: str) -> bool: + value = decoded.get(field) + if not isinstance(value, bool): + raise OgloProtocolError(f"OGLO USB identity has invalid {field}") + return value + + application_sha256 = required_string("application_sha256", max_chars=64) + if len(application_sha256) != 64 or any( + char not in "0123456789abcdef" for char in application_sha256 + ): + raise OgloProtocolError("OGLO USB identity has invalid application_sha256") + + identity: dict[str, Any] = { + "mcu_boot_id": required_string("mcu_boot_id", max_chars=128), + "boot_count": required_int("boot_count"), + "reset_reason": required_string("reset_reason", max_chars=64), + "fw_rev": required_string("fw_rev", max_chars=64), + "hw_rev": required_string("hw_rev", max_chars=128), + "serial": required_string("serial", max_chars=128), + "application_sha256": application_sha256, + "uptime_ms": required_int("uptime_ms"), + "wedge_recoveries": required_int("wedge_recoveries"), + "wedge_last_stall_ms": required_int("wedge_last_stall_ms"), + "wedge_guard": required_bool("wedge_guard"), + "journal_ready": required_bool("journal_ready"), + } + if identity["journal_ready"]: + journal_boot_id = required_string("journal_boot_id", max_chars=16) + if len(journal_boot_id) != 16 or any( + char not in "0123456789abcdef" for char in journal_boot_id + ): + raise OgloProtocolError("OGLO USB identity has invalid journal_boot_id") + identity["journal_boot_id"] = journal_boot_id + identity["journal_boot_counter"] = required_int("journal_boot_counter") + else: + identity["journal_error"] = required_string("journal_error", max_chars=128) + return identity def _usb_device_reset(serial_port: str, *, stream_id: str = "") -> bool: @@ -322,6 +398,10 @@ def __init__( self._usb_read_history: deque[tuple[int, int, str, int | None]] = deque( maxlen=_USB_INCIDENT_HISTORY_LIMIT ) + # The ID is allocated with the identity baseline, before an outage + # exists. If the link later fails, every host action and the after probe + # reuse it. A successful after probe is also the next outage's baseline. + self._pending_outage_id: str | None = None # ------------------------------------------------------------------ # Stream SPI — 4-phase lifecycle @@ -347,6 +427,7 @@ def connect(self) -> None: self._next_expected_seq = {} self._connect_error = None self._manifest = None + self._pending_outage_id = None self._ready_event.clear() self._stop_event.clear() @@ -822,6 +903,81 @@ def _log_usb_evidence( except Exception: # noqa: BLE001 - diagnostics must never affect capture logger.debug("[%s] failed to encode USB evidence log", self.id, exc_info=True) + def _read_usb_identity( + self, + ser: Any, + *, + quiesce_stream: bool, + ) -> _UsbIdentityProbe: + """Return a bounded, payload-free MCU identity probe result.""" + + try: + if quiesce_stream: + ser.write(QUIET_COMMANDS) + ser.flush() + ser.reset_input_buffer() + ser.write(_USB_IDENT_COMMAND) + ser.flush() + deadline = time.monotonic() + _USB_IDENT_PROBE_TIMEOUT_S + while time.monotonic() < deadline and not self._stop_event.is_set(): + line = ser.readline() + if not line: + continue + if len(line) > _USB_IDENT_MAX_LINE_BYTES: + return _UsbIdentityProbe(None, "malformed") + ident_at = line.find(_USB_IDENT_PREFIX) + if ident_at >= 0: + try: + identity = _parse_usb_identity( + line[ident_at + len(_USB_IDENT_PREFIX):].strip() + ) + except OgloProtocolError: + return _UsbIdentityProbe(None, "malformed") + return _UsbIdentityProbe(identity) + if line.startswith((b"#ERR", b"#UNKNOWN")): + return _UsbIdentityProbe(None, "unsupported") + if line.startswith(b"#CONFIG "): + return _UsbIdentityProbe(None, "unexpected_response") + except Exception: # noqa: BLE001 - evidence must not block capture + return _UsbIdentityProbe(None, "io_error") + return _UsbIdentityProbe(None, "aborted" if self._stop_event.is_set() else "timeout") + + def _log_identity_probe( + self, + *, + phase: str, + outage_id: str, + probe: _UsbIdentityProbe, + connection_adopted: bool | None = None, + ) -> None: + fields: dict[str, Any] = { + "phase": phase, + "outage_id": outage_id, + } + if connection_adopted is not None: + fields["connection_adopted"] = connection_adopted + if probe.identity is None: + self._log_usb_evidence( + "identity_probe_failed", + level=logging.WARNING, + failure_class=probe.failure_class or "unknown", + **fields, + ) + return + self._log_usb_evidence( + f"identity_{phase}", + **fields, + **probe.identity, + ) + + def _start_identity_epoch(self, probe: _UsbIdentityProbe) -> None: + self._pending_outage_id = str(uuid4()) + self._log_identity_probe( + phase="before", + outage_id=self._pending_outage_id, + probe=probe, + ) + def _record_usb_read( self, *, @@ -933,6 +1089,9 @@ def _run_usb_reader(self) -> None: time.sleep(0.2) manifest = self._read_usb_manifest(ser) self._apply_manifest(manifest) + self._start_identity_epoch( + self._read_usb_identity(ser, quiesce_stream=False) + ) ser.reset_input_buffer() ser.write(STREAM_ON_COMMAND) ser.flush() @@ -960,7 +1119,14 @@ def recover_or_die(reason: str) -> None: spinning quietly. """ nonlocal ser, buffer, last_packet_at - outage_id = str(uuid4()) + outage_id = self._pending_outage_id or str(uuid4()) + if self._pending_outage_id is None: + self._pending_outage_id = outage_id + self._log_identity_probe( + phase="before", + outage_id=outage_id, + probe=_UsbIdentityProbe(None, "baseline_unavailable"), + ) self._log_usb_evidence( "outage_observed", level=logging.WARNING, @@ -1078,17 +1244,16 @@ def recover_or_die(reason: str) -> None: pass self._serial = None - def _try_adopt_reconnected(self, ser: Any) -> bytes | None: - """Prove a fresh handle carries a live TAG stream; return its bytes. + def _try_adopt_reconnected(self, ser: Any) -> _UsbAdoption: + """Probe identity, then prove a fresh handle carries live TAG data. - Identity needs no handshake here: the stable by-id path embeds the - USB serial, so whatever answers on that path is this glove. A glove - whose MCU stayed up through the link blip is still in TAG mode and - talks immediately; a rebooted one is idle and gets one STREAM TAG ON - nudge after a short silence. Returns every byte read (real samples — - the caller seeds its parse buffer with them), or ``None`` when this - attempt's deadline passes without a valid packet. + The stable by-id path identifies the physical glove. GET IDENT adds the + boot/reset identity needed to explain the outage, while a valid TAG + packet remains the only adoption proof. The identity probe is bounded + and non-gating: old firmware or a malformed response is recorded later + as an evidence gap, then the existing reconnect policy continues. """ + identity_probe = self._read_usb_identity(ser, quiesce_stream=True) # Assert TAG mode unconditionally, before reading a single byte. A # glove that re-enumerated boots idle and will never speak again # unless told to; the old code only nudged when the read buffer was @@ -1099,7 +1264,7 @@ def _try_adopt_reconnected(self, ser: Any) -> bytes | None: ser.write(STREAM_ON_COMMAND) ser.flush() except Exception: - return None + return _UsbAdoption(None, identity_probe) buffer = b"" nudged = False @@ -1107,14 +1272,14 @@ def _try_adopt_reconnected(self, ser: Any) -> bytes | None: while not self._stop_event.is_set(): elapsed = time.monotonic() - started if elapsed >= _RECONNECT_ATTEMPT_DEADLINE_S: - return None + return _UsbAdoption(None, identity_probe) if not nudged and not buffer and elapsed >= _RECONNECT_STREAM_ON_AFTER_S: nudged = True try: ser.write(STREAM_ON_COMMAND) ser.flush() except Exception: - return None + return _UsbAdoption(None, identity_probe) read_started_ns = time.monotonic_ns() try: chunk = ser.read(4096) @@ -1125,7 +1290,7 @@ def _try_adopt_reconnected(self, ser: Any) -> bytes | None: diagnostic="read_exception", exception=exc, ) - return None + return _UsbAdoption(None, identity_probe) if not chunk: self._record_usb_read( started_ns=read_started_ns, @@ -1153,10 +1318,11 @@ def _try_adopt_reconnected(self, ser: Any) -> bytes | None: ), ) if packet_found: - return buffer + return _UsbAdoption(buffer, identity_probe) if len(buffer) > 65_536: - return None # a flood that never frames is not a TAG stream - return None + # A flood that never frames is not a TAG stream. + return _UsbAdoption(None, identity_probe) + return _UsbAdoption(None, identity_probe) def _usb_reconnect( self, @@ -1196,6 +1362,7 @@ def _usb_reconnect( open_failures = 0 adoption_failures = 0 last_errno: int | None = None + last_identity_probe = _UsbIdentityProbe(None, "no_reconnected_handle") def reconnect_summary() -> dict[str, Any]: summary: dict[str, Any] = { @@ -1249,8 +1416,9 @@ def reconnect_summary() -> dict[str, Any]: pass self._stop_event.wait(_RECONNECT_RETRY_INTERVAL_S) continue - seed = self._try_adopt_reconnected(ser) - if seed is None: + adoption = self._try_adopt_reconnected(ser) + last_identity_probe = adoption.identity_probe + if adoption.seed is None: adoption_failures += 1 try: ser.close() @@ -1299,6 +1467,12 @@ def reconnect_summary() -> dict[str, Any]: "loss_estimate_method": "outage_duration_x_nominal_rate", "loss_modality": "tactile", } + self._log_identity_probe( + phase="after", + outage_id=outage_id, + probe=adoption.identity_probe, + connection_adopted=True, + ) self._log_usb_evidence( "recovery_result", level=logging.WARNING, @@ -1313,7 +1487,18 @@ def reconnect_summary() -> dict[str, Any]: reconnect_summary=reconnect_summary(), **loss_fields, ) - return ser, seed + # The recovered board is now the authoritative baseline for the + # next incident, even when its identity probe failed. Preallocating + # the next ID lets that failure be explicit before any outage. + self._start_identity_epoch(adoption.identity_probe) + return ser, adoption.seed + self._log_identity_probe( + phase="after", + outage_id=outage_id, + probe=last_identity_probe, + connection_adopted=False, + ) + self._pending_outage_id = None self._log_usb_evidence( "recovery_result", level=logging.ERROR, diff --git a/tests/unit/adapters/test_oglo_silence_watchdog.py b/tests/unit/adapters/test_oglo_silence_watchdog.py index 5ab4f03..382cd9f 100644 --- a/tests/unit/adapters/test_oglo_silence_watchdog.py +++ b/tests/unit/adapters/test_oglo_silence_watchdog.py @@ -306,6 +306,6 @@ def test_adoption_requires_a_real_packet_not_just_bytes(oglo_usb, tmp_path, monk monkeypatch.setattr(module.stream, "_RECONNECT_ATTEMPT_DEADLINE_S", 0.3) monkeypatch.setattr(module.stream, "_RECONNECT_STREAM_ON_AFTER_S", 0.05) - assert stream._try_adopt_reconnected(chatter) is None, ( + assert stream._try_adopt_reconnected(chatter).seed is None, ( "a port that only chatters must never be adopted as a live stream" ) diff --git a/tests/unit/adapters/test_oglo_usb_reconnect.py b/tests/unit/adapters/test_oglo_usb_reconnect.py index da16a89..922045c 100644 --- a/tests/unit/adapters/test_oglo_usb_reconnect.py +++ b/tests/unit/adapters/test_oglo_usb_reconnect.py @@ -6,10 +6,10 @@ episode silently lost every remaining tactile sample. The reconnect hot path skips the GET CONFIG handshake — identity is pinned by -the stable ``/dev/serial/by-id`` path, which embeds the USB serial — and proves -adoption with actual TAG packets instead, so the data gap stays near the USB -re-enumeration floor. A device that is back but mute (wedged ESP32-S3 CDC -stack) gets one kernel USB reset mid-window. +the stable ``/dev/serial/by-id`` path, which embeds the USB serial — but performs +a bounded GET IDENT probe before proving adoption with an actual TAG packet. A +device that is back but mute (wedged ESP32-S3 CDC stack) gets one kernel USB +reset mid-window. """ import importlib @@ -37,6 +37,31 @@ class _SerialError(Exception): """Stands in for serial.SerialException without importing pyserial.""" +def _identity( + *, + mcu_boot_id="mcu-boot-1", + journal_boot_id="0123456789abcdef", + journal_boot_counter=7, + reset_reason="poweron", +): + return { + "mcu_boot_id": mcu_boot_id, + "boot_count": journal_boot_counter, + "reset_reason": reset_reason, + "fw_rev": "0.9.13", + "hw_rev": "RDR02_FLEX5_REV_D_TIA", + "serial": "OGLO-TEST-L", + "application_sha256": "ab" * 32, + "uptime_ms": 1234, + "wedge_recoveries": 0, + "wedge_last_stall_ms": 0, + "wedge_guard": False, + "journal_ready": True, + "journal_boot_counter": journal_boot_counter, + "journal_boot_id": journal_boot_id, + } + + class _FakeSerial: """Streams queued TAG packets, then optionally dies like a USB unplug. @@ -44,7 +69,14 @@ class _FakeSerial: outage: it stays silent until the host writes STREAM TAG ON. """ - def __init__(self, port, *, stream_on_gated=False): + def __init__( + self, + port, + *, + stream_on_gated=False, + identity=None, + identity_failure=None, + ): self.port, self.writes, self.closed = port, [], False self.dtr = True self.rts = True @@ -55,6 +87,8 @@ def __init__(self, port, *, stream_on_gated=False): self._to_read = b"" self._dead = False self._gated = stream_on_gated + self.identity = identity or _identity() + self.identity_failure = identity_failure self.fail_open = False def open(self): @@ -96,8 +130,15 @@ def read(self, n): return b"" def readline(self): - import json - + if self.writes and self.writes[-1] == b"GET IDENT\n": + if self.identity_failure == "timeout": + time.sleep(0.002) + return b"" + if self.identity_failure == "unsupported": + return b"#ERR unknown command\n" + if self.identity_failure == "malformed": + return b"#IDENT {not-json}\n" + return b"#IDENT " + json.dumps(self.identity).encode() + b"\n" cfg = { "device": "oglo", "schema_ver": 6, "side": "left", "serial": "OGLO-TEST-L", "fw_rev": "0.9.3", "rate_hz": 250, @@ -169,14 +210,20 @@ def _connected_recording_stream(module, holder, tmp_path, first): return stream, health -def test_still_streaming_glove_is_adopted_without_any_handshake(oglo_usb, tmp_path): - """MCU survived the blip: TAG frames flow the moment the port reopens. - No GET CONFIG round-trip may appear on the new handle — every avoided - handshake second is 250 lost tactile samples.""" +def test_still_streaming_glove_records_same_identity_across_outage( + oglo_usb, tmp_path, monkeypatch +): + """MCU survived the blip: before/after identity is exactly joinable.""" module, holder = oglo_usb first, second = _FakeSerial("p"), _FakeSerial("p") second.feed(tag(TAG_TYPE_TACTILE, 500, 60_000)) holder["queue"] = [first, second] + messages = [] + monkeypatch.setattr( + module.stream.logger, + "log", + lambda _level, template, *args: messages.append(template % args), + ) stream, health = _connected_recording_stream(module, holder, tmp_path, first) first.kill() @@ -186,6 +233,25 @@ def test_still_streaming_glove_is_adopted_without_any_handshake(oglo_usb, tmp_pa assert not any(b"GET CONFIG" in w for w in second.writes), ( "reconnect hot path must not spend time on a config handshake" ) + assert any(b"GET IDENT" in w for w in second.writes) + records = _usb_evidence_records(messages, module) + outage = next(event for event in records if event["event_type"] == "outage_observed") + before = next( + event + for event in records + if event["event_type"] == "identity_before" + and event["outage_id"] == outage["outage_id"] + ) + after = next( + event + for event in records + if event["event_type"] == "identity_after" + and event["outage_id"] == outage["outage_id"] + ) + assert before["stream_id"] == after["stream_id"] == "tactile_left" + assert before["mcu_boot_id"] == after["mcu_boot_id"] + assert before["journal_boot_id"] == after["journal_boot_id"] + assert after["connection_adopted"] is True kinds = [h.kind for h in health] assert HealthEventKind.WARNING in kinds assert any("reconnect" in h.detail.lower() for h in health) @@ -195,23 +261,130 @@ def test_still_streaming_glove_is_adopted_without_any_handshake(oglo_usb, tmp_pa assert report.frame_count == 2 -def test_rebooted_glove_gets_a_stream_on_nudge(oglo_usb, tmp_path): +def test_rebooted_glove_gets_a_stream_on_nudge(oglo_usb, tmp_path, monkeypatch): """MCU rebooted during the outage: silent until STREAM TAG ON.""" module, holder = oglo_usb first = _FakeSerial("p") - second = _FakeSerial("p", stream_on_gated=True) + second = _FakeSerial( + "p", + stream_on_gated=True, + identity=_identity( + mcu_boot_id="mcu-boot-2", + journal_boot_id="fedcba9876543210", + journal_boot_counter=8, + reset_reason="software", + ), + ) second.feed(tag(TAG_TYPE_TACTILE, 3, 9_000)) holder["queue"] = [first, second] + messages = [] + monkeypatch.setattr( + module.stream.logger, + "log", + lambda _level, template, *args: messages.append(template % args), + ) stream, _health = _connected_recording_stream(module, holder, tmp_path, first) first.kill() _wait(lambda: stream._frame_count >= 2) assert any(b"STREAM TAG ON" in w for w in second.writes) + records = _usb_evidence_records(messages, module) + outage = next(event for event in records if event["event_type"] == "outage_observed") + after = next( + event + for event in records + if event["event_type"] == "identity_after" + and event["outage_id"] == outage["outage_id"] + ) + assert after["mcu_boot_id"] == "mcu-boot-2" + assert after["journal_boot_id"] == "fedcba9876543210" + assert after["journal_boot_counter"] == 8 + assert after["reset_reason"] == "software" + assert after["application_sha256"] == "ab" * 32 stream.stop_recording() stream.disconnect() +def test_initial_identity_probe_failure_is_explicit_and_non_gating( + oglo_usb, tmp_path, monkeypatch +): + module, holder = oglo_usb + first = _FakeSerial("p", identity_failure="unsupported") + first.feed(tag(TAG_TYPE_TACTILE, 0, 1_000)) + holder["queue"] = [first] + messages = [] + monkeypatch.setattr( + module.stream.logger, + "log", + lambda _level, template, *args: messages.append(template % args), + ) + + stream = module.OgloTactileStream( + "tactile_left", + serial_port="/dev/serial/by-id/oglo-left", + hand="left", + output_dir=tmp_path, + ) + stream.connect() + records = _usb_evidence_records(messages, module) + failure = next( + event for event in records if event["event_type"] == "identity_probe_failed" + ) + assert failure["phase"] == "before" + assert failure["failure_class"] == "unsupported" + assert failure["outage_id"] + assert stream._thread.is_alive() + stream.disconnect() + + +def test_reconnect_identity_failure_keeps_recovery_and_outage_join( + oglo_usb, tmp_path, monkeypatch +): + module, holder = oglo_usb + first = _FakeSerial("p") + second = _FakeSerial("p", identity_failure="malformed") + second.feed(tag(TAG_TYPE_TACTILE, 9, 9_000)) + holder["queue"] = [first, second] + messages = [] + monkeypatch.setattr( + module.stream.logger, + "log", + lambda _level, template, *args: messages.append(template % args), + ) + + stream, _health = _connected_recording_stream(module, holder, tmp_path, first) + first.kill() + _wait(lambda: stream._frame_count >= 2) + records = _usb_evidence_records(messages, module) + outage = next(event for event in records if event["event_type"] == "outage_observed") + failure = next( + event + for event in records + if event["event_type"] == "identity_probe_failed" + and event["phase"] == "after" + ) + assert failure["outage_id"] == outage["outage_id"] + assert failure["failure_class"] == "malformed" + assert failure["connection_adopted"] is True + assert any( + event["event_type"] == "recovery_result" + and event["outcome"] == "recovered" + for event in records + ) + stream.stop_recording() + stream.disconnect() + + +def test_identity_parser_requires_signed_application_hash(oglo_usb): + module, _holder = oglo_usb + identity = _identity() + identity.pop("application_sha256") + + with pytest.raises(module.stream.OgloProtocolError, match="application_sha256"): + module.stream._parse_usb_identity(json.dumps(identity).encode()) + + def test_reconnect_reports_outage_as_estimated_drop(oglo_usb, tmp_path): module, holder = oglo_usb first, second = _FakeSerial("p"), _FakeSerial("p") From 349024cd403baccaae2d777a1eec2e2e1ea42449 Mon Sep 17 00:00:00 2001 From: Sungman Cho Date: Sat, 15 Aug 2026 04:24:25 +0900 Subject: [PATCH 3/6] feat(oglo): retain pre-outage IMU evidence --- src/syncfield/adapters/oglo/stream.py | 69 ++++++++++++++++ .../unit/adapters/test_oglo_usb_reconnect.py | 78 ++++++++++++++++++- 2 files changed, 144 insertions(+), 3 deletions(-) diff --git a/src/syncfield/adapters/oglo/stream.py b/src/syncfield/adapters/oglo/stream.py index db0f076..c4f1c6b 100644 --- a/src/syncfield/adapters/oglo/stream.py +++ b/src/syncfield/adapters/oglo/stream.py @@ -22,6 +22,7 @@ import asyncio import json import logging +import math import os import threading import time @@ -118,6 +119,15 @@ _USB_IDENT_PREFIX = b"#IDENT " _USB_IDENT_PROBE_TIMEOUT_S = 0.75 _USB_IDENT_MAX_LINE_BYTES = 4096 +# The on-board ICM-42688P/TDK accelerometer is configured for +/-8 g in the +# matching firmware. Keep a short, bounded host-side window so a complete MCU +# power loss cannot erase the motion immediately preceding a USB outage. This +# is correlation evidence, not a cause classifier: gravity contributes roughly +# 1,000 mg and a cable tug, impact, or ordinary hand motion can look alike. +_PRE_OUTAGE_IMU_WINDOW_NS = 5_000_000_000 +_PRE_OUTAGE_IMU_MAX_SAMPLES = 3_000 +_IMU_ACCEL_SCALE_MG_PER_LSB = 0.244 +_IMU_ACCEL_FULL_SCALE_G = 8 @dataclass(frozen=True) @@ -388,6 +398,9 @@ def __init__( self._substream_callbacks: list[Callable[[Any], None]] = [] self._imu_recent_ns: deque[int] = deque(maxlen=120) self._mag_recent_ns: deque[int] = deque(maxlen=120) + self._pre_outage_accel: deque[tuple[int, int, bool]] = deque( + maxlen=_PRE_OUTAGE_IMU_MAX_SAMPLES + ) self._imu_recent_lock = threading.Lock() # Drop detection across packets (seq_base continuity). @@ -428,6 +441,8 @@ def connect(self) -> None: self._connect_error = None self._manifest = None self._pending_outage_id = None + with self._imu_recent_lock: + self._pre_outage_accel.clear() self._ready_event.clear() self._stop_event.clear() @@ -1022,6 +1037,54 @@ def _usb_read_history_projection(self) -> list[dict[str, Any]]: projection.append(item) return projection + def _take_pre_outage_imu_projection(self, outage_at_ns: int) -> dict[str, Any]: + """Drain and summarize recent acceleration without retaining samples. + + The evidence stays metadata-only: only a count, peak magnitude, age, + and saturation bit are logged. Raw IMU samples remain in the normal + sensor artifact when recording is active. + """ + + cutoff_ns = outage_at_ns - _PRE_OUTAGE_IMU_WINDOW_NS + with self._imu_recent_lock: + recent = [ + sample + for sample in self._pre_outage_accel + if sample[0] >= cutoff_ns + ] + self._pre_outage_accel.clear() + + window_ms = _PRE_OUTAGE_IMU_WINDOW_NS // 1_000_000 + if not recent: + return { + "status": "unavailable", + "window_ms": window_ms, + "reason": "no_recent_imu_sample", + } + + peak_at_ns, peak_norm_sq, peak_saturated = max( + recent, key=lambda sample: sample[1] + ) + latest_at_ns = recent[-1][0] + return { + "status": "observed", + "window_ms": window_ms, + "sample_count": len(recent), + "latest_sample_age_ms": max( + 0, (outage_at_ns - latest_at_ns) // 1_000_000 + ), + "peak_sample_age_ms": max( + 0, (outage_at_ns - peak_at_ns) // 1_000_000 + ), + "peak_resultant_mg": round( + math.sqrt(peak_norm_sq) * _IMU_ACCEL_SCALE_MG_PER_LSB + ), + "peak_axis_saturated": peak_saturated, + "accelerometer_full_scale_g": _IMU_ACCEL_FULL_SCALE_G, + "scale_mg_per_lsb": _IMU_ACCEL_SCALE_MG_PER_LSB, + "classification": "evidence_only_not_cause", + } + # ------------------------------------------------------------------ # Payload decoding (unit-testable without asyncio / bleak) # ------------------------------------------------------------------ @@ -1119,6 +1182,7 @@ def recover_or_die(reason: str) -> None: spinning quietly. """ nonlocal ser, buffer, last_packet_at + outage_at_ns = time.monotonic_ns() outage_id = self._pending_outage_id or str(uuid4()) if self._pending_outage_id is None: self._pending_outage_id = outage_id @@ -1133,6 +1197,7 @@ def recover_or_die(reason: str) -> None: outage_id=outage_id, reason=reason, read_spans=self._usb_read_history_projection(), + pre_outage_imu=self._take_pre_outage_imu_projection(outage_at_ns), ) reconnected = self._usb_reconnect( reason=reason, @@ -1612,8 +1677,12 @@ def _handle_imu( device_ns: int, ) -> None: """Fan a wrist-IMU sample out to the live handler + substream file.""" + ax, ay, az = imu[:3] + accel_norm_sq = ax * ax + ay * ay + az * az + accel_saturated = any(abs(axis) >= 32_767 for axis in (ax, ay, az)) with self._imu_recent_lock: self._imu_recent_ns.append(recv_ns) + self._pre_outage_accel.append((recv_ns, accel_norm_sq, accel_saturated)) channels = {name: int(v) for name, v in zip(IMU_CHANNELS, imu)} with self._recording_lock: diff --git a/tests/unit/adapters/test_oglo_usb_reconnect.py b/tests/unit/adapters/test_oglo_usb_reconnect.py index 922045c..b4566e5 100644 --- a/tests/unit/adapters/test_oglo_usb_reconnect.py +++ b/tests/unit/adapters/test_oglo_usb_reconnect.py @@ -23,13 +23,16 @@ import pytest -from syncfield.adapters.oglo.usb_packet import TAG_MAGIC, TAG_TYPE_TACTILE +from syncfield.adapters.oglo.usb_packet import TAG_MAGIC, TAG_TYPE_IMU, TAG_TYPE_TACTILE from syncfield.clock import SessionClock from syncfield.types import HealthEventKind, SyncPoint -def tag(kind: int, seq: int, t_us: int) -> bytes: - payload = struct.pack("<80H", *range(80)) +def tag(kind: int, seq: int, t_us: int, values=None) -> bytes: + if kind == TAG_TYPE_IMU: + payload = struct.pack("<6h", *(values or (0, 0, 4096, 0, 0, 0))) + else: + payload = struct.pack("<80H", *(values or range(80))) return TAG_MAGIC + bytes([kind]) + struct.pack("= 2) + + records = _usb_evidence_records(messages, module) + outage = next(event for event in records if event["event_type"] == "outage_observed") + imu = outage["pre_outage_imu"] + assert imu["status"] == "observed" + assert imu["window_ms"] == 5_000 + assert imu["sample_count"] == 2 + assert 3_000 <= imu["peak_resultant_mg"] <= 3_200 + assert imu["peak_axis_saturated"] is False + assert imu["accelerometer_full_scale_g"] == 8 + assert imu["classification"] == "evidence_only_not_cause" + assert not stream._pre_outage_accel + + stream.stop_recording() + stream.disconnect() + + +def test_outage_marks_pre_outage_imu_unavailable_when_no_sample_arrived( + oglo_usb, tmp_path, monkeypatch +): + module, holder = oglo_usb + first, second = _FakeSerial("p"), _FakeSerial("p") + second.feed(tag(TAG_TYPE_TACTILE, 500, 60_000)) + holder["queue"] = [first, second] + messages = [] + monkeypatch.setattr( + module.stream.logger, + "log", + lambda _level, template, *args: messages.append(template % args), + ) + + stream, _health = _connected_recording_stream(module, holder, tmp_path, first) + first.kill() + _wait(lambda: stream._frame_count >= 2) + + records = _usb_evidence_records(messages, module) + outage = next(event for event in records if event["event_type"] == "outage_observed") + assert outage["pre_outage_imu"] == { + "reason": "no_recent_imu_sample", + "status": "unavailable", + "window_ms": 5_000, + } + + stream.stop_recording() + stream.disconnect() + + def test_rebooted_glove_gets_a_stream_on_nudge(oglo_usb, tmp_path, monkeypatch): """MCU rebooted during the outage: silent until STREAM TAG ON.""" module, holder = oglo_usb From 0c93a3a13e8436cce5c1c82754f931b5a42f0118 Mon Sep 17 00:00:00 2001 From: Sungman Cho Date: Sat, 15 Aug 2026 21:41:57 +0900 Subject: [PATCH 4/6] feat(oglo): version USB evidence and identity contract --- src/syncfield/adapters/oglo/stream.py | 102 ++++++++++++++++-- .../unit/adapters/oglo/ident_contract_v1.json | 18 ++++ .../unit/adapters/test_oglo_usb_reconnect.py | 80 ++++++++++++-- 3 files changed, 185 insertions(+), 15 deletions(-) create mode 100644 tests/unit/adapters/oglo/ident_contract_v1.json diff --git a/src/syncfield/adapters/oglo/stream.py b/src/syncfield/adapters/oglo/stream.py index c4f1c6b..07736fa 100644 --- a/src/syncfield/adapters/oglo/stream.py +++ b/src/syncfield/adapters/oglo/stream.py @@ -114,6 +114,8 @@ _RECONNECT_USB_RESET_AFTER_S = 3.0 _USBDEVFS_RESET = (ord("U") << 8) | 20 _USB_EVIDENCE_PREFIX = "OGLO_USB_EVIDENCE " +_USB_EVIDENCE_SCHEMA_VERSION = "oglo.usb_evidence.v1" +_USB_IDENT_SCHEMA_VERSION = 1 _USB_INCIDENT_HISTORY_LIMIT = 16 _USB_IDENT_COMMAND = b"GET IDENT\n" _USB_IDENT_PREFIX = b"#IDENT " @@ -170,26 +172,69 @@ def required_bool(field: str) -> bool: raise OgloProtocolError(f"OGLO USB identity has invalid {field}") return value - application_sha256 = required_string("application_sha256", max_chars=64) - if len(application_sha256) != 64 or any( - char not in "0123456789abcdef" for char in application_sha256 + ident_schema = decoded.get("ident_schema", 0) + if ( + not isinstance(ident_schema, int) + or isinstance(ident_schema, bool) + or ident_schema not in (0, _USB_IDENT_SCHEMA_VERSION) ): - raise OgloProtocolError("OGLO USB identity has invalid application_sha256") + raise OgloProtocolError("OGLO USB identity has unsupported ident_schema") + + mcu_boot_id = required_string("mcu_boot_id", max_chars=32) + if len(mcu_boot_id) != 32 or any( + char not in "0123456789abcdef" for char in mcu_boot_id + ): + raise OgloProtocolError("OGLO USB identity has invalid mcu_boot_id") identity: dict[str, Any] = { - "mcu_boot_id": required_string("mcu_boot_id", max_chars=128), + "ident_schema": ident_schema, + "mcu_boot_id": mcu_boot_id, "boot_count": required_int("boot_count"), "reset_reason": required_string("reset_reason", max_chars=64), "fw_rev": required_string("fw_rev", max_chars=64), "hw_rev": required_string("hw_rev", max_chars=128), "serial": required_string("serial", max_chars=128), - "application_sha256": application_sha256, "uptime_ms": required_int("uptime_ms"), "wedge_recoveries": required_int("wedge_recoveries"), "wedge_last_stall_ms": required_int("wedge_last_stall_ms"), "wedge_guard": required_bool("wedge_guard"), - "journal_ready": required_bool("journal_ready"), } + + application_sha256 = decoded.get("application_sha256") + if application_sha256 is None: + if ident_schema == _USB_IDENT_SCHEMA_VERSION: + status = required_string("application_sha256_status", max_chars=32) + if status != "unavailable": + raise OgloProtocolError( + "OGLO USB identity has invalid application_sha256_status" + ) + identity["application_sha256_status"] = "unavailable" + else: + if ( + not isinstance(application_sha256, str) + or len(application_sha256) != 64 + or any(char not in "0123456789abcdef" for char in application_sha256) + ): + raise OgloProtocolError("OGLO USB identity has invalid application_sha256") + if ident_schema == _USB_IDENT_SCHEMA_VERSION: + status = required_string("application_sha256_status", max_chars=32) + if status != "available": + raise OgloProtocolError( + "OGLO USB identity has invalid application_sha256_status" + ) + identity["application_sha256_status"] = "available" + identity["application_sha256"] = application_sha256 + + if "journal_ready" not in decoded: + if ident_schema == _USB_IDENT_SCHEMA_VERSION: + raise OgloProtocolError("OGLO USB identity has invalid journal_ready") + identity["journal_status"] = "unavailable" + return identity + + identity["journal_ready"] = required_bool("journal_ready") + identity["journal_status"] = ( + "available" if identity["journal_ready"] else "invalid" + ) if identity["journal_ready"]: journal_boot_id = required_string("journal_boot_id", max_chars=16) if len(journal_boot_id) != 16 or any( @@ -203,6 +248,40 @@ def required_bool(field: str) -> bool: return identity +def _usb_physical_identity(serial_port: str) -> dict[str, str | None]: + """Best-effort stable USB identity for joining host and adapter evidence.""" + + identity: dict[str, str | None] = { + "usb_serial": None, + "usb_physical_path": None, + "usb_controller": None, + } + try: + tty_name = Path(os.path.realpath(serial_port)).name + resolved = Path(os.path.realpath(f"/sys/class/tty/{tty_name}/device")) + device_dir = next( + candidate + for candidate in (resolved, *resolved.parents) + if (candidate / "idVendor").is_file() + and (candidate / "idProduct").is_file() + ) + identity["usb_physical_path"] = device_dir.name + serial_path = device_dir / "serial" + if serial_path.is_file(): + identity["usb_serial"] = serial_path.read_text().strip() or None + for candidate in (device_dir, *device_dir.parents): + if candidate.name.startswith("xhci-hcd."): + identity["usb_controller"] = candidate.name + break + driver = candidate / "driver" + if driver.exists() and driver.resolve().name == "xhci_hcd": + identity["usb_controller"] = candidate.name + break + except (OSError, RuntimeError, StopIteration, UnicodeError): + pass + return identity + + def _usb_device_reset(serial_port: str, *, stream_id: str = "") -> bool: """Best-effort kernel-level reset of the USB device behind *serial_port*. @@ -411,6 +490,7 @@ def __init__( self._usb_read_history: deque[tuple[int, int, str, int | None]] = deque( maxlen=_USB_INCIDENT_HISTORY_LIMIT ) + self._usb_evidence_identity = _usb_physical_identity(self._serial_port) # The ID is allocated with the identity baseline, before an outage # exists. If the link later fails, every host action and the after probe # reuse it. A successful after probe is also the next outage's baseline. @@ -902,10 +982,18 @@ def _log_usb_evidence( **fields: Any, ) -> None: """Write one metadata-only JSON record to the existing process log.""" + current_identity = _usb_physical_identity(self._serial_port) + for key, value in current_identity.items(): + if value is not None: + self._usb_evidence_identity[key] = value event: dict[str, Any] = { + "schema_version": _USB_EVIDENCE_SCHEMA_VERSION, "event_type": event_type, + "source_monotonic_ns": time.monotonic_ns(), + "source_realtime_ns": time.time_ns(), "stream_id": self.id, "serial_port": self._serial_port, + **self._usb_evidence_identity, **fields, } try: diff --git a/tests/unit/adapters/oglo/ident_contract_v1.json b/tests/unit/adapters/oglo/ident_contract_v1.json new file mode 100644 index 0000000..3a85cd4 --- /dev/null +++ b/tests/unit/adapters/oglo/ident_contract_v1.json @@ -0,0 +1,18 @@ +{ + "ident_schema": 1, + "mcu_boot_id": "00112233445566778899aabbccddeeff", + "boot_count": 7, + "reset_reason": "poweron", + "fw_rev": "0.9.14", + "hw_rev": "RDR02_FLEX5_REV_D_TIA", + "serial": "OGLO-TEST-L", + "application_sha256_status": "available", + "application_sha256": "abababababababababababababababababababababababababababababababab", + "uptime_ms": 1234, + "wedge_recoveries": 0, + "wedge_last_stall_ms": 0, + "wedge_guard": false, + "journal_ready": true, + "journal_boot_counter": 7, + "journal_boot_id": "0123456789abcdef" +} diff --git a/tests/unit/adapters/test_oglo_usb_reconnect.py b/tests/unit/adapters/test_oglo_usb_reconnect.py index b4566e5..3a1e828 100644 --- a/tests/unit/adapters/test_oglo_usb_reconnect.py +++ b/tests/unit/adapters/test_oglo_usb_reconnect.py @@ -18,6 +18,7 @@ import sys import threading import time +from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock @@ -42,18 +43,20 @@ class _SerialError(Exception): def _identity( *, - mcu_boot_id="mcu-boot-1", + mcu_boot_id="00112233445566778899aabbccddeeff", journal_boot_id="0123456789abcdef", journal_boot_counter=7, reset_reason="poweron", ): return { + "ident_schema": 1, "mcu_boot_id": mcu_boot_id, "boot_count": journal_boot_counter, "reset_reason": reset_reason, - "fw_rev": "0.9.13", + "fw_rev": "0.9.14", "hw_rev": "RDR02_FLEX5_REV_D_TIA", "serial": "OGLO-TEST-L", + "application_sha256_status": "available", "application_sha256": "ab" * 32, "uptime_ms": 1234, "wedge_recoveries": 0, @@ -341,7 +344,7 @@ def test_rebooted_glove_gets_a_stream_on_nudge(oglo_usb, tmp_path, monkeypatch): "p", stream_on_gated=True, identity=_identity( - mcu_boot_id="mcu-boot-2", + mcu_boot_id="ffeeddccbbaa99887766554433221100", journal_boot_id="fedcba9876543210", journal_boot_counter=8, reset_reason="software", @@ -369,7 +372,7 @@ def test_rebooted_glove_gets_a_stream_on_nudge(oglo_usb, tmp_path, monkeypatch): if event["event_type"] == "identity_after" and event["outage_id"] == outage["outage_id"] ) - assert after["mcu_boot_id"] == "mcu-boot-2" + assert after["mcu_boot_id"] == "ffeeddccbbaa99887766554433221100" assert after["journal_boot_id"] == "fedcba9876543210" assert after["journal_boot_counter"] == 8 assert after["reset_reason"] == "software" @@ -448,7 +451,7 @@ def test_reconnect_identity_failure_keeps_recovery_and_outage_join( stream.disconnect() -def test_identity_parser_requires_signed_application_hash(oglo_usb): +def test_identity_parser_rejects_a_hash_status_that_claims_missing_bytes(oglo_usb): module, _holder = oglo_usb identity = _identity() identity.pop("application_sha256") @@ -457,6 +460,64 @@ def test_identity_parser_requires_signed_application_hash(oglo_usb): module.stream._parse_usb_identity(json.dumps(identity).encode()) +def test_identity_parser_preserves_explicitly_unavailable_application_hash(oglo_usb): + module, _holder = oglo_usb + identity = _identity() + identity.pop("application_sha256") + identity["application_sha256_status"] = "unavailable" + + parsed = module.stream._parse_usb_identity(json.dumps(identity).encode()) + + assert parsed["application_sha256_status"] == "unavailable" + assert "application_sha256" not in parsed + + +def test_identity_parser_accepts_the_hand_flashed_0913_legacy_shape(oglo_usb): + module, _holder = oglo_usb + identity = _identity() + identity.pop("ident_schema") + identity.pop("application_sha256_status") + identity.pop("application_sha256") + identity.pop("journal_ready") + identity.pop("journal_boot_counter") + identity.pop("journal_boot_id") + identity["fw_rev"] = "0.9.13" + + parsed = module.stream._parse_usb_identity(json.dumps(identity).encode()) + + assert parsed["ident_schema"] == 0 + assert parsed["application_sha256_status"] == "unavailable" + assert parsed["journal_status"] == "unavailable" + + +def test_identity_parser_matches_the_shared_v1_golden_vector(oglo_usb): + module, _holder = oglo_usb + fixture = ( + Path(__file__).with_name("oglo") / "ident_contract_v1.json" + ).read_bytes() + + parsed = module.stream._parse_usb_identity(fixture) + + assert parsed["ident_schema"] == 1 + assert parsed["mcu_boot_id"] == "00112233445566778899aabbccddeeff" + assert parsed["application_sha256_status"] == "available" + assert parsed["journal_status"] == "available" + + +@pytest.mark.parametrize( + "mcu_boot_id", + ("mcu-boot-1", "A" * 32, "a" * 31, "a" * 33), +) +def test_identity_parser_rejects_noncanonical_mcu_boot_id( + oglo_usb, mcu_boot_id +): + module, _holder = oglo_usb + identity = _identity(mcu_boot_id=mcu_boot_id) + + with pytest.raises(module.stream.OgloProtocolError, match="mcu_boot_id"): + module.stream._parse_usb_identity(json.dumps(identity).encode()) + + def test_reconnect_reports_outage_as_estimated_drop(oglo_usb, tmp_path): module, holder = oglo_usb first, second = _FakeSerial("p"), _FakeSerial("p") @@ -631,15 +692,18 @@ def test_usb_flight_log_is_bounded_metadata_only(oglo_usb, tmp_path, monkeypatch records = _usb_evidence_records(messages, module) assert records[-1]["event_type"] == "outage_observed" assert records[-1]["outage_id"] == "outage-test" + assert records[-1]["schema_version"] == "oglo.usb_evidence.v1" + assert records[-1]["source_monotonic_ns"] > 0 + assert records[-1]["source_realtime_ns"] > 0 + assert "usb_serial" in records[-1] + assert "usb_physical_path" in records[-1] + assert "usb_controller" in records[-1] for redundant in ( "event_id", "origin_seq", - "source_monotonic_ns", - "source_realtime_ns", "host_boot_id", "invocation_id", "parent_event_id", - "schema", "raw_payload", ): assert redundant not in records[-1] From 12dae1016a8c12e5a99f32032c6ec837547440e9 Mon Sep 17 00:00:00 2001 From: Sungman Cho Date: Sat, 15 Aug 2026 22:04:00 +0900 Subject: [PATCH 5/6] refactor(oglo): keep USB evidence metadata additive --- src/syncfield/adapters/oglo/stream.py | 42 +++++++------------ ...t_contract_v1.json => ident_metadata.json} | 1 - .../unit/adapters/test_oglo_usb_reconnect.py | 9 +--- 3 files changed, 18 insertions(+), 34 deletions(-) rename tests/unit/adapters/oglo/{ident_contract_v1.json => ident_metadata.json} (96%) diff --git a/src/syncfield/adapters/oglo/stream.py b/src/syncfield/adapters/oglo/stream.py index 07736fa..6ecb312 100644 --- a/src/syncfield/adapters/oglo/stream.py +++ b/src/syncfield/adapters/oglo/stream.py @@ -114,8 +114,6 @@ _RECONNECT_USB_RESET_AFTER_S = 3.0 _USBDEVFS_RESET = (ord("U") << 8) | 20 _USB_EVIDENCE_PREFIX = "OGLO_USB_EVIDENCE " -_USB_EVIDENCE_SCHEMA_VERSION = "oglo.usb_evidence.v1" -_USB_IDENT_SCHEMA_VERSION = 1 _USB_INCIDENT_HISTORY_LIMIT = 16 _USB_IDENT_COMMAND = b"GET IDENT\n" _USB_IDENT_PREFIX = b"#IDENT " @@ -172,14 +170,6 @@ def required_bool(field: str) -> bool: raise OgloProtocolError(f"OGLO USB identity has invalid {field}") return value - ident_schema = decoded.get("ident_schema", 0) - if ( - not isinstance(ident_schema, int) - or isinstance(ident_schema, bool) - or ident_schema not in (0, _USB_IDENT_SCHEMA_VERSION) - ): - raise OgloProtocolError("OGLO USB identity has unsupported ident_schema") - mcu_boot_id = required_string("mcu_boot_id", max_chars=32) if len(mcu_boot_id) != 32 or any( char not in "0123456789abcdef" for char in mcu_boot_id @@ -187,7 +177,6 @@ def required_bool(field: str) -> bool: raise OgloProtocolError("OGLO USB identity has invalid mcu_boot_id") identity: dict[str, Any] = { - "ident_schema": ident_schema, "mcu_boot_id": mcu_boot_id, "boot_count": required_int("boot_count"), "reset_reason": required_string("reset_reason", max_chars=64), @@ -201,13 +190,19 @@ def required_bool(field: str) -> bool: } application_sha256 = decoded.get("application_sha256") + application_sha256_status = decoded.get("application_sha256_status") + if application_sha256_status is not None and application_sha256_status not in ( + "available", + "unavailable", + ): + raise OgloProtocolError( + "OGLO USB identity has invalid application_sha256_status" + ) if application_sha256 is None: - if ident_schema == _USB_IDENT_SCHEMA_VERSION: - status = required_string("application_sha256_status", max_chars=32) - if status != "unavailable": - raise OgloProtocolError( - "OGLO USB identity has invalid application_sha256_status" - ) + if application_sha256_status == "available": + raise OgloProtocolError( + "OGLO USB identity has invalid application_sha256_status" + ) identity["application_sha256_status"] = "unavailable" else: if ( @@ -216,18 +211,14 @@ def required_bool(field: str) -> bool: or any(char not in "0123456789abcdef" for char in application_sha256) ): raise OgloProtocolError("OGLO USB identity has invalid application_sha256") - if ident_schema == _USB_IDENT_SCHEMA_VERSION: - status = required_string("application_sha256_status", max_chars=32) - if status != "available": - raise OgloProtocolError( - "OGLO USB identity has invalid application_sha256_status" - ) + if application_sha256_status == "unavailable": + raise OgloProtocolError( + "OGLO USB identity has invalid application_sha256_status" + ) identity["application_sha256_status"] = "available" identity["application_sha256"] = application_sha256 if "journal_ready" not in decoded: - if ident_schema == _USB_IDENT_SCHEMA_VERSION: - raise OgloProtocolError("OGLO USB identity has invalid journal_ready") identity["journal_status"] = "unavailable" return identity @@ -987,7 +978,6 @@ def _log_usb_evidence( if value is not None: self._usb_evidence_identity[key] = value event: dict[str, Any] = { - "schema_version": _USB_EVIDENCE_SCHEMA_VERSION, "event_type": event_type, "source_monotonic_ns": time.monotonic_ns(), "source_realtime_ns": time.time_ns(), diff --git a/tests/unit/adapters/oglo/ident_contract_v1.json b/tests/unit/adapters/oglo/ident_metadata.json similarity index 96% rename from tests/unit/adapters/oglo/ident_contract_v1.json rename to tests/unit/adapters/oglo/ident_metadata.json index 3a85cd4..458b982 100644 --- a/tests/unit/adapters/oglo/ident_contract_v1.json +++ b/tests/unit/adapters/oglo/ident_metadata.json @@ -1,5 +1,4 @@ { - "ident_schema": 1, "mcu_boot_id": "00112233445566778899aabbccddeeff", "boot_count": 7, "reset_reason": "poweron", diff --git a/tests/unit/adapters/test_oglo_usb_reconnect.py b/tests/unit/adapters/test_oglo_usb_reconnect.py index 3a1e828..b07cd6f 100644 --- a/tests/unit/adapters/test_oglo_usb_reconnect.py +++ b/tests/unit/adapters/test_oglo_usb_reconnect.py @@ -49,7 +49,6 @@ def _identity( reset_reason="poweron", ): return { - "ident_schema": 1, "mcu_boot_id": mcu_boot_id, "boot_count": journal_boot_counter, "reset_reason": reset_reason, @@ -475,7 +474,6 @@ def test_identity_parser_preserves_explicitly_unavailable_application_hash(oglo_ def test_identity_parser_accepts_the_hand_flashed_0913_legacy_shape(oglo_usb): module, _holder = oglo_usb identity = _identity() - identity.pop("ident_schema") identity.pop("application_sha256_status") identity.pop("application_sha256") identity.pop("journal_ready") @@ -485,20 +483,18 @@ def test_identity_parser_accepts_the_hand_flashed_0913_legacy_shape(oglo_usb): parsed = module.stream._parse_usb_identity(json.dumps(identity).encode()) - assert parsed["ident_schema"] == 0 assert parsed["application_sha256_status"] == "unavailable" assert parsed["journal_status"] == "unavailable" -def test_identity_parser_matches_the_shared_v1_golden_vector(oglo_usb): +def test_identity_parser_matches_the_shared_additive_metadata_vector(oglo_usb): module, _holder = oglo_usb fixture = ( - Path(__file__).with_name("oglo") / "ident_contract_v1.json" + Path(__file__).with_name("oglo") / "ident_metadata.json" ).read_bytes() parsed = module.stream._parse_usb_identity(fixture) - assert parsed["ident_schema"] == 1 assert parsed["mcu_boot_id"] == "00112233445566778899aabbccddeeff" assert parsed["application_sha256_status"] == "available" assert parsed["journal_status"] == "available" @@ -692,7 +688,6 @@ def test_usb_flight_log_is_bounded_metadata_only(oglo_usb, tmp_path, monkeypatch records = _usb_evidence_records(messages, module) assert records[-1]["event_type"] == "outage_observed" assert records[-1]["outage_id"] == "outage-test" - assert records[-1]["schema_version"] == "oglo.usb_evidence.v1" assert records[-1]["source_monotonic_ns"] > 0 assert records[-1]["source_realtime_ns"] > 0 assert "usb_serial" in records[-1] From 535fb85c5f19fc7f6dc51e01f07ef38e2deb6549 Mon Sep 17 00:00:00 2001 From: Sungman Cho Date: Sat, 15 Aug 2026 22:55:58 +0900 Subject: [PATCH 6/6] fix(oglo): simplify USB outage evidence --- src/syncfield/adapters/oglo/stream.py | 179 ++++++------------ tests/unit/adapters/oglo/ident_metadata.json | 3 +- .../unit/adapters/test_oglo_usb_reconnect.py | 132 ++++--------- 3 files changed, 92 insertions(+), 222 deletions(-) diff --git a/src/syncfield/adapters/oglo/stream.py b/src/syncfield/adapters/oglo/stream.py index 6ecb312..fbdf928 100644 --- a/src/syncfield/adapters/oglo/stream.py +++ b/src/syncfield/adapters/oglo/stream.py @@ -22,7 +22,6 @@ import asyncio import json import logging -import math import os import threading import time @@ -114,22 +113,10 @@ _RECONNECT_USB_RESET_AFTER_S = 3.0 _USBDEVFS_RESET = (ord("U") << 8) | 20 _USB_EVIDENCE_PREFIX = "OGLO_USB_EVIDENCE " -_USB_INCIDENT_HISTORY_LIMIT = 16 _USB_IDENT_COMMAND = b"GET IDENT\n" _USB_IDENT_PREFIX = b"#IDENT " _USB_IDENT_PROBE_TIMEOUT_S = 0.75 _USB_IDENT_MAX_LINE_BYTES = 4096 -# The on-board ICM-42688P/TDK accelerometer is configured for +/-8 g in the -# matching firmware. Keep a short, bounded host-side window so a complete MCU -# power loss cannot erase the motion immediately preceding a USB outage. This -# is correlation evidence, not a cause classifier: gravity contributes roughly -# 1,000 mg and a cable tug, impact, or ordinary hand motion can look alike. -_PRE_OUTAGE_IMU_WINDOW_NS = 5_000_000_000 -_PRE_OUTAGE_IMU_MAX_SAMPLES = 3_000 -_IMU_ACCEL_SCALE_MG_PER_LSB = 0.244 -_IMU_ACCEL_FULL_SCALE_G = 8 - - @dataclass(frozen=True) class _UsbIdentityProbe: identity: dict[str, Any] | None @@ -227,12 +214,6 @@ def required_bool(field: str) -> bool: "available" if identity["journal_ready"] else "invalid" ) if identity["journal_ready"]: - journal_boot_id = required_string("journal_boot_id", max_chars=16) - if len(journal_boot_id) != 16 or any( - char not in "0123456789abcdef" for char in journal_boot_id - ): - raise OgloProtocolError("OGLO USB identity has invalid journal_boot_id") - identity["journal_boot_id"] = journal_boot_id identity["journal_boot_counter"] = required_int("journal_boot_counter") else: identity["journal_error"] = required_string("journal_error", max_chars=128) @@ -468,24 +449,19 @@ def __init__( self._substream_callbacks: list[Callable[[Any], None]] = [] self._imu_recent_ns: deque[int] = deque(maxlen=120) self._mag_recent_ns: deque[int] = deque(maxlen=120) - self._pre_outage_accel: deque[tuple[int, int, bool]] = deque( - maxlen=_PRE_OUTAGE_IMU_MAX_SAMPLES - ) self._imu_recent_lock = threading.Lock() # Drop detection across packets (seq_base continuity). self._next_expected_seq: dict[str, int] = {} - # Metadata-only flight recorder. It is emitted to the normal process - # log when an outage occurs; sensor and raw USB bytes never enter it. - self._usb_read_history: deque[tuple[int, int, str, int | None]] = deque( - maxlen=_USB_INCIDENT_HISTORY_LIMIT - ) + # Metadata-only read summary. Repeated zero-byte reads are collapsed so + # a long silence cannot evict the last valid frame from the evidence. + self._last_valid_usb_frame_at_ns: int | None = None + self._last_nonzero_usb_read: tuple[int, int, str] | None = None + self._consecutive_zero_usb_reads = 0 + self._last_usb_read_exception_errno: int | None = None self._usb_evidence_identity = _usb_physical_identity(self._serial_port) - # The ID is allocated with the identity baseline, before an outage - # exists. If the link later fails, every host action and the after probe - # reuse it. A successful after probe is also the next outage's baseline. - self._pending_outage_id: str | None = None + self._identity_baseline: _UsbIdentityProbe | None = None # ------------------------------------------------------------------ # Stream SPI — 4-phase lifecycle @@ -511,9 +487,8 @@ def connect(self) -> None: self._next_expected_seq = {} self._connect_error = None self._manifest = None - self._pending_outage_id = None - with self._imu_recent_lock: - self._pre_outage_accel.clear() + self._identity_baseline = None + self._reset_usb_read_diagnostics() self._ready_event.clear() self._stop_event.clear() @@ -1063,14 +1038,6 @@ def _log_identity_probe( **probe.identity, ) - def _start_identity_epoch(self, probe: _UsbIdentityProbe) -> None: - self._pending_outage_id = str(uuid4()) - self._log_identity_probe( - phase="before", - outage_id=self._pending_outage_id, - probe=probe, - ) - def _record_usb_read( self, *, @@ -1083,9 +1050,21 @@ def _record_usb_read( if not isinstance(errno, int) or isinstance(errno, bool): errno = None duration_us = max(0, (time.monotonic_ns() - started_ns) // 1_000) - self._usb_read_history.append( - (duration_us, byte_count, diagnostic, errno) - ) + if diagnostic == "valid_frame": + self._last_valid_usb_frame_at_ns = time.monotonic_ns() + self._last_usb_read_exception_errno = None + if diagnostic == "zero_byte_read": + self._consecutive_zero_usb_reads += 1 + else: + self._consecutive_zero_usb_reads = 0 + if byte_count > 0: + self._last_nonzero_usb_read = ( + duration_us, + byte_count, + diagnostic, + ) + if diagnostic == "read_exception" and errno is not None: + self._last_usb_read_exception_errno = errno @staticmethod def _classify_usb_read( @@ -1102,66 +1081,33 @@ def _classify_usb_read( return "malformed_header_or_frame" return "unframed_traffic" - def _usb_read_history_projection(self) -> list[dict[str, Any]]: - projection: list[dict[str, Any]] = [] - for duration_us, byte_count, classification, errno in self._usb_read_history: - item: dict[str, Any] = { + def _usb_read_summary(self, outage_at_ns: int) -> dict[str, Any]: + summary: dict[str, Any] = { + "consecutive_zero_reads": self._consecutive_zero_usb_reads, + } + if self._last_valid_usb_frame_at_ns is not None: + summary["last_valid_frame_age_ms"] = max( + 0, + (outage_at_ns - self._last_valid_usb_frame_at_ns) // 1_000_000, + ) + if self._last_nonzero_usb_read is not None: + duration_us, byte_count, classification = self._last_nonzero_usb_read + summary["last_nonzero_read"] = { "duration_us": duration_us, "byte_count": byte_count, "classification": classification, } - if errno is not None: - item["errno"] = errno - projection.append(item) - return projection - - def _take_pre_outage_imu_projection(self, outage_at_ns: int) -> dict[str, Any]: - """Drain and summarize recent acceleration without retaining samples. - - The evidence stays metadata-only: only a count, peak magnitude, age, - and saturation bit are logged. Raw IMU samples remain in the normal - sensor artifact when recording is active. - """ - - cutoff_ns = outage_at_ns - _PRE_OUTAGE_IMU_WINDOW_NS - with self._imu_recent_lock: - recent = [ - sample - for sample in self._pre_outage_accel - if sample[0] >= cutoff_ns - ] - self._pre_outage_accel.clear() - - window_ms = _PRE_OUTAGE_IMU_WINDOW_NS // 1_000_000 - if not recent: - return { - "status": "unavailable", - "window_ms": window_ms, - "reason": "no_recent_imu_sample", - } + if self._last_usb_read_exception_errno is not None: + summary["last_read_exception_errno"] = ( + self._last_usb_read_exception_errno + ) + return summary - peak_at_ns, peak_norm_sq, peak_saturated = max( - recent, key=lambda sample: sample[1] - ) - latest_at_ns = recent[-1][0] - return { - "status": "observed", - "window_ms": window_ms, - "sample_count": len(recent), - "latest_sample_age_ms": max( - 0, (outage_at_ns - latest_at_ns) // 1_000_000 - ), - "peak_sample_age_ms": max( - 0, (outage_at_ns - peak_at_ns) // 1_000_000 - ), - "peak_resultant_mg": round( - math.sqrt(peak_norm_sq) * _IMU_ACCEL_SCALE_MG_PER_LSB - ), - "peak_axis_saturated": peak_saturated, - "accelerometer_full_scale_g": _IMU_ACCEL_FULL_SCALE_G, - "scale_mg_per_lsb": _IMU_ACCEL_SCALE_MG_PER_LSB, - "classification": "evidence_only_not_cause", - } + def _reset_usb_read_diagnostics(self) -> None: + self._last_valid_usb_frame_at_ns = None + self._last_nonzero_usb_read = None + self._consecutive_zero_usb_reads = 0 + self._last_usb_read_exception_errno = None # ------------------------------------------------------------------ # Payload decoding (unit-testable without asyncio / bleak) @@ -1230,8 +1176,8 @@ def _run_usb_reader(self) -> None: time.sleep(0.2) manifest = self._read_usb_manifest(ser) self._apply_manifest(manifest) - self._start_identity_epoch( - self._read_usb_identity(ser, quiesce_stream=False) + self._identity_baseline = self._read_usb_identity( + ser, quiesce_stream=False ) ser.reset_input_buffer() ser.write(STREAM_ON_COMMAND) @@ -1261,22 +1207,21 @@ def recover_or_die(reason: str) -> None: """ nonlocal ser, buffer, last_packet_at outage_at_ns = time.monotonic_ns() - outage_id = self._pending_outage_id or str(uuid4()) - if self._pending_outage_id is None: - self._pending_outage_id = outage_id - self._log_identity_probe( - phase="before", - outage_id=outage_id, - probe=_UsbIdentityProbe(None, "baseline_unavailable"), - ) + outage_id = str(uuid4()) + self._log_identity_probe( + phase="before", + outage_id=outage_id, + probe=self._identity_baseline + or _UsbIdentityProbe(None, "baseline_unavailable"), + ) self._log_usb_evidence( "outage_observed", level=logging.WARNING, outage_id=outage_id, reason=reason, - read_spans=self._usb_read_history_projection(), - pre_outage_imu=self._take_pre_outage_imu_projection(outage_at_ns), + read_summary=self._usb_read_summary(outage_at_ns), ) + self._reset_usb_read_diagnostics() reconnected = self._usb_reconnect( reason=reason, outage_id=outage_id, @@ -1630,10 +1575,7 @@ def reconnect_summary() -> dict[str, Any]: reconnect_summary=reconnect_summary(), **loss_fields, ) - # The recovered board is now the authoritative baseline for the - # next incident, even when its identity probe failed. Preallocating - # the next ID lets that failure be explicit before any outage. - self._start_identity_epoch(adoption.identity_probe) + self._identity_baseline = adoption.identity_probe return ser, adoption.seed self._log_identity_probe( phase="after", @@ -1641,7 +1583,6 @@ def reconnect_summary() -> dict[str, Any]: probe=last_identity_probe, connection_adopted=False, ) - self._pending_outage_id = None self._log_usb_evidence( "recovery_result", level=logging.ERROR, @@ -1755,12 +1696,8 @@ def _handle_imu( device_ns: int, ) -> None: """Fan a wrist-IMU sample out to the live handler + substream file.""" - ax, ay, az = imu[:3] - accel_norm_sq = ax * ax + ay * ay + az * az - accel_saturated = any(abs(axis) >= 32_767 for axis in (ax, ay, az)) with self._imu_recent_lock: self._imu_recent_ns.append(recv_ns) - self._pre_outage_accel.append((recv_ns, accel_norm_sq, accel_saturated)) channels = {name: int(v) for name, v in zip(IMU_CHANNELS, imu)} with self._recording_lock: diff --git a/tests/unit/adapters/oglo/ident_metadata.json b/tests/unit/adapters/oglo/ident_metadata.json index 458b982..0e02904 100644 --- a/tests/unit/adapters/oglo/ident_metadata.json +++ b/tests/unit/adapters/oglo/ident_metadata.json @@ -12,6 +12,5 @@ "wedge_last_stall_ms": 0, "wedge_guard": false, "journal_ready": true, - "journal_boot_counter": 7, - "journal_boot_id": "0123456789abcdef" + "journal_boot_counter": 7 } diff --git a/tests/unit/adapters/test_oglo_usb_reconnect.py b/tests/unit/adapters/test_oglo_usb_reconnect.py index b07cd6f..41b31e2 100644 --- a/tests/unit/adapters/test_oglo_usb_reconnect.py +++ b/tests/unit/adapters/test_oglo_usb_reconnect.py @@ -44,7 +44,6 @@ class _SerialError(Exception): def _identity( *, mcu_boot_id="00112233445566778899aabbccddeeff", - journal_boot_id="0123456789abcdef", journal_boot_counter=7, reset_reason="poweron", ): @@ -63,7 +62,6 @@ def _identity( "wedge_guard": False, "journal_ready": True, "journal_boot_counter": journal_boot_counter, - "journal_boot_id": journal_boot_id, } @@ -255,8 +253,15 @@ def test_still_streaming_glove_records_same_identity_across_outage( ) assert before["stream_id"] == after["stream_id"] == "tactile_left" assert before["mcu_boot_id"] == after["mcu_boot_id"] - assert before["journal_boot_id"] == after["journal_boot_id"] + assert before["journal_boot_counter"] == after["journal_boot_counter"] assert after["connection_adopted"] is True + incident_records = [event for event in records if "outage_id" in event] + assert {event["outage_id"] for event in incident_records} == { + outage["outage_id"] + } + assert sum( + event["event_type"] == "identity_before" for event in incident_records + ) == 1 kinds = [h.kind for h in health] assert HealthEventKind.WARNING in kinds assert any("reconnect" in h.detail.lower() for h in health) @@ -266,75 +271,6 @@ def test_still_streaming_glove_records_same_identity_across_outage( assert report.frame_count == 2 -def test_outage_records_recent_imu_peak_as_evidence_not_cause( - oglo_usb, tmp_path, monkeypatch -): - module, holder = oglo_usb - first, second = _FakeSerial("p"), _FakeSerial("p") - second.feed(tag(TAG_TYPE_TACTILE, 500, 60_000)) - holder["queue"] = [first, second] - messages = [] - monkeypatch.setattr( - module.stream.logger, - "log", - lambda _level, template, *args: messages.append(template % args), - ) - - stream, _health = _connected_recording_stream(module, holder, tmp_path, first) - first.feed( - tag(TAG_TYPE_IMU, 0, 2_500, (0, 0, 4096, 0, 0, 0)) - + tag(TAG_TYPE_IMU, 1, 3_000, (12_000, 0, 4096, 0, 0, 0)) - ) - _wait(lambda: len(stream._pre_outage_accel) == 2) - first.kill() - _wait(lambda: stream._frame_count >= 2) - - records = _usb_evidence_records(messages, module) - outage = next(event for event in records if event["event_type"] == "outage_observed") - imu = outage["pre_outage_imu"] - assert imu["status"] == "observed" - assert imu["window_ms"] == 5_000 - assert imu["sample_count"] == 2 - assert 3_000 <= imu["peak_resultant_mg"] <= 3_200 - assert imu["peak_axis_saturated"] is False - assert imu["accelerometer_full_scale_g"] == 8 - assert imu["classification"] == "evidence_only_not_cause" - assert not stream._pre_outage_accel - - stream.stop_recording() - stream.disconnect() - - -def test_outage_marks_pre_outage_imu_unavailable_when_no_sample_arrived( - oglo_usb, tmp_path, monkeypatch -): - module, holder = oglo_usb - first, second = _FakeSerial("p"), _FakeSerial("p") - second.feed(tag(TAG_TYPE_TACTILE, 500, 60_000)) - holder["queue"] = [first, second] - messages = [] - monkeypatch.setattr( - module.stream.logger, - "log", - lambda _level, template, *args: messages.append(template % args), - ) - - stream, _health = _connected_recording_stream(module, holder, tmp_path, first) - first.kill() - _wait(lambda: stream._frame_count >= 2) - - records = _usb_evidence_records(messages, module) - outage = next(event for event in records if event["event_type"] == "outage_observed") - assert outage["pre_outage_imu"] == { - "reason": "no_recent_imu_sample", - "status": "unavailable", - "window_ms": 5_000, - } - - stream.stop_recording() - stream.disconnect() - - def test_rebooted_glove_gets_a_stream_on_nudge(oglo_usb, tmp_path, monkeypatch): """MCU rebooted during the outage: silent until STREAM TAG ON.""" module, holder = oglo_usb @@ -344,7 +280,6 @@ def test_rebooted_glove_gets_a_stream_on_nudge(oglo_usb, tmp_path, monkeypatch): stream_on_gated=True, identity=_identity( mcu_boot_id="ffeeddccbbaa99887766554433221100", - journal_boot_id="fedcba9876543210", journal_boot_counter=8, reset_reason="software", ), @@ -372,7 +307,6 @@ def test_rebooted_glove_gets_a_stream_on_nudge(oglo_usb, tmp_path, monkeypatch): and event["outage_id"] == outage["outage_id"] ) assert after["mcu_boot_id"] == "ffeeddccbbaa99887766554433221100" - assert after["journal_boot_id"] == "fedcba9876543210" assert after["journal_boot_counter"] == 8 assert after["reset_reason"] == "software" assert after["application_sha256"] == "ab" * 32 @@ -380,7 +314,7 @@ def test_rebooted_glove_gets_a_stream_on_nudge(oglo_usb, tmp_path, monkeypatch): stream.disconnect() -def test_initial_identity_probe_failure_is_explicit_and_non_gating( +def test_initial_identity_probe_failure_is_non_gating_and_deferred_until_outage( oglo_usb, tmp_path, monkeypatch ): module, holder = oglo_usb @@ -402,12 +336,7 @@ def test_initial_identity_probe_failure_is_explicit_and_non_gating( ) stream.connect() records = _usb_evidence_records(messages, module) - failure = next( - event for event in records if event["event_type"] == "identity_probe_failed" - ) - assert failure["phase"] == "before" - assert failure["failure_class"] == "unsupported" - assert failure["outage_id"] + assert records == [] assert stream._thread.is_alive() stream.disconnect() @@ -478,7 +407,6 @@ def test_identity_parser_accepts_the_hand_flashed_0913_legacy_shape(oglo_usb): identity.pop("application_sha256") identity.pop("journal_ready") identity.pop("journal_boot_counter") - identity.pop("journal_boot_id") identity["fw_rev"] = "0.9.13" parsed = module.stream._parse_usb_identity(json.dumps(identity).encode()) @@ -650,7 +578,9 @@ def test_present_but_mute_device_escalates_to_one_usb_reset( stream.disconnect() -def test_usb_flight_log_is_bounded_metadata_only(oglo_usb, tmp_path, monkeypatch): +def test_usb_read_summary_preserves_last_valid_frame_through_long_silence( + oglo_usb, tmp_path, monkeypatch +): module, _holder = oglo_usb stream = module.OgloTactileStream( "tactile_left", @@ -659,19 +589,24 @@ def test_usb_flight_log_is_bounded_metadata_only(oglo_usb, tmp_path, monkeypatch output_dir=tmp_path, ) - for index in range(70): + valid_started_ns = time.monotonic_ns() + stream._record_usb_read( + started_ns=valid_started_ns, + byte_count=192, + diagnostic="valid_frame", + ) + for _ in range(70): stream._record_usb_read( - started_ns=index, - byte_count=index, - diagnostic="valid_frame", + started_ns=time.monotonic_ns(), + byte_count=0, + diagnostic="zero_byte_read", ) - assert len(stream._usb_read_history) == 16 - assert all(isinstance(entry, tuple) for entry in stream._usb_read_history) - projection = stream._usb_read_history_projection() - assert len(projection) == 16 - assert projection[0]["byte_count"] == 54 - assert set(projection[0]) == {"duration_us", "byte_count", "classification"} + projection = stream._usb_read_summary(time.monotonic_ns()) + assert projection["consecutive_zero_reads"] == 70 + assert projection["last_valid_frame_age_ms"] >= 0 + assert projection["last_nonzero_read"]["byte_count"] == 192 + assert projection["last_nonzero_read"]["classification"] == "valid_frame" messages = [] monkeypatch.setattr( @@ -682,7 +617,7 @@ def test_usb_flight_log_is_bounded_metadata_only(oglo_usb, tmp_path, monkeypatch stream._log_usb_evidence( "outage_observed", outage_id="outage-test", - read_spans=projection, + read_summary=projection, ) records = _usb_evidence_records(messages, module) @@ -704,7 +639,7 @@ def test_usb_flight_log_is_bounded_metadata_only(oglo_usb, tmp_path, monkeypatch assert redundant not in records[-1] -def test_usb_read_history_keeps_only_real_errno(oglo_usb, tmp_path): +def test_usb_read_summary_keeps_only_real_errno(oglo_usb, tmp_path): module, _holder = oglo_usb stream = module.OgloTactileStream( "tactile_left", @@ -724,7 +659,6 @@ def test_usb_read_history_keeps_only_real_errno(oglo_usb, tmp_path): diagnostic="zero_byte_read", ) - projection = stream._usb_read_history_projection() - assert projection[0]["errno"] == 5 - assert "errno" not in projection[1] - assert all("exception_message" not in item for item in projection) + projection = stream._usb_read_summary(time.monotonic_ns()) + assert projection["last_read_exception_errno"] == 5 + assert "exception_message" not in projection