fix(sensor): age-skip old legacy DSH sessions instead of reporting drift

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

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)