From a995382f9b0879034044bc830ed76e2a96035f3a Mon Sep 17 00:00:00 2001 From: Rudy Celekli Date: Mon, 19 Oct 2026 15:32:12 -0400 Subject: [PATCH] fix(sensor): age-skip old legacy DSH sessions instead of reporting drift DshParser checked the session generation before the lookback window, so every legacy (pre-v3) session directory was counted as unsupported_schema on every run, however old it was. That marked DSH health "partial" with suspected_schema_drift for anyone with finished legacy sessions. Legacy generations older than the lookback window are now counted as file_age_skipped; recent ones are still reported as unsupported. Co-Authored-By: Claude Opus 5.5 --- Sensor/adr_sensor/parsers/dsh_parser.py | 15 +++++++++++---- Sensor/tests/test_dsh_parser.py | 19 +++++++++++++++++++ 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/Sensor/adr_sensor/parsers/dsh_parser.py b/Sensor/adr_sensor/parsers/dsh_parser.py index fb857544..54521aae 100644 --- a/Sensor/adr_sensor/parsers/dsh_parser.py +++ b/Sensor/adr_sensor/parsers/dsh_parser.py @@ -45,7 +45,7 @@ def parse_all(self) -> List[AgentEvent]: if self.max_age_days > 0: cutoff = datetime.now(timezone.utc) - timedelta(days=self.max_age_days) - files = self._select_session_generations() + files = self._select_session_generations(cutoff) entries: Dict[str, AgentEvent] = {} for path in files: failure_code = "file_stat_error" @@ -72,8 +72,12 @@ def _revision(entry: AgentEvent) -> tuple: updated_at = DshParser._timestamp(context.get("last_event_at")) or entry.timestamp return updated_at, context.get("event_count", 0) - def _select_session_generations(self) -> List[Path]: - """Select the highest generation per directory when it is current v3.""" + def _select_session_generations(self, cutoff: Optional[datetime] = None) -> List[Path]: + """Select the highest generation per directory when it is current v3. + + Sessions outside the lookback window are age-skipped, not reported as + unsupported, so long-finished legacy sessions do not signal drift. + """ selected: Dict[Path, tuple[int, float, Path]] = {} candidates = [*self.base_path.glob("**/session.jsonl"), *self.base_path.glob("**/session.jsonl.zstd")] candidates.extend(self.base_path.glob("**/session.v*.jsonl")) @@ -94,8 +98,11 @@ def _select_session_generations(self) -> List[Path]: if current is None or candidate[:2] > current[:2]: selected[path.parent] = candidate current_paths = [] - for version, _, path in sorted(selected.values(), key=lambda item: item[1], reverse=True): + for version, mtime, path in sorted(selected.values(), key=lambda item: item[1], reverse=True): if version != MAX_SUPPORTED_SESSION_VERSION: + if cutoff is not None and mtime < cutoff.timestamp(): + self.record_diagnostic("file_age_skipped") + continue self.record_diagnostic("unsupported_schema") logger.warning("[DSH] Unsupported session generation v%s in %s", version, path.parent) continue diff --git a/Sensor/tests/test_dsh_parser.py b/Sensor/tests/test_dsh_parser.py index dddc76e1..7a980d86 100644 --- a/Sensor/tests/test_dsh_parser.py +++ b/Sensor/tests/test_dsh_parser.py @@ -414,6 +414,25 @@ def test_future_or_legacy_generation_is_not_parsed_as_v3(tmp_path): assert DshParser().parse_session_file(legacy_path) is None +def test_old_legacy_generation_is_age_skipped_not_reported_as_drift(tmp_path): + events = [ + {"type": "session", "version": 2, "id": "legacy", "createdAt": 1750000000000}, + {"type": "user/message", "data": {"role": "user", "content": [{"type": "text", "text": "hello world"}]}}, + ] + old = write_session(tmp_path, events, "session.v2.jsonl", session_dir_name="old-legacy") + os.utime(old, (1, 1)) + + parser = DshParser(base_path=tmp_path / "sessions") + assert parser.parse_all() == [] + assert parser.get_diagnostics() == {"file_age_skipped": 1} + + recent = write_session(tmp_path, events, "session.v2.jsonl", session_dir_name="recent-legacy") + parser = DshParser(base_path=tmp_path / "sessions") + assert parser.parse_all() == [] + assert parser.get_diagnostics() == {"file_age_skipped": 1, "unsupported_schema": 1} + assert recent.exists() + + def test_default_home_follows_dsh_home_or_user_home(monkeypatch, tmp_path): monkeypatch.delenv("DSH_HOME", raising=False) monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path)