diff --git a/callstack/events/types.py b/callstack/events/types.py index 62b4051..6dc631d 100644 --- a/callstack/events/types.py +++ b/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) diff --git a/callstack/sms/service.py b/callstack/sms/service.py index 729fd30..672290f 100644 --- a/callstack/sms/service.py +++ b/callstack/sms/service.py @@ -256,7 +256,7 @@ 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) @@ -264,8 +264,11 @@ async def _on_incoming(self, event: _RawSMSNotification) -> None: 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) diff --git a/callstack/sms/store.py b/callstack/sms/store.py index 219d951..c38e689 100644 --- a/callstack/sms/store.py +++ b/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: diff --git a/server.py b/server.py index c83490d..5d1ef7b 100644 --- a/server.py +++ b/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) diff --git a/tests/test_api_auth.py b/tests/test_api_auth.py index eb4950b..e18eb8d 100644 --- a/tests/test_api_auth.py +++ b/tests/test_api_auth.py @@ -16,8 +16,11 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) from callstack.events.bus import EventBus +from callstack.events.types import _RawSMSNotification from callstack.protocol.executor import ATCommandExecutor -from callstack.sms.types import DeliveryReport +from callstack.sms.service import SMSService +from callstack.sms.store import SMSStore +from callstack.sms.types import DeliveryReport, SMS from callstack.ussd import USSDService import server from server import APIKeyAuth, create_app @@ -271,6 +274,443 @@ async def post(self, url, **_kwargs): assert "private sms body secret" not in caplog.text +class TestServerSMSRecording: + @pytest.mark.parametrize("has_durable_history", [True, False]) + async def test_run_server_records_messages_globally_only_for_legacy_sms_services( + self, monkeypatch, has_durable_history + ): + previous_messages = list(server.received_messages) + server.received_messages.clear() + webhook_calls = [] + + async def record_webhook(sender, body): + webhook_calls.append((sender, body)) + + class FakeSMS: + def on_message(self, callback): + self.callback = callback + + fake_sms = FakeSMS() + if has_durable_history: + async def list_persisted_messages(limit=100): + return [] + + setattr(fake_sms, "list_persisted_messages", list_persisted_messages) + + class FakeModem: + instance = None + + def __init__(self, _config): + self.sms = fake_sms + self.bus = EventBus() + FakeModem.instance = self + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return None + + def on_call(self, callback): + return callback + + async def run_forever(self): + await self.sms.callback( + server.IncomingSMSEvent( + sender="5551234", + body="inbound message", + timestamp=datetime(2026, 7, 27, tzinfo=timezone.utc), + ) + ) + + class FakeRunner: + def __init__(self, _app): + pass + + async def setup(self): + pass + + async def cleanup(self): + pass + + class FakeSite: + def __init__(self, _runner, _host, _port): + pass + + async def start(self): + pass + + monkeypatch.setattr(server, "Modem", FakeModem) + monkeypatch.setattr(server, "notify_webhooks", record_webhook) + monkeypatch.setattr(server, "create_app", lambda *_args, **_kwargs: object()) + monkeypatch.setattr(server.web, "AppRunner", FakeRunner) + monkeypatch.setattr(server.web, "TCPSite", FakeSite) + try: + await server.run_server(cast(server.ModemConfig, SimpleNamespace())) + finally: + recorded_messages = list(server.received_messages) + server.received_messages[:] = previous_messages + + assert webhook_calls == [("5551234", "inbound message")] + if has_durable_history: + assert recorded_messages == [] + else: + assert recorded_messages == [ + { + "sender": "5551234", + "body": "inbound message", + "received_at": "2026-07-27T00:00:00+00:00", + } + ] + + async def test_run_server_records_unpersisted_durable_sms_events_for_http_fallback( + self, monkeypatch + ): + previous_messages = list(server.received_messages) + server.received_messages.clear() + webhook_calls = [] + + async def record_webhook(sender, body): + webhook_calls.append((sender, body)) + + class FakeSMS: + async def list_persisted_messages(self, limit=100): + return [] + + def on_message(self, callback): + self.callback = callback + + fake_sms = FakeSMS() + + class FakeModem: + def __init__(self, _config): + self.sms = fake_sms + self.bus = EventBus() + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return None + + def on_call(self, callback): + return callback + + async def run_forever(self): + await self.sms.callback( + server.IncomingSMSEvent( + sender="5551234", + body="unpersisted direct message", + timestamp=datetime(2026, 7, 27, tzinfo=timezone.utc), + persisted=False, + ) + ) + + class FakeRunner: + def __init__(self, _app): + pass + + async def setup(self): + pass + + async def cleanup(self): + pass + + class FakeSite: + def __init__(self, _runner, _host, _port): + pass + + async def start(self): + pass + + monkeypatch.setattr(server, "Modem", FakeModem) + monkeypatch.setattr(server, "notify_webhooks", record_webhook) + monkeypatch.setattr(server, "create_app", lambda *_args, **_kwargs: object()) + monkeypatch.setattr(server.web, "AppRunner", FakeRunner) + monkeypatch.setattr(server.web, "TCPSite", FakeSite) + try: + await server.run_server(cast(server.ModemConfig, SimpleNamespace())) + finally: + recorded_messages = list(server.received_messages) + server.received_messages[:] = previous_messages + + assert webhook_calls == [("5551234", "unpersisted direct message")] + assert recorded_messages == [ + { + "sender": "5551234", + "body": "unpersisted direct message", + "received_at": "2026-07-27T00:00:00+00:00", + } + ] + + +class TestSMSMessagesEndpoint: + async def test_messages_endpoint_reads_only_inbound_persisted_history_in_legacy_shape(self, aiohttp_client): + class FakeSMS: + def __init__(self): + self.limits = [] + + async def list_persisted_messages(self, limit=100): + self.limits.append(limit) + return [ + SMS( + id=7, + sender="5551234", + recipient="", + body="durable message", + timestamp=datetime(2026, 7, 27, tzinfo=timezone.utc), + status="unread", + reference=0, + storage_index=4, + ), + SMS( + id=8, + recipient="5556789", + body="outbound message", + timestamp=datetime(2026, 7, 27, tzinfo=timezone.utc), + status="sent", + reference=8, + ), + ] + + fake_sms = FakeSMS() + modem = SimpleNamespace(sms=fake_sms, ussd=SimpleNamespace(), bus=EventBus(), connected=True) + client = await aiohttp_client(create_app(modem)) + + resp = await client.get("/sms/messages?limit=2") + + assert resp.status == 200 + assert fake_sms.limits == [2] + assert await resp.json() == [ + { + "sender": "5551234", + "body": "durable message", + "received_at": "2026-07-27T00:00:00+00:00", + } + ] + + async def test_messages_endpoint_merges_interleaved_durable_and_fallback_history_chronologically(self, aiohttp_client): + previous_messages = list(server.received_messages) + server.received_messages[:] = [ + {"sender": "111", "body": "first fallback", "received_at": "2026-07-27T00:00:00+00:00"}, + {"sender": "222", "body": "middle fallback", "received_at": "2026-07-27T00:02:00+00:00"}, + ] + + class FakeSMS: + async def list_persisted_messages(self, limit=100): + return [ + SMS( + sender="333", + body="older durable inbound", + timestamp=datetime(2026, 7, 27, 0, 1, tzinfo=timezone.utc), + status="unread", + ), + SMS( + sender="444", + body="newest durable inbound", + timestamp=datetime(2026, 7, 27, 0, 3, tzinfo=timezone.utc), + status="unread", + ) + ] + + modem = SimpleNamespace(sms=FakeSMS(), ussd=SimpleNamespace(), bus=EventBus(), connected=True) + client = await aiohttp_client(create_app(modem)) + try: + resp = await client.get("/sms/messages?limit=1") + payload = await resp.json() + finally: + server.received_messages[:] = previous_messages + + assert resp.status == 200 + assert payload == [ + {"sender": "444", "body": "newest durable inbound", "received_at": "2026-07-27T00:03:00+00:00"} + ] + + async def test_messages_endpoint_treats_utc_overflow_legacy_fallback_timestamp_as_oldest( + self, aiohttp_client + ): + previous_messages = list(server.received_messages) + server.received_messages[:] = [ + { + "sender": "111", + "body": "UTC overflow fallback", + "received_at": "0001-01-01T00:00:00+23:59", + } + ] + + class FakeSMS: + async def list_persisted_messages(self, limit=100): + return [ + SMS( + sender="222", + body="newer durable inbound", + timestamp=datetime(2026, 7, 27, tzinfo=timezone.utc), + status="unread", + ) + ] + + modem = SimpleNamespace(sms=FakeSMS(), ussd=SimpleNamespace(), bus=EventBus(), connected=True) + client = await aiohttp_client(create_app(modem)) + try: + resp = await client.get("/sms/messages?limit=1") + payload = await resp.json() + finally: + server.received_messages[:] = previous_messages + + assert resp.status == 200 + assert payload == [ + { + "sender": "222", + "body": "newer durable inbound", + "received_at": "2026-07-27T00:00:00+00:00", + } + ] + + async def test_messages_endpoint_serializes_direct_delivery_timestamp_as_aware_utc( + self, aiohttp_client + ): + previous_messages = list(server.received_messages) + server.received_messages.clear() + try: + store = SMSStore() + bus = EventBus() + sms = SMSService(cast(ATCommandExecutor, SimpleNamespace()), bus, store) + await sms._on_incoming( + _RawSMSNotification( + sender="5551234", + body="direct message", + raw='+CMT: "5551234","","26/07/27,00:00:00+00"', + ) + ) + modem = SimpleNamespace(sms=sms, ussd=SimpleNamespace(), bus=bus, connected=True) + client = await aiohttp_client(create_app(modem)) + + resp = await client.get("/sms/messages?limit=1") + payload = await resp.json() + finally: + server.received_messages[:] = previous_messages + + assert resp.status == 200 + assert payload[0]["sender"] == "5551234" + assert payload[0]["body"] == "direct message" + assert set(payload[0]) == {"sender", "body", "received_at"} + assert datetime.fromisoformat(payload[0]["received_at"]).tzinfo == timezone.utc + + async def test_messages_endpoint_legacy_fallback_is_bounded_and_preserves_legacy_shape(self, aiohttp_client): + previous_messages = list(server.received_messages) + server.received_messages[:] = [ + {"sender": "111", "body": "first", "received_at": "2026-07-27T00:00:00+00:00"}, + {"sender": "222", "body": "second", "received_at": "2026-07-27T00:01:00+00:00"}, + {"sender": "333", "body": "third", "received_at": "2026-07-27T00:02:00+00:00"}, + ] + modem = SimpleNamespace(sms=SimpleNamespace(), ussd=SimpleNamespace(), bus=EventBus(), connected=True) + client = await aiohttp_client(create_app(modem)) + try: + resp = await client.get("/sms/messages?limit=2") + payload = await resp.json() + finally: + server.received_messages[:] = previous_messages + + assert resp.status == 200 + assert payload == [ + {"sender": "222", "body": "second", "received_at": "2026-07-27T00:01:00+00:00"}, + {"sender": "333", "body": "third", "received_at": "2026-07-27T00:02:00+00:00"}, + ] + + async def test_messages_endpoint_uses_safe_default_limit_for_persisted_history(self, aiohttp_client): + class FakeSMS: + def __init__(self): + self.limits = [] + + async def list_persisted_messages(self, limit=100): + self.limits.append(limit) + return [] + + fake_sms = FakeSMS() + modem = SimpleNamespace(sms=fake_sms, ussd=SimpleNamespace(), bus=EventBus(), connected=True) + client = await aiohttp_client(create_app(modem)) + + resp = await client.get("/sms/messages") + + assert resp.status == 200 + assert fake_sms.limits == [50] + assert await resp.json() == [] + + async def test_messages_endpoint_rejects_invalid_limit_before_reading_persisted_history(self, aiohttp_client): + class FakeSMS: + def __init__(self): + self.limits = [] + + async def list_persisted_messages(self, limit=100): + self.limits.append(limit) + raise AssertionError("invalid limits must not reach persisted history") + + fake_sms = FakeSMS() + modem = SimpleNamespace(sms=fake_sms, ussd=SimpleNamespace(), bus=EventBus(), connected=True) + client = await aiohttp_client(create_app(modem)) + + resp = await client.get("/sms/messages?limit=0") + + assert resp.status == 400 + assert await resp.json() == {"error": "invalid 'limit'"} + assert fake_sms.limits == [] + + async def test_messages_endpoint_preserves_sqlite_history_across_app_recreation(self, aiohttp_client, tmp_path): + db_path = str(tmp_path / "sms.db") + expected_history = [ + { + "sender": "5551234", + "body": "saved before restart", + "received_at": "2026-07-27T00:00:00+00:00", + } + ] + + store_before_restart = SMSStore(db_path=db_path) + try: + await store_before_restart.initialize() + await store_before_restart.save( + SMS( + sender="5551234", + body="saved before restart", + timestamp=datetime(2026, 7, 27, tzinfo=timezone.utc), + status="unread", + ) + ) + await store_before_restart.save( + SMS(recipient="5556789", body="outbound before restart", status="sent") + ) + sms_before_restart = SMSService( + cast(ATCommandExecutor, SimpleNamespace()), EventBus(), store_before_restart + ) + modem_before_restart = SimpleNamespace( + sms=sms_before_restart, ussd=SimpleNamespace(), bus=EventBus(), connected=True + ) + client_before_restart = await aiohttp_client(create_app(modem_before_restart)) + response_before_restart = await client_before_restart.get("/sms/messages") + + assert response_before_restart.status == 200 + assert await response_before_restart.json() == expected_history + finally: + await store_before_restart.close() + + store_after_restart = SMSStore(db_path=db_path) + try: + await store_after_restart.initialize() + sms_after_restart = SMSService( + cast(ATCommandExecutor, SimpleNamespace()), EventBus(), store_after_restart + ) + modem_after_restart = SimpleNamespace( + sms=sms_after_restart, ussd=SimpleNamespace(), bus=EventBus(), connected=True + ) + client_after_restart = await aiohttp_client(create_app(modem_after_restart)) + response_after_restart = await client_after_restart.get("/sms/messages") + + assert response_after_restart.status == 200 + assert await response_after_restart.json() == expected_history + finally: + await store_after_restart.close() + + class TestDeliveryReportEndpoint: async def test_delivery_reports_endpoint_reads_sms_store_with_limit_and_redacts_recipient(self, aiohttp_client): class FakeSMS: diff --git a/tests/test_sms_service.py b/tests/test_sms_service.py index a6cc06a..7c9bab3 100644 --- a/tests/test_sms_service.py +++ b/tests/test_sms_service.py @@ -638,6 +638,7 @@ async def track(e): # The re-emitted enriched event (empty raw, populated body) enriched = [e for e in all_events if e.body == "Direct message" and not e.raw] assert len(enriched) >= 1 + assert enriched[0].persisted is True assert await store.count() == 1 @@ -666,6 +667,7 @@ async def track(event): assert len(received) == 1 assert received[0].sender == sender assert received[0].body == body + assert received[0].persisted is False assert "RuntimeError" in caplog.text assert sender not in caplog.text assert body not in caplog.text @@ -822,6 +824,14 @@ async def test_receive_cmt_info_log_redacts_sender_number(sms_service, bus, capl # -- Message Management -- +async def test_list_persisted_messages_returns_only_local_inbound_history(sms_service, store): + """Durable history stays distinct from SIM listing and excludes outbound SMS.""" + inbound = await store.save(SMS(sender="5551234", body="received", status="unread")) + await store.save(SMS(recipient="5556789", body="sent", status="sent")) + + assert await sms_service.list_persisted_messages(limit=1) == [inbound] + + async def test_list_messages(sms_service, transport): """List messages from SIM.""" transport.feed(