feat: serve durable inbound SMS history

callstack/events/types.py

@@ -42,6 +42,9 @@ class IncomingSMSEvent(Event):
     sender: str = ""
     body: str = ""
     raw: str = ""
+    # Direct +CMT delivery may still be emitted after a local-store failure so
+    # subscribers can handle it; callers can retain a bounded fallback record.
+    persisted: bool = True
 
 
 @dataclass(frozen=True)

callstack/sms/service.py

@@ -256,16 +256,19 @@ async def _on_incoming(self, event: _RawSMSNotification) -> None:
                 sender=sender,
                 body=event.body,
                 status="unread",
-                timestamp=datetime.now(),
+                timestamp=datetime.now(timezone.utc),
             )
             try:
                 await self._store.save(sms)
             except Exception as exc:
                 logger.warning(
                     "Failed to persist direct SMS delivery (%s)", type(exc).__name__
                 )
+                persisted = False
+            else:
+                persisted = True
             await self._bus.emit(
-                IncomingSMSEvent(sender=sender, body=event.body)
+                IncomingSMSEvent(sender=sender, body=event.body, persisted=persisted)
             )
             logger.info("Incoming SMS from %s (direct)", redact_phone_number(sender))
 
@@ -441,6 +444,10 @@ async def messages(self, filter_sender: Optional[str] = None):
 
     # -- Message Management --
 
+    async def list_persisted_messages(self, limit: int = 100) -> list[SMS]:
+        """List locally persisted inbound SMS history without reading the SIM."""
+        return await self._store.list_incoming(limit=limit)
+
     async def list_delivery_reports(self, limit: int = 100) -> list[DeliveryReport]:
         """List delivery reports persisted by the SMS store."""
         return await self._store.list_delivery_reports(limit=limit)

callstack/sms/store.py

@@ -5,7 +5,7 @@
 import logging
 import os
 from datetime import datetime, timezone
-from typing import Optional
+from typing import List, Optional
 
 from callstack.sms.types import DeliveryReport, SMS
 
@@ -407,6 +407,12 @@ async def list(
                 results = [m for m in results if m.status == status]
             return results[-limit:]
 
+    async def list_incoming(self, limit: int = 100) -> List[SMS]:
+        """List locally persisted inbound SMS history, newest last."""
+        limit = _validate_list_limit(limit)
+        async with self._lock:
+            return [sms for sms in self._messages if sms.is_incoming][-limit:]
+
     async def delete(self, id: int) -> bool:
         """Delete a message by internal ID."""
         async with self._lock:

server.py

@@ -8,8 +8,9 @@
 import time
 from collections import defaultdict
 from contextlib import suppress
+from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any
+from typing import Any, Awaitable, Callable, cast
 
 import aiohttp
 from aiohttp import web
@@ -221,6 +222,39 @@ def _delivery_report_payload(report: Any) -> dict[str, Any]:
     }
 
 
+def _received_sms_payload(sms: Any) -> dict[str, Any]:
+    """Serialize durable inbound history in the legacy received-message shape."""
+    if isinstance(sms, dict):
+        return {
+            "sender": sms.get("sender", ""),
+            "body": sms.get("body", ""),
+            "received_at": sms.get("received_at"),
+        }
+    timestamp = getattr(sms, "timestamp", None)
+    return {
+        "sender": getattr(sms, "sender", ""),
+        "body": getattr(sms, "body", ""),
+        "received_at": timestamp.isoformat() if timestamp else None,
+    }
+
+
+def _received_sms_sort_key(message: dict[str, Any]) -> datetime:
+    """Return an aware UTC ordering key without rejecting legacy bad timestamps."""
+    timestamp = message.get("received_at")
+    if not isinstance(timestamp, str):
+        return datetime.min.replace(tzinfo=timezone.utc)
+    try:
+        received_at = datetime.fromisoformat(timestamp)
+    except (TypeError, ValueError):
+        return datetime.min.replace(tzinfo=timezone.utc)
+    if received_at.tzinfo is None:
+        return received_at.replace(tzinfo=timezone.utc)
+    try:
+        return received_at.astimezone(timezone.utc)
+    except OverflowError:
+        return datetime.min.replace(tzinfo=timezone.utc)
+
+
 def _is_sms_body_encoding_error(exc: SMSSendError) -> bool:
     return "SMS body cannot be encoded" in exc.detail
 
@@ -400,7 +434,26 @@ async def subscribe(request: web.Request) -> web.Response:
         return web.json_response({"status": "subscribed", "url": url})
 
     async def list_messages(request: web.Request) -> web.Response:
-        return web.json_response(received_messages)
+        limit, error = _bounded_query_limit(request)
+        if error is not None:
+            return error
+        assert limit is not None
+        list_persisted_messages = cast(
+            Callable[..., Awaitable[list[Any]]] | None,
+            getattr(modem.sms, "list_persisted_messages", None),
+        )
+        if callable(list_persisted_messages):
+            messages = await list_persisted_messages(limit=limit)
+            persisted_messages = [
+                _received_sms_payload(message)
+                for message in messages
+                if getattr(message, "is_incoming", bool(getattr(message, "sender", "")))
+            ]
+            fallback_messages = [_received_sms_payload(message) for message in received_messages]
+            history = persisted_messages + fallback_messages
+            history.sort(key=_received_sms_sort_key)
+            return web.json_response(history[-limit:])
+        return web.json_response([_received_sms_payload(message) for message in received_messages[-limit:]])
 
     async def list_delivery_reports(request: web.Request) -> web.Response:
         limit, error = _bounded_query_limit(request)
@@ -491,11 +544,15 @@ async def handle_call(session: CallSession) -> None:
 
         # -- SMS handling: store + forward to webhooks --
         async def on_sms(event: IncomingSMSEvent) -> None:
-            received_messages.append({
-                "sender": event.sender,
-                "body": event.body,
-                "received_at": event.timestamp.isoformat(),
-            })
+            if (
+                not callable(getattr(modem.sms, "list_persisted_messages", None))
+                or not event.persisted
+            ):
+                received_messages.append({
+                    "sender": event.sender,
+                    "body": event.body,
+                    "received_at": event.timestamp.isoformat(),
+                })
             await notify_webhooks(event.sender, event.body)
 
         modem.sms.on_message(on_sms)