diff --git a/scripts/observer/observer.py b/scripts/observer/observer.py
index 3b1c680..dd6ee84 100644
--- a/scripts/observer/observer.py
+++ b/scripts/observer/observer.py
@@ -44,6 +44,9 @@ WORLD_DIR = Path(RUNTIME_PATH) / "world"
 OBSERVER_STATE_FILE = STATE_DIR / "observer_checkpoint.json"
 FAILED_EVENTS_DIR = STATE_DIR / "observer_failed_events"
 
+# node_health events arrive every 60 s; 5 missed beats → declare offline
+NODE_OFFLINE_TTL_SECS = int(os.getenv("NODE_OFFLINE_TTL_SECS", "300"))
+
 REPO_ROOT = Path(__file__).parent.parent.parent
 INVENTORY_TOPOLOGY = REPO_ROOT / "inventory" / "topology.yaml"
 
@@ -224,6 +227,22 @@ class Observer:
 
         now = time.time()
 
+        # Mark nodes offline when last_seen exceeds NODE_OFFLINE_TTL_SECS.
+        # Handles crash/network-partition cases where node_offline is never emitted.
+        for node_name, node_info in self.world_state["nodes"].items():
+            if node_info.get("status") != "online":
+                continue
+            last_seen = _parse_ts(node_info.get("last_seen"))
+            if last_seen == 0.0:
+                continue  # Never reported — keep status unknown, don't flip
+            age = now - last_seen
+            if age > NODE_OFFLINE_TTL_SECS:
+                logger.warning(
+                    "Node %s last seen %ds ago (> TTL %ds) — marking offline",
+                    node_name, int(age), NODE_OFFLINE_TTL_SECS,
+                )
+                node_info["status"] = "offline"
+
         try:
             # Collect incident_ids currently referenced by any service entry.
             linked_ids: set = {
diff --git a/services/control-plane/tests/test_incident_lifecycle.py b/services/control-plane/tests/test_incident_lifecycle.py
index 9c0b7b2..4960519 100644
--- a/services/control-plane/tests/test_incident_lifecycle.py
+++ b/services/control-plane/tests/test_incident_lifecycle.py
@@ -377,3 +377,93 @@ def test_run_once_quarantines_bad_event_and_processes_next_for_same_node(tmp_pat
     assert not bad_event.exists()
     assert obs.world_state["nodes"]["lustro"]["status"] == "online"
     assert obs.node_checkpoints["lustro"] == str(good_event)
+
+
+# ---------------------------------------------------------------------------
+# 7. Node offline TTL — liveness detection via last_seen age
+# ---------------------------------------------------------------------------
+
+def test_prune_marks_online_node_offline_after_ttl(tmp_path):
+    """Node online with last_seen > NODE_OFFLINE_TTL_SECS must be flipped to offline."""
+    obs = _make_observer_simple(tmp_path)
+    import observer.observer as obs_mod
+
+    # Use "vps" — present in the test topology so it won't be pruned as stale.
+    obs.world_state["nodes"]["vps"] = {
+        "status": "online",
+        "last_seen": time.time() - (obs_mod.NODE_OFFLINE_TTL_SECS + 60),  # over TTL
+        "roles": [],
+    }
+
+    obs._prune_stale_world()
+
+    assert obs.world_state["nodes"]["vps"]["status"] == "offline"
+
+
+def test_prune_does_not_mark_node_offline_within_ttl(tmp_path):
+    """Node online with recent last_seen must stay online."""
+    obs = _make_observer_simple(tmp_path)
+    import observer.observer as obs_mod
+
+    obs.world_state["nodes"]["vps"] = {
+        "status": "online",
+        "last_seen": time.time() - (obs_mod.NODE_OFFLINE_TTL_SECS - 60),  # within TTL
+        "roles": [],
+    }
+
+    obs._prune_stale_world()
+
+    assert obs.world_state["nodes"]["vps"]["status"] == "online"
+
+
+def test_prune_does_not_flip_already_offline_node(tmp_path):
+    """Node already offline must not be changed by TTL logic."""
+    obs = _make_observer_simple(tmp_path)
+
+    obs.world_state["nodes"]["vps"] = {
+        "status": "offline",
+        "last_seen": time.time() - 9999,
+        "roles": [],
+    }
+
+    obs._prune_stale_world()
+
+    assert obs.world_state["nodes"]["vps"]["status"] == "offline"
+
+
+def test_prune_does_not_flip_never_seen_node(tmp_path):
+    """Node with last_seen=None must stay in its original status."""
+    obs = _make_observer_simple(tmp_path)
+
+    obs.world_state["nodes"]["vps"] = {
+        "status": "online",
+        "last_seen": None,
+        "roles": [],
+    }
+
+    obs._prune_stale_world()
+
+    # last_seen=None means we have no data — don't flip to offline
+    assert obs.world_state["nodes"]["vps"]["status"] == "online"
+
+
+def test_prune_marks_node_offline_with_iso_last_seen(tmp_path):
+    """TTL check must work when last_seen is an ISO-8601 string."""
+    obs = _make_observer_simple(tmp_path)
+    import observer.observer as obs_mod
+
+    from datetime import datetime, timezone
+    stale_ts = datetime.fromtimestamp(
+        time.time() - (obs_mod.NODE_OFFLINE_TTL_SECS + 120),
+        tz=timezone.utc,
+    ).isoformat()
+
+    obs.world_state["nodes"]["vps"] = {
+        "status": "online",
+        "last_seen": stale_ts,
+        "roles": [],
+    }
+
+    obs._prune_stale_world()
+
+    assert obs.world_state["nodes"]["vps"]["status"] == "offline"