Levandor

telegram_connector — Outbound Relay (v1) Design

4 min read

The pmv2 fleet’s single Telegram I/O plane. Captures the approved (2026-06-22) v1 design: a stateless, durable JetStream consumer that faithfully relays telegram.outbound messages to the Telegram Bot API. Full spec lives at /Users/levander/coding/pmv2/telegram_connector/docs/superpowers/specs/2026-06-22-telegram-connector-outbound-design.md.

For Agents

telegram_connector is the sole bot-token holder and sole Telegram sender for the entire pmv2 fleet — the single rate-limit authority. It holds no trading logic, no DB, and is stateless in v1. Other fleet services (cohort_algo, position_manager, weather, crypto) never send to Telegram directly; they publish to the telegram.outbound NATS subject and this service relays. Repo: /Users/levander/coding/pmv2/telegram_connector (own git repo + cargo workspace).

Boundary & Role

  • Single-purpose: durably consume the telegram.outbound JetStream subject and relay each message to Bot API sendMessage.
  • Sole holder of the bot token; sole Telegram sender → single rate-limit authority for the fleet.
  • No trading logic, no other service’s DB, stateless v1 (no SUPABASE_DB_URL).
  • Dumb, faithful relay: passes text through verbatim with the publisher-supplied parse_mode. Escaping/formatting is the publisher’s job — the connector never re-escapes.

v1 Scope: OUTBOUND ONLY

In scope: durable JetStream pull consumer on subject telegram.outbound (stream TELEGRAM_OUTBOUND) → Telegram Bot API sendMessage.

Out of scope (deferred to a separate later spec): inbound (getUpdates → auth → ops.commands / ops.results relay). The inbound spec is gated on @positionmanager locking the ops.* contract (babylon #pmv2 #311). Also out of scope: proactive rate limiting, persistence/ops state, multiple bot tokens, non-sendMessage methods (photos, edits).

The Wire — telegram.outbound Envelope

Deserialize-only in this service. Matched field-for-field to the reference publisher at cohort_algo/crates/cohort_algo/src/telegram_outbound.rs until the shared pmv2-contracts standalone repo publishes (then vendor it).

TelegramOutbound {
  msg_id: String,             // idempotency key = Nats-Msg-Id; stream dedup within duplicate_window
  chat: Option<String>,       // None => default TELEGRAM_CHAT_ID
  text: String,               // already-formatted body — relayed verbatim
  parse_mode: Option<String>, // "HTML" | "MarkdownV2" | null
  kind: Option<String>,       // "info" | "alert" | "trade" — advisory only (log attribute)
}
  • Subject telegram.outbound → stream TELEGRAM_OUTBOUND (1d retention, 5-min duplicate_window), provisioned by @deploy.
  • Dedup is the stream’s job (keyed on Nats-Msg-Id = msg_id). The connector keeps no dedup state.

Delivery Semantics

At-least-once. Ack only AFTER a confirmed send (or a deliberate terminal drop).

OutcomeTriggersPolicy
SentHTTP 200, ok: trueack
Transient429 (+retry_after), 5xx, network/timeoutin-process retry (honor retry_after else backoff); if still failing → no ack → JetStream redelivers, bounded by max_deliver; on exhaustion → dead-letter + ERROR
Terminal400 (bad format / chat not found), 403 (bot blocked)drop + ack (avoid head-of-line block & burning redelivery budget) + ERROR log + best-effort meta-alert

Meta-alert loop guard

On a terminal drop, after structured-logging it, the connector sends one best-effort plain-text meta-alert to TELEGRAM_CHAT_ID (dropped msg_id=… error_code=… desc=…) so a swallowed alert stays visible. The meta-alert itself has no parse_mode, and a failed meta-alert is only logged, never meta-alerted about. Config-toggle META_ALERT_ON_DROP (default on).

Concurrency & Rate Limiting

  • Concurrency (v1): sequential — one message in-flight at a time. Low volume, preserves order, trivially respects per-chat limits. Concurrent workers deferred (YAGNI).
  • Rate limiting (v1): reactive only — on 429, honor retry_after. No proactive token bucket until volume proves it necessary (YAGNI).

Testability Decision

The relay loop sits behind a MessageSource trait so it is unit-testable with a fake (no broker required):

  • Unit: outbound (de)serialization parity (golden), telegram error classification from sample Bot API JSON, config dormant guard, relay policy against a fake MessageSource + fake telegram client.
  • Integration: wiremock the Bot API for the full relay; a separate optional test exercises real async-nats JetStream against a testcontainer (provisions its own TELEGRAM_OUTBOUND + tg_outbound).
  • Green gate: build + tests + clippy -D warnings + fmt + dormant-safe smoke (unset token → exit(0), no panic).

Module Decomposition

Single bin crate. config (typed env + dormant guard — exit 0 if TELEGRAM_BOT_TOKEN/NATS_URL unset), telemetry (OTLP to local collector), outbound (the envelope type), telegram (reqwest/rustls client, injectable base URL, error classification), consumer (JetStream nkey auth; bind-or-create own durable PULL consumer tg_outbound; does not create the stream), relay (core loop behind MessageSource), main (wire-up + graceful SIGINT/SIGTERM drain).

Conventions

edition 2024, resolver “3”, rust 1.85, unsafe_code = "forbid". Deps: tokio, async-nats (+nkeys), reqwest (rustls), serde/serde_json, anyhow, thiserror, tracing + tracing-opentelemetry + opentelemetry-otlp; dev: wiremock (+ optional testcontainers). No code comments. Mirrors sibling-repo workspace layout. See pmv2-fleet-infra-observability-nats-conventions for the fleet-wide OTel/logging/NATS rules this service follows.

Open Coordination (babylon #pmv2)

  • #310 — pmv2-contracts standalone repo confirmed (Andras). Vendor its envelope type when it publishes (swap the hand-rolled outbound type).
  • #311 — inbound ops.* contract with @positionmanager still open; gates the inbound spec, not this one.
  • Provisioning: TELEGRAM_OUTBOUND (1d, 5-min dedup) confirmed; outbound consumer is tg_outbound (connector-owned).

Graph View