fix(mail): soften transient liveness bounces

src/lingtai/adapters/posix/mail.py

@@ -34,6 +34,13 @@
 
 logger = logging.getLogger(__name__)
 
+# A single strict 2s heartbeat read is too brittle for cross-network email on
+# external volumes or during agent wake/refresh.  Delivery still requires the
+# same Core presence predicate, but the mailman thread gives transient heartbeat
+# lag a short grace window before refusing and emitting a bounce.
+MAIL_DELIVERY_LIVENESS_GRACE_SECONDS: float = 1.5
+MAIL_DELIVERY_LIVENESS_RETRY_INTERVAL_SECONDS: float = 0.2
+
 
 def _presence_store_for(recipient_dir: Path):
     """Build a target-bound POSIX presence adapter for a recipient directory."""
@@ -42,6 +49,28 @@ def _presence_store_for(recipient_dir: Path):
     return PosixAgentPresenceStoreAdapter(recipient_dir)
 
 
+
+def _recipient_alive_with_grace(presence_store) -> bool:
+    """Return True once the recipient is observed alive within the mail grace.
+
+    CPR and other lifecycle actions still use the Core strict heartbeat
+    predicate directly.  Email delivery is different because the mailman thread
+    can afford a brief wait before producing a user-visible bounce; this avoids
+    false "not running" failures from heartbeat write/read races while
+    preserving the same liveness source of truth.
+    """
+
+    deadline = time.monotonic() + max(0.0, MAIL_DELIVERY_LIVENESS_GRACE_SECONDS)
+    interval = max(0.0, MAIL_DELIVERY_LIVENESS_RETRY_INTERVAL_SECONDS)
+    while True:
+        if _presence_observe_alive(presence_store, wall_now=time.time()):
+            return True
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            return False
+        time.sleep(min(interval, remaining))
+
+
 class PosixFilesystemMailAdapter(MailTransportPort):
     """Filesystem-based mail delivery.
 
@@ -126,11 +155,12 @@ def send(
         if not _presence_is_agent(recipient_presence.observe_manifest()):
             return f"No agent at {address}"
 
-        if not _presence_observe_alive(
-            recipient_presence,
-            wall_now=time.time(),
-        ):
-            return f"Agent at {address} is not running"
+        if not _recipient_alive_with_grace(recipient_presence):
+            return (
+                f"Agent at {address} is not currently deliverable "
+                "(liveness uncertain: heartbeat was not fresh during the "
+                f"{MAIL_DELIVERY_LIVENESS_GRACE_SECONDS:g}s mail-delivery grace window)"
+            )
 
         # --- create inbox entry ---------------------------------------
         msg_id = _new_mailbox_id()

tests/test_filesystem_mail.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import json
+import threading
 import time
 
 import pytest
@@ -90,9 +91,12 @@ def test_send_fails_no_agent_json(self, tmp_path):
         assert result is not None  # error string
         assert "no agent" in result.lower()
 
-    def test_send_fails_stale_heartbeat(self, tmp_path):
+    def test_send_fails_stale_heartbeat(self, tmp_path, monkeypatch):
+        import lingtai.adapters.posix.mail as mail_mod
         from lingtai.adapters.posix.mail import PosixFilesystemMailAdapter
 
+        monkeypatch.setattr(mail_mod, "MAIL_DELIVERY_LIVENESS_GRACE_SECONDS", 0.0)
+
         sender_dir = _make_agent_dir(tmp_path, "sender01")
         recip_dir = _make_agent_dir(tmp_path, "recip01")
         (recip_dir / "mailbox" / "inbox").mkdir(parents=True)
@@ -102,7 +106,41 @@ def test_send_fails_stale_heartbeat(self, tmp_path):
         svc = PosixFilesystemMailAdapter(sender_dir, mailbox_rel="mailbox")
         result = svc.send(str(recip_dir), {"message": "hello"})
         assert result is not None
-        assert "not running" in result.lower()
+        assert "not currently deliverable" in result.lower()
+        assert "heartbeat" in result.lower()
+
+    def test_send_retries_transient_stale_heartbeat(self, tmp_path, monkeypatch):
+        """A heartbeat that becomes fresh during the grace window should deliver."""
+        import lingtai.adapters.posix.mail as mail_mod
+        from lingtai.adapters.posix.mail import PosixFilesystemMailAdapter
+
+        monkeypatch.setattr(mail_mod, "MAIL_DELIVERY_LIVENESS_GRACE_SECONDS", 0.5)
+        monkeypatch.setattr(mail_mod, "MAIL_DELIVERY_LIVENESS_RETRY_INTERVAL_SECONDS", 0.01)
+
+        sender_dir = _make_agent_dir(tmp_path, "sender01")
+        recip_dir = _make_agent_dir(tmp_path, "recip01")
+        (recip_dir / "mailbox" / "inbox").mkdir(parents=True)
+        heartbeat = recip_dir / ".agent.heartbeat"
+        heartbeat.write_text(str(time.time() - 10))
+
+        def _refresh_heartbeat():
+            time.sleep(0.05)
+            heartbeat.write_text(str(time.time()))
+
+        thread = threading.Thread(target=_refresh_heartbeat)
+        thread.start()
+        try:
+            svc = PosixFilesystemMailAdapter(sender_dir, mailbox_rel="mailbox")
+            result = svc.send(str(recip_dir), {"message": "hello after lag"})
+        finally:
+            thread.join(timeout=1.0)
+
+        assert result is None
+        inbox = recip_dir / "mailbox" / "inbox"
+        msgs = list(inbox.iterdir())
+        assert len(msgs) == 1
+        data = json.loads((msgs[0] / "message.json").read_text())
+        assert data["message"] == "hello after lag"
 
     def test_send_self(self, tmp_path):
         """Send to own address should work (self-send)."""
@@ -160,9 +198,12 @@ def test_send_to_human_skips_heartbeat(self, tmp_path):
         msg = json.loads((entries[0] / "message.json").read_text())
         assert msg["message"] == "hello human"
 
-    def test_send_fails_no_heartbeat_file(self, tmp_path):
+    def test_send_fails_no_heartbeat_file(self, tmp_path, monkeypatch):
+        import lingtai.adapters.posix.mail as mail_mod
         from lingtai.adapters.posix.mail import PosixFilesystemMailAdapter
 
+        monkeypatch.setattr(mail_mod, "MAIL_DELIVERY_LIVENESS_GRACE_SECONDS", 0.0)
+
         sender_dir = _make_agent_dir(tmp_path, "sender01")
         recip_dir = tmp_path / "recip01"
         recip_dir.mkdir()
@@ -175,7 +216,8 @@ def test_send_fails_no_heartbeat_file(self, tmp_path):
         svc = PosixFilesystemMailAdapter(sender_dir, mailbox_rel="mailbox")
         result = svc.send(str(recip_dir), {"message": "hello"})
         assert result is not None
-        assert "not running" in result.lower()
+        assert "not currently deliverable" in result.lower()
+        assert "heartbeat" in result.lower()
 
     def test_send_atomic_write(self, tmp_path):
         """Verify no .tmp file is left behind after successful send."""