diff --git a/tests/plugins/host_agent/test_scheduler.py b/tests/plugins/host_agent/test_scheduler.py
new file mode 100644
index 0000000..49be93a
--- /dev/null
+++ b/tests/plugins/host_agent/test_scheduler.py
@@ -0,0 +1,48 @@
+"""Pure-function tests for host_agent staleness filter."""
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+from plugins.host_agent.scheduler import _filter_stale
+
+
+@dataclass
+class FakeReg:
+    host_id: str
+    last_seen_at: datetime | None
+
+
+NOW = datetime(2026, 4, 15, 12, 0, 0, tzinfo=timezone.utc)
+
+
+def test_filter_stale_returns_only_old_rows():
+    regs = [
+        FakeReg("fresh", NOW - timedelta(seconds=30)),
+        FakeReg("stale", NOW - timedelta(seconds=600)),
+        FakeReg("never", None),
+    ]
+    stale = _filter_stale(regs, now=NOW, stale_after_seconds=180)
+    ids = {r.host_id for r in stale}
+    assert ids == {"stale"}
+
+
+def test_filter_stale_empty_input():
+    assert _filter_stale([], now=NOW, stale_after_seconds=180) == []
+
+
+def test_filter_stale_all_fresh():
+    regs = [
+        FakeReg("a", NOW - timedelta(seconds=10)),
+        FakeReg("b", NOW - timedelta(seconds=30)),
+    ]
+    assert _filter_stale(regs, now=NOW, stale_after_seconds=60) == []
+
+
+def test_filter_stale_boundary_exact_cutoff_is_not_stale():
+    # A row exactly at the cutoff should not be flagged (strict less-than).
+    regs = [FakeReg("edge", NOW - timedelta(seconds=180))]
+    assert _filter_stale(regs, now=NOW, stale_after_seconds=180) == []
+
+
+def test_filter_stale_skips_none_last_seen():
+    regs = [FakeReg("never", None)]
+    assert _filter_stale(regs, now=NOW, stale_after_seconds=1) == []