From ae6f968454096bfb441c48b7322333db955ab47d Mon Sep 17 00:00:00 2001 From: Vladislav Poverin <123168793+vlobus@users.noreply.github.com> Date: Tue, 14 Jul 2026 22:17:21 +0200 Subject: [PATCH 1/2] t9 telegram: thin bot api client over shared connector resilience, single-id allowlist poller in lifespan, notify job with claim-send-commit once semantics on notified_at, html-escaped formatting --- .env.example | 8 + .../recommend/src/hodlin_recommend/config.py | 7 + .../src/hodlin_recommend/connectors/base.py | 23 +- .../src/hodlin_recommend/delivery/__init__.py | 1 + .../hodlin_recommend/delivery/formatting.py | 42 ++++ .../src/hodlin_recommend/delivery/poller.py | 81 ++++++ .../src/hodlin_recommend/delivery/telegram.py | 111 +++++++++ .../src/hodlin_recommend/ingest/jobs.py | 40 +++ .../src/hodlin_recommend/ingest/scheduler.py | 16 +- .../recommend/src/hodlin_recommend/main.py | 17 ++ .../src/hodlin_recommend/serving/app.py | 19 ++ .../hodlin_recommend/store/repositories.py | 79 +++++- tests/integration/test_jobs.py | 7 + tests/integration/test_notify.py | 123 ++++++++++ tests/test_delivery.py | 230 ++++++++++++++++++ tests/test_scheduler.py | 13 +- 16 files changed, 808 insertions(+), 9 deletions(-) create mode 100644 packages/recommend/src/hodlin_recommend/delivery/__init__.py create mode 100644 packages/recommend/src/hodlin_recommend/delivery/formatting.py create mode 100644 packages/recommend/src/hodlin_recommend/delivery/poller.py create mode 100644 packages/recommend/src/hodlin_recommend/delivery/telegram.py create mode 100644 tests/integration/test_notify.py create mode 100644 tests/test_delivery.py diff --git a/.env.example b/.env.example index 8f1a698..cab32da 100644 --- a/.env.example +++ b/.env.example @@ -20,3 +20,11 @@ MASSIVE_RATE_PER_MIN=5 ANTHROPIC_API_KEY=your-anthropic-key # One short call per anomaly; the cheap/fast tier is plenty. ANTHROPIC_MODEL=claude-haiku-4-5-20251001 + +# --- Telegram (delivery) — token from @BotFather ---------------------------- +TELEGRAM_BOT_TOKEN=your-bot-token +TELEGRAM_BASE_URL=https://api.telegram.org +# Your numeric user id (message @userinfobot to find it). This single id is +# both the alert destination and the allowlist — everyone else is ignored. +TELEGRAM_CHAT_ID=123456789 +TELEGRAM_RATE_PER_MIN=30 diff --git a/packages/recommend/src/hodlin_recommend/config.py b/packages/recommend/src/hodlin_recommend/config.py index b8b4f71..5d928c4 100644 --- a/packages/recommend/src/hodlin_recommend/config.py +++ b/packages/recommend/src/hodlin_recommend/config.py @@ -33,3 +33,10 @@ class Settings(BaseSettings): # Anthropic — the single explanation call per anomaly (T7). anthropic_api_key: str anthropic_model: str + + # Telegram — delivery (T9). ``chat_id`` is both the destination and the + # single-ID allowlist: inbound messages from anyone else are dropped. + telegram_bot_token: str + telegram_base_url: str + telegram_chat_id: int + telegram_rate_per_min: float diff --git a/packages/recommend/src/hodlin_recommend/connectors/base.py b/packages/recommend/src/hodlin_recommend/connectors/base.py index 63fd9e3..3751af2 100644 --- a/packages/recommend/src/hodlin_recommend/connectors/base.py +++ b/packages/recommend/src/hodlin_recommend/connectors/base.py @@ -116,18 +116,33 @@ async def request_json( source: str, url: str, params: Mapping[str, str] | None = None, + method: str = "GET", + json_body: Any | None = None, + http_timeout: httpx.Timeout | None = None, rate: RateLimiter | None = None, retry: RetryPolicy = DEFAULT_RETRY, ) -> Any: - """GET ``url`` and return parsed JSON, applying the shared rate-limit + retry - policy and wrapping any HTTP failure as ``SourceUnavailable``.""" + """Call ``url`` and return parsed JSON, applying the shared rate-limit + + retry policy and wrapping any HTTP failure as ``SourceUnavailable``. + + ``method``/``json_body`` exist for POST-style APIs (Telegram); + ``http_timeout`` overrides the client default per request (a long poll + must outlive it). Note retries make non-idempotent POSTs at-least-once: + a reply lost on the wire is retried even if the server acted — callers + choose semantics. + """ + request_timeout = http_timeout if http_timeout is not None else httpx.USE_CLIENT_DEFAULT async def _once() -> Any: if rate is not None: async with rate: - response = await client.get(url, params=params) + response = await client.request( + method, url, params=params, json=json_body, timeout=request_timeout + ) else: - response = await client.get(url, params=params) + response = await client.request( + method, url, params=params, json=json_body, timeout=request_timeout + ) response.raise_for_status() # parse_float=Decimal keeps money exact: JSON numbers never become floats # (which the domain models reject), so precision survives from the wire. diff --git a/packages/recommend/src/hodlin_recommend/delivery/__init__.py b/packages/recommend/src/hodlin_recommend/delivery/__init__.py new file mode 100644 index 0000000..9baceb2 --- /dev/null +++ b/packages/recommend/src/hodlin_recommend/delivery/__init__.py @@ -0,0 +1 @@ +"""Telegram delivery (T9): outbound anomaly alerts + a thin inbound poller.""" diff --git a/packages/recommend/src/hodlin_recommend/delivery/formatting.py b/packages/recommend/src/hodlin_recommend/delivery/formatting.py new file mode 100644 index 0000000..6a5a546 --- /dev/null +++ b/packages/recommend/src/hodlin_recommend/delivery/formatting.py @@ -0,0 +1,42 @@ +"""Message formatting — pure functions, and the escaping boundary (T9). + +Messages use Telegram HTML parse mode, so every dynamic field is escaped +*here*, at the one place text becomes markup. The reasoning is LLM prose over +untrusted headlines (T7): it may quote anything, and none of it may become +tags. What stays literal markup is only what this module writes itself. +""" + +from html import escape + +from hodlin_recommend.domain.models import Anomaly, Explanation + + +def format_anomaly(anomaly: Anomaly, explanation: Explanation) -> str: + """One alert: the move in numbers, then the LLM's why, then lineage.""" + arrow = ( + "\N{CHART WITH UPWARDS TREND}" + if anomaly.direction == "up" + else ("\N{CHART WITH DOWNWARDS TREND}") + ) + news_cited = sum(1 for ref in explanation.evidence if ref.kind == "news") + lines = [ + f"{arrow} {escape(anomaly.symbol)} {escape(anomaly.interval)} " + f"bar {anomaly.bar_ts:%Y-%m-%d %H:%M} UTC", + f"move {anomaly.return_pct:+f}% \N{MIDDLE DOT} z-score {anomaly.z_score} " + f"\N{MIDDLE DOT} baseline {anomaly.window} bars", + "", + escape(explanation.reasoning), + "", + f"{news_cited} news source(s) cited \N{MIDDLE DOT} " + f"{escape(explanation.model_version)}", + ] + return "\n".join(lines) + + +def format_status(latest: tuple[Anomaly, Explanation] | None) -> str: + """The reply to any allowlisted inbound message: the newest explained + anomaly, or an honest 'nothing yet'.""" + if latest is None: + return "No explained anomalies yet \N{EM DASH} you'll be notified here when one lands." + anomaly, explanation = latest + return "Latest anomaly:\n\n" + format_anomaly(anomaly, explanation) diff --git a/packages/recommend/src/hodlin_recommend/delivery/poller.py b/packages/recommend/src/hodlin_recommend/delivery/poller.py new file mode 100644 index 0000000..2143f0a --- /dev/null +++ b/packages/recommend/src/hodlin_recommend/delivery/poller.py @@ -0,0 +1,81 @@ +"""Inbound long-poll loop (T9): thin by design, default-deny by design. + +The poller reads raw updates and enforces the single-ID allowlist where +messages *enter* the system: an update whose chat or sender isn't the one +allowlisted ID is dropped without a reply (strangers learn nothing, not even +that the bot exists). An allowlisted message gets the latest explained +anomaly back — the reply text comes through an injected async callable so the +poller itself never touches the database and unit tests never need one. + +Runs as a background task owned by the app lifespan; cancellation is the stop +signal. A dead Telegram API is absorbed with a backoff, same policy as any +dead source: the loop must outlive the outage. +""" + +import asyncio +from collections.abc import Awaitable, Callable +from typing import Any, Protocol + +from hodlin_recommend.connectors.base import SourceUnavailable +from hodlin_recommend.delivery.formatting import format_status +from hodlin_recommend.delivery.telegram import Messenger, UpdateSource +from hodlin_recommend.store.db import SessionFactory +from hodlin_recommend.store.repositories import AnomalyRepository + +# Tuning, not secrets (D17): how long to sit out a Telegram outage before the +# next poll attempt. +ERROR_BACKOFF_S = 5.0 + + +class TelegramAPI(Messenger, UpdateSource, Protocol): + """Both halves — what the poller needs (read updates, send replies).""" + + +class UpdatePoller: + def __init__( + self, + api: TelegramAPI, + *, + allowed_chat_id: int, + reply_text: Callable[[], Awaitable[str]], + ) -> None: + self._api = api + self._allowed = allowed_chat_id + self._reply_text = reply_text + + async def run(self) -> None: + """Poll forever; the owner cancels this task to stop it. ``offset`` + acknowledges processed updates so Telegram never redelivers them.""" + offset: int | None = None + while True: + try: + updates = await self._api.get_updates(offset) + except SourceUnavailable: + await asyncio.sleep(ERROR_BACKOFF_S) + continue + for update in updates: + offset = int(update["update_id"]) + 1 + await self._handle(update) + + async def _handle(self, update: dict[str, Any]) -> None: + message = update.get("message") or {} + chat_id = (message.get("chat") or {}).get("id") + sender_id = (message.get("from") or {}).get("id") + if chat_id != self._allowed or sender_id != self._allowed: + return # default-deny: no reply, no error, no acknowledgement + try: + await self._api.send(self._allowed, await self._reply_text()) + except SourceUnavailable: + return # the reply is best-effort; the user can just ask again + + +def latest_anomaly_reply(session_factory: SessionFactory) -> Callable[[], Awaitable[str]]: + """The production reply builder: a fresh session per inbound message, + the newest explained anomaly formatted, or an honest 'nothing yet'.""" + + async def reply() -> str: + async with session_factory() as session: + latest = await AnomalyRepository(session).latest_explained() + return format_status(latest) + + return reply diff --git a/packages/recommend/src/hodlin_recommend/delivery/telegram.py b/packages/recommend/src/hodlin_recommend/delivery/telegram.py new file mode 100644 index 0000000..fe69a93 --- /dev/null +++ b/packages/recommend/src/hodlin_recommend/delivery/telegram.py @@ -0,0 +1,111 @@ +"""Thin Telegram Bot API client behind Protocol seams (T9). + +Two endpoints are all delivery needs — ``sendMessage`` out, ``getUpdates`` +in — so this is a hand-rolled client over the shared connector resilience +(rate limit + retry + ``SourceUnavailable``), not a bot framework: Telegram +is a sink that fails exactly like a source, and the jobs already know how to +treat a dead source. ``Messenger`` is the outbound seam the notify job mocks; +``UpdateSource`` is the inbound seam the poller mocks; ``TelegramClient`` is +the one concrete implementing both. + +Delivery semantics are at-least-once by construction: the shared retry can +repeat a ``sendMessage`` whose reply was lost after the server acted, and the +notify job commits its claim only after a successful send. A rare duplicate +alert beats a silently missing one. +""" + +from typing import Any, Protocol, runtime_checkable + +import httpx + +from hodlin_recommend.connectors.base import ( + DEFAULT_RETRY, + RateLimiter, + RetryPolicy, + SourceUnavailable, + request_json, +) + +# Tuning, not secrets (D17): how long Telegram holds a getUpdates open. The +# HTTP read timeout must outlive it or every idle poll "times out". +LONG_POLL_S = 50 + + +@runtime_checkable +class Messenger(Protocol): + """Outbound half — everything the notify job needs.""" + + async def send(self, chat_id: int, text: str) -> None: ... + + +@runtime_checkable +class UpdateSource(Protocol): + """Inbound half — everything the poller reads. Returns raw Bot API update + dicts; interpreting them (and enforcing the allowlist) is the poller's job.""" + + async def get_updates(self, offset: int | None) -> list[dict[str, Any]]: ... + + +class TelegramClient: + """The one concrete: both halves over the HTTP Bot API.""" + + source = "telegram" + + def __init__( + self, + client: httpx.AsyncClient, + *, + token: str, + base_url: str, + rate: RateLimiter, + retry: RetryPolicy = DEFAULT_RETRY, + ) -> None: + self._client = client + # Telegram puts the secret in the path; it must never be logged. + self._base = f"{base_url.rstrip('/')}/bot{token}" + self._rate = rate + self._retry = retry + + async def _call( + self, + api_method: str, + payload: dict[str, Any], + *, + http_timeout: httpx.Timeout | None = None, + ) -> Any: + data = await request_json( + self._client, + source=self.source, + url=f"{self._base}/{api_method}", + method="POST", + json_body=payload, + http_timeout=http_timeout, + rate=self._rate, + retry=self._retry, + ) + # Telegram can answer HTTP 200 with ok=false; that's still a failure. + if not isinstance(data, dict) or not data.get("ok"): + raise SourceUnavailable(self.source, f"API answered not-ok: {data!r}") + return data.get("result") + + async def send(self, chat_id: int, text: str) -> None: + await self._call( + "sendMessage", + { + "chat_id": chat_id, + "text": text, + # HTML parse mode pairs with formatting.py escaping every + # dynamic field — untrusted text can't become markup. + "parse_mode": "HTML", + "disable_web_page_preview": True, + }, + ) + + async def get_updates(self, offset: int | None) -> list[dict[str, Any]]: + payload: dict[str, Any] = {"timeout": LONG_POLL_S, "allowed_updates": ["message"]} + if offset is not None: + payload["offset"] = offset + result = await self._call( + "getUpdates", payload, http_timeout=httpx.Timeout(LONG_POLL_S + 10) + ) + return list(result) if isinstance(result, list) else [] diff --git a/packages/recommend/src/hodlin_recommend/ingest/jobs.py b/packages/recommend/src/hodlin_recommend/ingest/jobs.py index bf3427a..24b25d6 100644 --- a/packages/recommend/src/hodlin_recommend/ingest/jobs.py +++ b/packages/recommend/src/hodlin_recommend/ingest/jobs.py @@ -20,6 +20,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from hodlin_recommend.connectors.base import NewsSource, PriceBarSource, SourceUnavailable +from hodlin_recommend.delivery.formatting import format_anomaly +from hodlin_recommend.delivery.telegram import Messenger from hodlin_recommend.domain.anomaly import detect_series from hodlin_recommend.domain.asset_config import AssetConfig from hodlin_recommend.domain.explanation import ExplainerLLM, LLMUnavailable, MalformedReply @@ -43,6 +45,7 @@ NEWS_LOOKBACK = timedelta(days=3) DETECT_TAIL = 8 EXPLAIN_BATCH = 5 +NOTIFY_BATCH = 5 @dataclass(frozen=True) @@ -206,6 +209,43 @@ async def work(session: AsyncSession) -> JobOutcome: return await run_audited(session_factory, "explain_anomalies", work) +async def notify_anomalies( + session_factory: SessionFactory, + *, + messenger: Messenger, + chat_id: int, +) -> JobOutcome: + """Send each explained-but-unnotified anomaly to the allowlisted chat, + oldest first, at most once. + + The once-guarantee is a compare-and-set on ``anomalies.notified_at``: + claim (uncommitted) -> send -> commit. A failed send rolls the claim back, + so the anomaly is retried next tick; a crash after a successful send but + before the commit re-sends next tick — at-least-once, because a rare + duplicate alert beats a silently missing one. Telegram being down aborts + the batch like any dead sink and marks the run "error". + """ + + async def work(session: AsyncSession) -> JobOutcome: + repo = AnomalyRepository(session) + queue = await repo.explained_unnotified(limit=NOTIFY_BATCH) + sent = 0 + for anomaly, explanation in queue: + claimed = await repo.mark_notified(anomaly.symbol, anomaly.interval, anomaly.bar_ts) + if not claimed: + continue # someone else won the race; their send stands + try: + await messenger.send(chat_id, format_anomaly(anomaly, explanation)) + except SourceUnavailable as exc: + await session.rollback() # release the claim; retry next tick + return JobOutcome(items=sent, detail=str(exc), status="error") + await session.commit() # claim becomes durable only after the send + sent += 1 + return JobOutcome(items=sent) + + return await run_audited(session_factory, "notify_anomalies", work) + + async def run_backfill( session_factory: SessionFactory, source: PriceBarSource, diff --git a/packages/recommend/src/hodlin_recommend/ingest/scheduler.py b/packages/recommend/src/hodlin_recommend/ingest/scheduler.py index e80ac77..7450595 100644 --- a/packages/recommend/src/hodlin_recommend/ingest/scheduler.py +++ b/packages/recommend/src/hodlin_recommend/ingest/scheduler.py @@ -1,4 +1,4 @@ -"""Scheduler wiring (T8, D5): four recurring jobs on an ``AsyncIOScheduler``. +"""Scheduler wiring (T8/T9, D5): five recurring jobs on an ``AsyncIOScheduler``. APScheduler's asyncio scheduler runs coroutine jobs as tasks on the app's own event loop — no extra threads or processes, and the jobs share the process's @@ -9,7 +9,9 @@ missed tick may be and still fire. Intervals are tuning, not secrets (D17): daily bars don't need minute-level -polling; the explain tick is shorter so a fresh anomaly gets its "why" fast. +polling; the explain tick is shorter so a fresh anomaly gets its "why" fast, +and the notify tick shorter still so the explained anomaly reaches Telegram +within a minute. The builder only assembles — construction of the concretes stays in the composition root (``main.py``), and tests hand in fakes. """ @@ -22,6 +24,7 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler from hodlin_recommend.connectors.base import NewsSource, PriceBarSource +from hodlin_recommend.delivery.telegram import Messenger from hodlin_recommend.domain.asset_config import DEFAULT_ASSETS, AssetConfig from hodlin_recommend.domain.explanation import ExplainerLLM from hodlin_recommend.domain.sentiment import SentimentModel @@ -33,6 +36,7 @@ NEWS_EVERY_S = 900 DETECT_EVERY_S = 900 EXPLAIN_EVERY_S = 300 +NOTIFY_EVERY_S = 60 MISFIRE_GRACE_S = 60 JOB_DEFAULTS = { @@ -49,6 +53,8 @@ def build_scheduler( news_source: NewsSource, llm: ExplainerLLM, sentiment_model: SentimentModel, + messenger: Messenger, + chat_id: int, inference_executor: ThreadPoolExecutor | None = None, assets: Sequence[AssetConfig] = DEFAULT_ASSETS, backfill_on_start: bool = True, @@ -91,6 +97,12 @@ def build_scheduler( seconds=EXPLAIN_EVERY_S, id="explain_anomalies", ) + scheduler.add_job( + partial(jobs.notify_anomalies, session_factory, messenger=messenger, chat_id=chat_id), + "interval", + seconds=NOTIFY_EVERY_S, + id="notify_anomalies", + ) if backfill_on_start: # No trigger = a date trigger of "now" — stamped at *build* time, so # the grace must be unlimited: if startup takes longer than the diff --git a/packages/recommend/src/hodlin_recommend/main.py b/packages/recommend/src/hodlin_recommend/main.py index 8430eb0..22b61ee 100644 --- a/packages/recommend/src/hodlin_recommend/main.py +++ b/packages/recommend/src/hodlin_recommend/main.py @@ -22,6 +22,8 @@ from hodlin_recommend.connectors.base import RateLimiter from hodlin_recommend.connectors.finnhub import FinnhubNewsSource from hodlin_recommend.connectors.massive import MassivePriceBarSource +from hodlin_recommend.delivery.poller import UpdatePoller, latest_anomaly_reply +from hodlin_recommend.delivery.telegram import TelegramClient from hodlin_recommend.domain.explanation import AnthropicExplainer from hodlin_recommend.domain.sentiment import FinBertModel from hodlin_recommend.ingest.scheduler import build_scheduler @@ -49,6 +51,18 @@ def main() -> None: rate=RateLimiter(settings.massive_rate_per_min), ) + telegram = TelegramClient( + client, + token=settings.telegram_bot_token, + base_url=settings.telegram_base_url, + rate=RateLimiter(settings.telegram_rate_per_min), + ) + poller = UpdatePoller( + telegram, + allowed_chat_id=settings.telegram_chat_id, + reply_text=latest_anomaly_reply(session_factory), + ) + sentiment_model = FinBertModel() llm = AnthropicExplainer(api_key=settings.anthropic_api_key, model=settings.anthropic_model) inference_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="inference") @@ -59,6 +73,8 @@ def main() -> None: news_source=news_source, llm=llm, sentiment_model=sentiment_model, + messenger=telegram, + chat_id=settings.telegram_chat_id, inference_executor=inference_executor, ) @@ -70,6 +86,7 @@ def main() -> None: sentiment_model=sentiment_model, inference_executor=inference_executor, scheduler=scheduler, + poller=poller, session_factory=session_factory, resources=resources, ) diff --git a/packages/recommend/src/hodlin_recommend/serving/app.py b/packages/recommend/src/hodlin_recommend/serving/app.py index be0eb79..5ea20d8 100644 --- a/packages/recommend/src/hodlin_recommend/serving/app.py +++ b/packages/recommend/src/hodlin_recommend/serving/app.py @@ -24,6 +24,7 @@ """ import asyncio +import contextlib from collections.abc import AsyncIterator from concurrent.futures import ThreadPoolExecutor from contextlib import AsyncExitStack, asynccontextmanager @@ -53,11 +54,20 @@ def start(self) -> None: ... def shutdown(self, wait: bool = True) -> None: ... +@runtime_checkable +class PollerLike(Protocol): + """A run-forever loop the lifespan owns as a background task; cancelling + the task is the stop signal. Satisfied by ``UpdatePoller``.""" + + async def run(self) -> None: ... + + def create_app( *, sentiment_model: SentimentModel, inference_executor: ThreadPoolExecutor | None = None, scheduler: SchedulerLike | None = None, + poller: PollerLike | None = None, session_factory: SessionFactory | None = None, resources: AsyncExitStack | None = None, ) -> FastAPI: @@ -67,11 +77,20 @@ def create_app( @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: + poll_task: asyncio.Task[None] | None = None try: if scheduler is not None: scheduler.start() + if poller is not None: + poll_task = asyncio.create_task(poller.run(), name="telegram-poller") yield finally: + if poll_task is not None: + # Cancellation is the poller's stop signal; await the unwind + # so no inbound handler is mid-flight when resources close. + poll_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await poll_task if scheduler is not None: # AsyncIOScheduler defers the actual stop to a loop callback # and cancels (not awaits) in-flight ticks: yield once so the diff --git a/packages/recommend/src/hodlin_recommend/store/repositories.py b/packages/recommend/src/hodlin_recommend/store/repositories.py index 19a2b16..3a57034 100644 --- a/packages/recommend/src/hodlin_recommend/store/repositories.py +++ b/packages/recommend/src/hodlin_recommend/store/repositories.py @@ -11,7 +11,7 @@ from datetime import UTC, datetime from hodlin_contracts import EvidenceRef -from sqlalchemy import select +from sqlalchemy import select, update from sqlalchemy.dialects.postgresql import insert from sqlalchemy.ext.asyncio import AsyncSession @@ -189,6 +189,31 @@ async def recent_for_symbol( ] +def _anomaly_with_explanation( + row: tables.Anomaly, explanation_row: tables.Explanation, symbol: str +) -> tuple[Anomaly, Explanation]: + """Map one joined (anomaly, explanation) row pair to domain models — + shared by the delivery-queue and latest-explained reads.""" + anomaly = Anomaly( + symbol=symbol, + interval=row.interval, + bar_ts=row.bar_ts, + z_score=row.z_score, + return_pct=row.return_pct, + direction=row.direction, + window=row.window, + ) + explanation = Explanation( + symbol=symbol, + interval=row.interval, + bar_ts=row.bar_ts, + reasoning=explanation_row.reasoning, + evidence=tuple(EvidenceRef.model_validate(ref) for ref in explanation_row.evidence), + model_version=explanation_row.model_version, + ) + return anomaly, explanation + + class AnomalyRepository: def __init__(self, session: AsyncSession) -> None: self._session = session @@ -243,6 +268,58 @@ async def for_symbol(self, symbol: str, interval: str) -> list[Anomaly]: for row in rows ] + async def explained_unnotified(self, *, limit: int) -> list[tuple[Anomaly, Explanation]]: + """The delivery queue: anomalies that have their "why" but haven't been + sent yet, oldest first so alerts arrive in chronological order.""" + stmt = ( + select(tables.Anomaly, tables.Explanation, tables.Asset.symbol) + .join(tables.Asset, tables.Anomaly.asset_id == tables.Asset.id) + .join(tables.Explanation, tables.Explanation.anomaly_id == tables.Anomaly.id) + .where(tables.Anomaly.notified_at.is_(None)) + .order_by(tables.Anomaly.bar_ts, tables.Anomaly.id) + .limit(limit) + ) + rows = (await self._session.execute(stmt)).all() + return [ + _anomaly_with_explanation(row, explanation, symbol) for row, explanation, symbol in rows + ] + + async def latest_explained(self) -> tuple[Anomaly, Explanation] | None: + """The newest anomaly that has an explanation — the poller's reply to + an allowlisted 'what's up?', notified or not.""" + stmt = ( + select(tables.Anomaly, tables.Explanation, tables.Asset.symbol) + .join(tables.Asset, tables.Anomaly.asset_id == tables.Asset.id) + .join(tables.Explanation, tables.Explanation.anomaly_id == tables.Anomaly.id) + .order_by(tables.Anomaly.bar_ts.desc(), tables.Anomaly.id.desc()) + .limit(1) + ) + first = (await self._session.execute(stmt)).first() + if first is None: + return None + row, explanation, symbol = first + return _anomaly_with_explanation(row, explanation, symbol) + + async def mark_notified(self, symbol: str, interval: str, bar_ts: datetime) -> bool: + """Atomically claim one anomaly for delivery: flip ``notified_at`` only + if it is still NULL. Returns whether *this* caller won the claim — the + compare-and-set that makes "each anomaly notifies once" hold even + across overlapping processes, not just ticks.""" + asset_id = select(tables.Asset.id).where(tables.Asset.symbol == symbol).scalar_subquery() + stmt = ( + update(tables.Anomaly) + .where( + tables.Anomaly.asset_id == asset_id, + tables.Anomaly.interval == interval, + tables.Anomaly.bar_ts == bar_ts, + tables.Anomaly.notified_at.is_(None), + ) + .values(notified_at=datetime.now(UTC)) + .returning(tables.Anomaly.id) + ) + claimed = (await self._session.scalars(stmt)).all() + return len(claimed) == 1 + async def unexplained(self, *, limit: int) -> list[Anomaly]: """The newest ``limit`` anomalies with no explanation yet — the explain job's work queue. Newest first because fresh moves are what delivery diff --git a/tests/integration/test_jobs.py b/tests/integration/test_jobs.py index 87bca22..f3d882a 100644 --- a/tests/integration/test_jobs.py +++ b/tests/integration/test_jobs.py @@ -251,12 +251,19 @@ async def test_scheduled_run_lands_in_ingest_runs_and_app_reports_ready( engine: AsyncEngine, ) -> None: factory = create_session_factory(engine) + + class QuietMessenger: + async def send(self, chat_id: int, text: str) -> None: + return None + scheduler = build_scheduler( session_factory=factory, bar_source=FakeBarSource(_bars()), news_source=FakeNewsSource(), llm=MockLLM('{"reasoning": "why", "evidence_indices": []}'), sentiment_model=FakeSentimentModel(), + messenger=QuietMessenger(), + chat_id=42, assets=[_ASSET], ) app = create_app( diff --git a/tests/integration/test_notify.py b/tests/integration/test_notify.py new file mode 100644 index 0000000..2d1404c --- /dev/null +++ b/tests/integration/test_notify.py @@ -0,0 +1,123 @@ +"""The T9 acceptance against a real Postgres: the allowlisted chat receives a +formatted anomaly+why message, each anomaly notifies exactly once (claim -> +send -> commit on ``anomalies.notified_at``), a failed send releases the +claim for the next tick, and an anomaly without its "why" is never sent. +""" + +from datetime import UTC, datetime +from decimal import Decimal + +from hodlin_contracts import EvidenceRef +from hodlin_recommend.connectors.base import SourceUnavailable +from hodlin_recommend.domain.models import Anomaly, Asset, Explanation +from hodlin_recommend.ingest import jobs +from hodlin_recommend.store import tables +from hodlin_recommend.store.db import SessionFactory, create_session_factory +from hodlin_recommend.store.repositories import ( + AnomalyRepository, + AssetRepository, + ExplanationRepository, +) +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncEngine + +_CHAT_ID = 42 +_BAR_TS = datetime(2024, 6, 24, tzinfo=UTC) + + +def _anomaly(bar_ts: datetime = _BAR_TS) -> Anomaly: + return Anomaly( + symbol="BTC-USD", + interval="1d", + bar_ts=bar_ts, + z_score=Decimal("-2.8"), + return_pct=Decimal("-4.5271"), + direction="down", + window=15, + ) + + +def _explanation(bar_ts: datetime = _BAR_TS) -> Explanation: + return Explanation( + symbol="BTC-USD", + interval="1d", + bar_ts=bar_ts, + reasoning="Likely the exchange hack.", + evidence=( + EvidenceRef( + kind="anomaly", source="hodlin/anomaly", ref="BTC-USD/1d/x", observed_at=bar_ts + ), + ), + model_version="mock:1", + ) + + +class FakeMessenger: + def __init__(self, *, dead: bool = False) -> None: + self.dead = dead + self.sent: list[tuple[int, str]] = [] + + async def send(self, chat_id: int, text: str) -> None: + if self.dead: + raise SourceUnavailable("telegram", "api overloaded") + self.sent.append((chat_id, text)) + + +async def _seed(factory: SessionFactory, *, explained: bool = True) -> None: + async with factory() as session: + await AssetRepository(session).upsert(Asset(symbol="BTC-USD", kind="crypto")) + await AnomalyRepository(session).upsert_many([_anomaly()]) + if explained: + await ExplanationRepository(session).upsert(_explanation()) + await session.commit() + + +async def test_allowlisted_chat_gets_anomaly_plus_why_exactly_once(engine: AsyncEngine) -> None: + factory = create_session_factory(engine) + await _seed(factory) + messenger = FakeMessenger() + + first = await jobs.notify_anomalies(factory, messenger=messenger, chat_id=_CHAT_ID) + second = await jobs.notify_anomalies(factory, messenger=messenger, chat_id=_CHAT_ID) + + assert (first.status, first.items) == ("ok", 1) + assert (second.status, second.items) == ("ok", 0) + assert len(messenger.sent) == 1 # once means once, across ticks + chat_id, text = messenger.sent[0] + assert chat_id == _CHAT_ID + assert "BTC-USD" in text + assert "-4.5271%" in text + assert "Likely the exchange hack." in text # the why rides along + async with factory() as session: + (run, _) = sorted( + (await session.scalars(select(tables.IngestRun))).all(), key=lambda r: r.id + ) + assert (run.job, run.status, run.items) == ("notify_anomalies", "ok", 1) + + +async def test_failed_send_releases_the_claim_for_the_next_tick(engine: AsyncEngine) -> None: + factory = create_session_factory(engine) + await _seed(factory) + + down = await jobs.notify_anomalies( + factory, messenger=FakeMessenger(dead=True), chat_id=_CHAT_ID + ) + assert down.status == "error" + assert down.items == 0 + + recovered = FakeMessenger() + retry = await jobs.notify_anomalies(factory, messenger=recovered, chat_id=_CHAT_ID) + + assert (retry.status, retry.items) == ("ok", 1) # the claim was rolled back + assert len(recovered.sent) == 1 + + +async def test_unexplained_anomaly_is_never_sent(engine: AsyncEngine) -> None: + factory = create_session_factory(engine) + await _seed(factory, explained=False) + messenger = FakeMessenger() + + outcome = await jobs.notify_anomalies(factory, messenger=messenger, chat_id=_CHAT_ID) + + assert (outcome.status, outcome.items) == ("ok", 0) + assert messenger.sent == [] # an alert without its why is not an alert diff --git a/tests/test_delivery.py b/tests/test_delivery.py new file mode 100644 index 0000000..02fd1dd --- /dev/null +++ b/tests/test_delivery.py @@ -0,0 +1,230 @@ +"""Delivery tests — all offline (respx mocks the Bot API; no token, no network). + +Covers the T9 pieces that are pure or mockable without Postgres: formatting +escapes untrusted text, the thin client speaks the Bot API and degrades to +``SourceUnavailable``, and the poller enforces the single-ID allowlist — +a stranger's message produces no reply at all, while the allowlisted user +gets the injected reply text. The notify-once flow runs against real Postgres +in ``tests/integration/test_notify.py``. +""" + +import asyncio +import json +from datetime import UTC, datetime +from decimal import Decimal +from typing import Any + +import httpx +import pytest +import respx +from hodlin_contracts import EvidenceRef +from hodlin_recommend.connectors.base import RateLimiter, RetryPolicy, SourceUnavailable +from hodlin_recommend.delivery.formatting import format_anomaly, format_status +from hodlin_recommend.delivery.poller import UpdatePoller +from hodlin_recommend.delivery.telegram import Messenger, TelegramClient, UpdateSource +from hodlin_recommend.domain.models import Anomaly, Explanation + +_BASE = "https://api.test" +_SEND_URL = f"{_BASE}/botTOKEN/sendMessage" +_FAST = RetryPolicy(attempts=2, wait_initial=0.0, wait_max=0.0) + +_BAR_TS = datetime(2024, 6, 24, tzinfo=UTC) + +_ANOMALY = Anomaly( + symbol="BTC-USD", + interval="1d", + bar_ts=_BAR_TS, + z_score=Decimal("-2.801498"), + return_pct=Decimal("-4.5271"), + direction="down", + window=15, +) + + +def _explanation(reasoning: str = "Likely the exchange hack.") -> Explanation: + return Explanation( + symbol="BTC-USD", + interval="1d", + bar_ts=_BAR_TS, + reasoning=reasoning, + evidence=( + EvidenceRef( + kind="anomaly", + source="hodlin/anomaly", + ref="BTC-USD/1d/x", + observed_at=_BAR_TS, + ), + EvidenceRef( + kind="news", + source="finnhub", + ref="https://example.test/hack", + observed_at=_BAR_TS, + ), + ), + model_version="anthropic:claude-x", + ) + + +@pytest.fixture +def rate() -> RateLimiter: + return RateLimiter(10_000) + + +# Formatting — the escaping boundary ----------------------------------------- + + +def test_format_carries_the_numbers_and_the_why() -> None: + text = format_anomaly(_ANOMALY, _explanation()) + + assert "BTC-USD" in text + assert "-4.5271%" in text + assert "z-score -2.801498" in text + assert "Likely the exchange hack." in text + assert "1 news source(s) cited" in text # the anomaly self-ref doesn't count + + +def test_format_escapes_untrusted_reasoning() -> None: + # LLM prose over hostile headlines must never become Telegram markup. + hostile = 'See this & act now' + text = format_anomaly(_ANOMALY, _explanation(reasoning=hostile)) + + assert "BTC-USD" in text # our own markup stays literal markup + + +def test_status_is_honest_when_nothing_explained_yet() -> None: + assert "No explained anomalies yet" in format_status(None) + assert "Latest anomaly:" in format_status((_ANOMALY, _explanation())) + + +# The thin client ------------------------------------------------------------ + + +def _client(client: httpx.AsyncClient, rate: RateLimiter) -> TelegramClient: + return TelegramClient(client, token="TOKEN", base_url=_BASE, rate=rate, retry=_FAST) + + +async def test_send_posts_html_message(rate: RateLimiter) -> None: + async with httpx.AsyncClient() as http, respx.mock: + route = respx.post(_SEND_URL).mock( + return_value=httpx.Response(200, json={"ok": True, "result": {"message_id": 1}}) + ) + await _client(http, rate).send(42, "hi") + + assert route.call_count == 1 + payload = json.loads(route.calls.last.request.content) + assert payload["chat_id"] == 42 + assert payload["text"] == "hi" + assert payload["parse_mode"] == "HTML" + + +async def test_ok_false_is_a_failure_even_on_http_200(rate: RateLimiter) -> None: + async with httpx.AsyncClient() as http, respx.mock: + respx.post(_SEND_URL).mock( + return_value=httpx.Response(200, json={"ok": False, "description": "chat not found"}) + ) + with pytest.raises(SourceUnavailable) as excinfo: + await _client(http, rate).send(42, "hi") + assert excinfo.value.source == "telegram" + + +async def test_5xx_retries_then_raises_unavailable(rate: RateLimiter) -> None: + async with httpx.AsyncClient() as http, respx.mock: + route = respx.post(_SEND_URL).mock(return_value=httpx.Response(502)) + with pytest.raises(SourceUnavailable): + await _client(http, rate).send(42, "hi") + assert route.call_count == _FAST.attempts + + +async def test_get_updates_unwraps_result(rate: RateLimiter) -> None: + updates = [{"update_id": 7, "message": {"text": "hi"}}] + async with httpx.AsyncClient() as http, respx.mock: + route = respx.post(f"{_BASE}/botTOKEN/getUpdates").mock( + return_value=httpx.Response(200, json={"ok": True, "result": updates}) + ) + got = await _client(http, rate).get_updates(offset=None) + + assert got == updates + payload = json.loads(route.calls.last.request.content) + assert "offset" not in payload # first poll: take whatever is pending + assert payload["timeout"] > 0 # long poll, not a busy loop + + +def test_client_satisfies_both_protocol_halves(rate: RateLimiter) -> None: + client = TelegramClient(httpx.AsyncClient(), token="T", base_url=_BASE, rate=rate) + assert isinstance(client, Messenger) + assert isinstance(client, UpdateSource) + + +# The poller: single-ID allowlist -------------------------------------------- + +_ALLOWED = 42 +_STRANGER = 666 + + +def _update(update_id: int, sender: int, chat: int | None = None) -> dict[str, Any]: + return { + "update_id": update_id, + "message": { + "from": {"id": sender}, + "chat": {"id": chat if chat is not None else sender}, + "text": "what's up?", + }, + } + + +class ScriptedAPI: + """Serves one scripted batch of updates, then long-polls forever (until + the test cancels the poller) — deterministic, no timing guesses.""" + + def __init__(self, updates: list[dict[str, Any]]) -> None: + self.batches = [updates] + self.sent: list[tuple[int, str]] = [] + self.offsets: list[int | None] = [] + self.drained = asyncio.Event() + + async def get_updates(self, offset: int | None) -> list[dict[str, Any]]: + self.offsets.append(offset) + if self.batches: + return self.batches.pop(0) + self.drained.set() + await asyncio.Event().wait() # park forever; cancellation ends the test + raise AssertionError("unreachable") + + async def send(self, chat_id: int, text: str) -> None: + self.sent.append((chat_id, text)) + + +async def _run_until_drained(api: ScriptedAPI) -> None: + async def reply_text() -> str: + return "latest anomaly summary" + + poller = UpdatePoller(api, allowed_chat_id=_ALLOWED, reply_text=reply_text) + task = asyncio.create_task(poller.run()) + try: + await asyncio.wait_for(api.drained.wait(), timeout=5) + finally: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +async def test_allowlisted_user_gets_the_reply_and_strangers_get_silence() -> None: + api = ScriptedAPI([_update(1, _STRANGER), _update(2, _ALLOWED)]) + + await _run_until_drained(api) + + assert api.sent == [(_ALLOWED, "latest anomaly summary")] # one reply, to us only + assert api.offsets == [None, 3] # both updates acknowledged, stranger included + + +async def test_stranger_in_allowed_chat_is_still_rejected() -> None: + # Sender and chat must *both* match: a group scenario where the chat id + # looks right but the sender doesn't stays rejected. + api = ScriptedAPI([_update(1, sender=_STRANGER, chat=_ALLOWED)]) + + await _run_until_drained(api) + + assert api.sent == [] diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 0418c9f..be7be63 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -1,7 +1,7 @@ """Scheduler wiring tests — offline, no Postgres, no real ticks on real time. What's asserted is *configuration and lifecycle*, the parts that are ours: -the four recurring jobs exist with overlap protection (``max_instances=1``), +the five recurring jobs exist with overlap protection (``max_instances=1``), coalescing, and a misfire grace; backfill registers as a one-shot; the app lifespan starts the scheduler, a job actually fires on the loop, and shutdown leaves nothing running. Firing on a *schedule* against a real database is the @@ -24,6 +24,7 @@ EXPLAIN_EVERY_S, MISFIRE_GRACE_S, NEWS_EVERY_S, + NOTIFY_EVERY_S, build_scheduler, ) from hodlin_recommend.serving.app import SchedulerLike, create_app @@ -66,6 +67,11 @@ async def complete(self, *, system: str, user: str) -> str: return '{"reasoning": "why", "evidence_indices": []}' +class FakeMessenger: + async def send(self, chat_id: int, text: str) -> None: + return None + + def _build(*, backfill_on_start: bool = True) -> AsyncIOScheduler: # A real factory over an engine that never connects — jobs never run here. factory = create_session_factory(create_engine("postgresql+asyncpg://x:x@localhost/x")) @@ -75,6 +81,8 @@ def _build(*, backfill_on_start: bool = True) -> AsyncIOScheduler: news_source=FakeNewsSource(), llm=MockLLM(), sentiment_model=FakeSentimentModel(), + messenger=FakeMessenger(), + chat_id=42, backfill_on_start=backfill_on_start, ) @@ -84,10 +92,11 @@ def _build(*, backfill_on_start: bool = True) -> AsyncIOScheduler: "ingest_news": NEWS_EVERY_S, "detect_anomalies": DETECT_EVERY_S, "explain_anomalies": EXPLAIN_EVERY_S, + "notify_anomalies": NOTIFY_EVERY_S, } -async def test_four_recurring_jobs_with_overlap_protection() -> None: +async def test_recurring_jobs_registered_with_overlap_protection() -> None: scheduler = _build() scheduler.start(paused=True) # materializes pending jobs without firing any try: From e290a491266e702e106d93d569b6a8d6ccd37794 Mon Sep 17 00:00:00 2001 From: Vladislav Poverin <123168793+vlobus@users.noreply.github.com> Date: Tue, 14 Jul 2026 22:35:25 +0200 Subject: [PATCH 2/2] t9 telegram: review fixes - redact secrets from wrapped http errors (token was landing in ingest_runs.detail), poller survives any exception + acks poisoned updates, shutdown tolerates a crashed poll task --- .../src/hodlin_recommend/connectors/base.py | 15 ++++- .../hodlin_recommend/connectors/finnhub.py | 1 + .../hodlin_recommend/connectors/massive.py | 1 + .../src/hodlin_recommend/delivery/poller.py | 31 +++++++---- .../src/hodlin_recommend/delivery/telegram.py | 6 +- .../src/hodlin_recommend/serving/app.py | 4 +- tests/test_delivery.py | 55 +++++++++++++++---- 7 files changed, 87 insertions(+), 26 deletions(-) diff --git a/packages/recommend/src/hodlin_recommend/connectors/base.py b/packages/recommend/src/hodlin_recommend/connectors/base.py index 3751af2..7e466c1 100644 --- a/packages/recommend/src/hodlin_recommend/connectors/base.py +++ b/packages/recommend/src/hodlin_recommend/connectors/base.py @@ -121,6 +121,7 @@ async def request_json( http_timeout: httpx.Timeout | None = None, rate: RateLimiter | None = None, retry: RetryPolicy = DEFAULT_RETRY, + secrets: tuple[str, ...] = (), ) -> Any: """Call ``url`` and return parsed JSON, applying the shared rate-limit + retry policy and wrapping any HTTP failure as ``SourceUnavailable``. @@ -130,6 +131,11 @@ async def request_json( must outlive it). Note retries make non-idempotent POSTs at-least-once: a reply lost on the wire is retried even if the server acted — callers choose semantics. + + ``secrets`` are redacted from the wrapped error message: httpx errors + quote the full URL, which carries API keys (query params) or the bot + token (path) — and ``SourceUnavailable`` text ends up *persisted* in + ``ingest_runs.detail``, so the secret must be scrubbed at the wrap. """ request_timeout = http_timeout if http_timeout is not None else httpx.USE_CLIENT_DEFAULT @@ -158,6 +164,13 @@ async def _once() -> Any: with attempt: return await _once() except (httpx.HTTPError, json.JSONDecodeError) as exc: - raise SourceUnavailable(source, exc) from exc + raise SourceUnavailable(source, _redact(str(exc), secrets)) from exc # Unreachable: the loop either returns or reraises, but satisfies the type. raise SourceUnavailable(source, "retries exhausted") + + +def _redact(text: str, secrets: tuple[str, ...]) -> str: + for secret in secrets: + if secret: + text = text.replace(secret, "***") + return text diff --git a/packages/recommend/src/hodlin_recommend/connectors/finnhub.py b/packages/recommend/src/hodlin_recommend/connectors/finnhub.py index a6acbca..0a472a9 100644 --- a/packages/recommend/src/hodlin_recommend/connectors/finnhub.py +++ b/packages/recommend/src/hodlin_recommend/connectors/finnhub.py @@ -68,6 +68,7 @@ async def get_news(self, symbol: str, since: datetime) -> list[NewsItem]: params=params, rate=self._rate, retry=self._retry, + secrets=(self._api_key,), # httpx errors quote the URL, key included ) if not isinstance(payload, list): raise SourceUnavailable(self.source, "expected a JSON array of articles") diff --git a/packages/recommend/src/hodlin_recommend/connectors/massive.py b/packages/recommend/src/hodlin_recommend/connectors/massive.py index 0030fe5..7530b2d 100644 --- a/packages/recommend/src/hodlin_recommend/connectors/massive.py +++ b/packages/recommend/src/hodlin_recommend/connectors/massive.py @@ -85,6 +85,7 @@ async def get_candles( params=params, rate=self._rate, retry=self._retry, + secrets=(self._api_key,), # httpx errors quote the URL, key included ) bars = payload.get("bars") if isinstance(payload, Mapping) else None if not isinstance(bars, list): diff --git a/packages/recommend/src/hodlin_recommend/delivery/poller.py b/packages/recommend/src/hodlin_recommend/delivery/poller.py index 2143f0a..096e2da 100644 --- a/packages/recommend/src/hodlin_recommend/delivery/poller.py +++ b/packages/recommend/src/hodlin_recommend/delivery/poller.py @@ -8,15 +8,16 @@ poller itself never touches the database and unit tests never need one. Runs as a background task owned by the app lifespan; cancellation is the stop -signal. A dead Telegram API is absorbed with a backoff, same policy as any -dead source: the loop must outlive the outage. +signal — and the *only* thing that stops it. Any other failure (Telegram +down, a DB blip while building the reply, a malformed update) is absorbed +with a backoff or a skip: a background task that dies silently is a bot that +looks alive and answers no one until the next restart. """ import asyncio from collections.abc import Awaitable, Callable from typing import Any, Protocol -from hodlin_recommend.connectors.base import SourceUnavailable from hodlin_recommend.delivery.formatting import format_status from hodlin_recommend.delivery.telegram import Messenger, UpdateSource from hodlin_recommend.store.db import SessionFactory @@ -45,17 +46,28 @@ def __init__( async def run(self) -> None: """Poll forever; the owner cancels this task to stop it. ``offset`` - acknowledges processed updates so Telegram never redelivers them.""" + acknowledges processed updates so Telegram never redelivers them. + + ``except Exception`` (never ``CancelledError``) is deliberate at both + levels: a failed poll backs off and retries; a poisoned update is + acknowledged and skipped rather than redelivered into the same crash + forever. Replies are best-effort — the user can just ask again. + """ offset: int | None = None while True: try: updates = await self._api.get_updates(offset) - except SourceUnavailable: + except Exception: # the loop must outlive any outage await asyncio.sleep(ERROR_BACKOFF_S) continue for update in updates: - offset = int(update["update_id"]) + 1 - await self._handle(update) + update_id = update.get("update_id") + if isinstance(update_id, int): + offset = update_id + 1 # acknowledge first: never re-crash on it + try: + await self._handle(update) + except Exception: # one bad update must not kill the loop + continue async def _handle(self, update: dict[str, Any]) -> None: message = update.get("message") or {} @@ -63,10 +75,7 @@ async def _handle(self, update: dict[str, Any]) -> None: sender_id = (message.get("from") or {}).get("id") if chat_id != self._allowed or sender_id != self._allowed: return # default-deny: no reply, no error, no acknowledgement - try: - await self._api.send(self._allowed, await self._reply_text()) - except SourceUnavailable: - return # the reply is best-effort; the user can just ask again + await self._api.send(self._allowed, await self._reply_text()) def latest_anomaly_reply(session_factory: SessionFactory) -> Callable[[], Awaitable[str]]: diff --git a/packages/recommend/src/hodlin_recommend/delivery/telegram.py b/packages/recommend/src/hodlin_recommend/delivery/telegram.py index fe69a93..5e0b56a 100644 --- a/packages/recommend/src/hodlin_recommend/delivery/telegram.py +++ b/packages/recommend/src/hodlin_recommend/delivery/telegram.py @@ -61,7 +61,10 @@ def __init__( retry: RetryPolicy = DEFAULT_RETRY, ) -> None: self._client = client - # Telegram puts the secret in the path; it must never be logged. + # Telegram puts the secret in the URL *path*, and httpx error text + # quotes the URL — so the token rides every wrapped error unless + # request_json redacts it (its text lands in ingest_runs.detail). + self._token = token self._base = f"{base_url.rstrip('/')}/bot{token}" self._rate = rate self._retry = retry @@ -82,6 +85,7 @@ async def _call( http_timeout=http_timeout, rate=self._rate, retry=self._retry, + secrets=(self._token,), ) # Telegram can answer HTTP 200 with ok=false; that's still a failure. if not isinstance(data, dict) or not data.get("ok"): diff --git a/packages/recommend/src/hodlin_recommend/serving/app.py b/packages/recommend/src/hodlin_recommend/serving/app.py index 5ea20d8..86bd8d3 100644 --- a/packages/recommend/src/hodlin_recommend/serving/app.py +++ b/packages/recommend/src/hodlin_recommend/serving/app.py @@ -88,8 +88,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: if poll_task is not None: # Cancellation is the poller's stop signal; await the unwind # so no inbound handler is mid-flight when resources close. + # Suppress Exception too: a task that somehow crashed earlier + # re-raises here, and it must not abort the rest of teardown. poll_task.cancel() - with contextlib.suppress(asyncio.CancelledError): + with contextlib.suppress(asyncio.CancelledError, Exception): await poll_task if scheduler is not None: # AsyncIOScheduler defers the actual stop to a loop callback diff --git a/tests/test_delivery.py b/tests/test_delivery.py index 02fd1dd..f82fb47 100644 --- a/tests/test_delivery.py +++ b/tests/test_delivery.py @@ -10,6 +10,7 @@ import asyncio import json +from collections.abc import Awaitable, Callable from datetime import UTC, datetime from decimal import Decimal from typing import Any @@ -130,12 +131,19 @@ async def test_ok_false_is_a_failure_even_on_http_200(rate: RateLimiter) -> None assert excinfo.value.source == "telegram" -async def test_5xx_retries_then_raises_unavailable(rate: RateLimiter) -> None: +async def test_5xx_retries_then_raises_unavailable_with_token_redacted( + rate: RateLimiter, +) -> None: async with httpx.AsyncClient() as http, respx.mock: route = respx.post(_SEND_URL).mock(return_value=httpx.Response(502)) - with pytest.raises(SourceUnavailable): + with pytest.raises(SourceUnavailable) as excinfo: await _client(http, rate).send(42, "hi") assert route.call_count == _FAST.attempts + # httpx quotes the full URL (token in the path) in its error text, and + # SourceUnavailable text gets persisted to ingest_runs.detail — so the + # secret must be scrubbed at the wrap. + assert "TOKEN" not in str(excinfo.value) + assert "***" in str(excinfo.value) async def test_get_updates_unwraps_result(rate: RateLimiter) -> None: @@ -152,10 +160,11 @@ async def test_get_updates_unwraps_result(rate: RateLimiter) -> None: assert payload["timeout"] > 0 # long poll, not a busy loop -def test_client_satisfies_both_protocol_halves(rate: RateLimiter) -> None: - client = TelegramClient(httpx.AsyncClient(), token="T", base_url=_BASE, rate=rate) - assert isinstance(client, Messenger) - assert isinstance(client, UpdateSource) +async def test_client_satisfies_both_protocol_halves(rate: RateLimiter) -> None: + async with httpx.AsyncClient() as http: + client = TelegramClient(http, token="T", base_url=_BASE, rate=rate) + assert isinstance(client, Messenger) + assert isinstance(client, UpdateSource) # The poller: single-ID allowlist -------------------------------------------- @@ -176,11 +185,11 @@ def _update(update_id: int, sender: int, chat: int | None = None) -> dict[str, A class ScriptedAPI: - """Serves one scripted batch of updates, then long-polls forever (until + """Serves scripted batches of updates, then long-polls forever (until the test cancels the poller) — deterministic, no timing guesses.""" - def __init__(self, updates: list[dict[str, Any]]) -> None: - self.batches = [updates] + def __init__(self, *batches: list[dict[str, Any]]) -> None: + self.batches = list(batches) self.sent: list[tuple[int, str]] = [] self.offsets: list[int | None] = [] self.drained = asyncio.Event() @@ -197,11 +206,13 @@ async def send(self, chat_id: int, text: str) -> None: self.sent.append((chat_id, text)) -async def _run_until_drained(api: ScriptedAPI) -> None: - async def reply_text() -> str: +async def _run_until_drained( + api: ScriptedAPI, reply_text: Callable[[], Awaitable[str]] | None = None +) -> None: + async def default_reply() -> str: return "latest anomaly summary" - poller = UpdatePoller(api, allowed_chat_id=_ALLOWED, reply_text=reply_text) + poller = UpdatePoller(api, allowed_chat_id=_ALLOWED, reply_text=reply_text or default_reply) task = asyncio.create_task(poller.run()) try: await asyncio.wait_for(api.drained.wait(), timeout=5) @@ -228,3 +239,23 @@ async def test_stranger_in_allowed_chat_is_still_rejected() -> None: await _run_until_drained(api) assert api.sent == [] + + +async def test_a_crashing_reply_does_not_kill_the_loop() -> None: + """A DB blip while building one reply must not silently kill the poller: + the poisoned update is acknowledged and skipped, and the next message + gets its answer.""" + api = ScriptedAPI([_update(1, _ALLOWED)], [_update(2, _ALLOWED)]) + calls = 0 + + async def flaky_reply() -> str: + nonlocal calls + calls += 1 + if calls == 1: + raise RuntimeError("db blip") + return "recovered" + + await _run_until_drained(api, flaky_reply) + + assert api.sent == [(_ALLOWED, "recovered")] # second message answered + assert api.offsets == [None, 2, 3] # the crashing update was still acked