telegram_connector — Fleet I/O Plane Integration

Status: Outbound relay LIVE (pin 3baa6e4). Inbound commands deployed + activated (round-trip goes live once position_manager’s ops consumer launches).

Repository: wowjeeez/telegram_connector (separate from polymarket-fetch). Canonical README at /Users/levander/coding/pmv2/telegram_connector/README.md.

Role in pmv2 Fleet

telegram_connector is the single Telegram I/O plane for all pmv2 trading services. It is:

Sole bot-token holder — no other service has the Telegram Bot API credential
Sole Telegram sender — all notifications route through it (via telegram.outbound JetStream subject)
Only reader of operator commands — inbound messages from Telegram → parsed here, not elsewhere
Completely decoupled from trading logic — pure I/O, no algo state, no position DB

Outbound Flow: Services → Operator

Pattern A - Broadcast Notification:

cohort_algo, position_manager, crypto_algo, etc.
    ↓ publish { msg_id, chat?, text, parse_mode?, kind? } to telegram.outbound
NATS JetStream TELEGRAM_OUTBOUND stream (1d, 5m dedup)
    ↓ durable pull consumer tg_outbound
telegram_connector
    ↓ classify response (Sent / Transient / Terminal)
Telegram Bot API sendMessage
    ↓
Operator's Telegram chat

Wire Contract: TelegramOutbound { msg_id: String, chat?: String, text: String, parse_mode?: String, kind?: String } from pmv2-contracts.

Delivery: At-least-once (ack-after-send). Transient errors (429, 5xx, network) → nak + backoff. Terminal errors (400, 403) → drop + loop-guarded meta-alert.

Key characteristic: Publisher owns message formatting; connector relays verbatim (no re-escaping).

Example: Autotrade Mode Change Alert

{
  "msg_id": "cohort|set_mode|live|timestamp",
  "chat": null,
  "text": "<b>Mode:</b> LIVE\n<b>Source:</b> polymarket\n<b>Issued by:</b> @operator_user",
  "parse_mode": "HTML",
  "kind": "trade"
}

Publishes to telegram.outbound with Nats-Msg-Id: cohort|set_mode|live|timestamp. NATS dedup window is 5 minutes — if the same msg_id is republished within 5m (e.g., retry), stream returns the original message (idempotent).

Inbound Flow: Operator → Services

Pattern B - Operator Command (Round-trip):

Operator sends /autotrade live
    ↓ getUpdates long-poll
telegram_connector
    ↓ authorize (role check) + validate + publish
ops.commands subject on OPS stream (7d, 5m dedup)
    ↓
position_manager (owns ops consumer)
    ↓ processes command, publishes ops.results
ops.results subject on OPS stream
    ↓
telegram_connector (pending-map correlation)
    ↓ relay result back to operator
Operator sees: "applied" or "needs_confirm" or "error"

Activation Gating

Inbound activates ONLY when TELEGRAM_OWNER_IDS is set in the connector’s env. This is additive — outbound always runs.

To activate inbound:

Add TELEGRAM_OWNER_IDS=123456,789012 (comma-sep Telegram user IDs)
Add SUPABASE_DB_URL=postgresql://... (for role perms, optional but recommended)
Re-pin telegram_connector in the fleet deploy

If not set, the connector runs outbound-only (unchanged behavior).

Commands (Code-Defined)

New commands requested via /COMMAND_REQUEST.md in telegram_connector repo. Current set:

Command	Effect	Min Role	Implementation
`/autotrade <mode> [source] [confirm]`	SetMode	Operator	position_manager or service-specific
`/halt`	Halt (= off)	Operator	position_manager or service-specific
`/status`	Status query	Viewer	position_manager or service-specific
`/enable <source>`	Enable source	Operator	position_manager or service-specific
`/disable <source>`	Disable source	Operator	position_manager or service-specific
`/reload-commands`	Refresh perms	Admin	telegram_connector (local)

Role hierarchy: Viewer < Operator < Admin. DB-backed in tg_operators table (read-only from connector; written out-of-band). Admin users from TELEGRAM_OWNER_IDS env are resolved in-memory before any DB lookup (kill-switch independence).

Wire Contracts (pmv2-contracts @ 4c7a54f)

Shared crate: wowjeeez/pmv2-contracts (private git). Vendored by telegram_connector.

TelegramOutbound (outbound)

pub struct TelegramOutbound {
    pub msg_id: String,
    pub chat: Option<String>,
    pub text: String,
    pub parse_mode: Option<String>,
    pub kind: Option<String>,
}

Subject: telegram.outbound. Stream: TELEGRAM_OUTBOUND (1d retention).

OpsCommand (inbound publish)

pub struct OpsCommand {
    pub schema_version: u32,
    pub command_id: String,      // reply key
    pub issued_at: i64,          // unix millis
    pub actor: String,           // telegram user_id as string
    pub command: Command,        // SetMode | Halt | Status | Enable | Disable
}
 
pub enum Command {
    #[serde(rename = "set_mode")]
    SetMode { mode: Mode, source: Option<String> },
    #[serde(rename = "halt")]
    Halt,
    #[serde(rename = "status")]
    Status,
    #[serde(rename = "enable")]
    Enable { source: String },
    #[serde(rename = "disable")]
    Disable { source: String },
}
 
pub enum Mode { Off, Dry, Live }

Subject: ops.commands. Stream: OPS (7d retention, 5m dedup).

OpsResult (inbound consume)

pub struct OpsResult {
    pub schema_version: u32,
    pub command_id: String,      // echoes OpsCommand.command_id
    pub status: OpsStatus,       // applied | rejected | needs_confirm | error
    pub state: Option<OpsState>, // PM-defined snapshot
    pub message: String,         // human-readable detail
}
 
pub enum OpsStatus { Applied, Rejected, NeedsConfirm, Error }

Subject: ops.results. Stream: OPS. Consumer: tg_ops_results (DeliverPolicy::New, skip backlog).

NATS Integration Points

Subject	Stream	Direction	Durable	Retention	Dedup	Who Creates
`telegram.outbound`	`TELEGRAM_OUTBOUND`	consume	`tg_outbound`	1d	5m	@deploy (stream), telegram_connector (consumer)
`ops.commands`	`OPS`	publish	—	7d	5m	telegram_connector (publishes)
`ops.results`	`OPS`	consume	`tg_ops_results`	7d	5m	telegram_connector (consumer)

NKey permissions for telegram_connector: SUB telegram.outbound, PUB ops.commands, SUB ops.results. Seed: NATS_TELEGRAM_NKEY_SEED.

Observability

Telemetry pipeline: telegram_connector → OTLP gRPC/HTTP → local OTel collector on polymarket-infra → SigNoz Cloud (never direct-to-SigNoz).

Service name: pmv2-telegram-connector. Traces: send_message, classify_response, poll, dispatch. Logs: ERROR (operator-visible failures), WARN (transients), INFO (state transitions <1/s), DEBUG (per-message).

Security: Bot token never in logs. reqwest errors stripped via without_url() (token lives in URL).

Fleet logging discipline: See pmv2-fleet-infra-observability-nats-conventions.

Configuration

Var	Required	Secret	Note
`TELEGRAM_BOT_TOKEN`	yes	yes	sole holder
`TELEGRAM_CHAT_ID`	yes	no	default chat
`NATS_URL`	yes	no	fleet NATS
`NATS_TELEGRAM_NKEY_SEED`	yes	yes	NATS auth
`TELEGRAM_OWNER_IDS`	inbound only	no	comma-sep i64; activates inbound
`SUPABASE_DB_URL`	inbound only	yes	role perms (optional; owners-only if missing)

Dormant-safe: missing TELEGRAM_BOT_TOKEN or NATS_URL → log + exit 0 (no panic).

Build & Deploy

Pattern A (cargo + systemd, ansible). Rust 2024, unsafe_code = "forbid". Vendoring pmv2-contracts requires CARGO_NET_GIT_FETCH_WITH_CLI + clone token.

34 passing tests (unit + wiremock integration). Clippy clean. Startup announces version hash + commit subject.

Coordination Hub

Fleet coordination runs on babylon hub pmv2. Key discussions:

#310: pmv2-contracts repo structure
#311–#395: ops.* contract co-design (position_manager + telegram_connector)
#377–#389: OpsCommand, OpsResult, schema versioning, two-step confirm for global live

Engineering workflow: superpowers brainstorm → spec → writing-plan → subagent-dev (git worktrees, FF merge) → whole-branch review before push to main.

Future Work

Telegram Bot API v7 webhooks: Replace long-poll with push (lower latency)
Batch sends: Consume N messages, send as thread/replies
Per-chat rate limits: Track per-chat instead of globally
Command extensions: /get-position, /market-info, etc.

telegram-connector-complete — master overview + final status
telegram-connector-outbound-design — relay ack/nak/term, rate limiting, resilience
telegram-connector-inbound-framework — auth, perms, restart-replay safety
telegram-connector-inbound-commands — command implementations
pmv2-fleet-infra-observability-nats-conventions — fleet NATS + OTel conventions
pmv2-position-manager-nats-consumer — ops consumer implementation (position_manager side)
polymarket-fetch — pmv2 trading fleet

Levandor

Explorer

pmv2 Fleet — telegram_connector Integration

telegram_connector — Fleet I/O Plane Integration

Role in pmv2 Fleet

Outbound Flow: Services → Operator

Example: Autotrade Mode Change Alert

Inbound Flow: Operator → Services

Activation Gating

Commands (Code-Defined)

Wire Contracts (pmv2-contracts @ 4c7a54f)

TelegramOutbound (outbound)

OpsCommand (inbound publish)

OpsResult (inbound consume)

NATS Integration Points

Observability

Configuration

Build & Deploy

Coordination Hub

Future Work

Graph View

Table of Contents

Backlinks

Levandor

Explorer

pmv2 Fleet — telegram_connector Integration

telegram_connector — Fleet I/O Plane Integration

Role in pmv2 Fleet

Outbound Flow: Services → Operator

Example: Autotrade Mode Change Alert

Inbound Flow: Operator → Services

Activation Gating

Commands (Code-Defined)

Wire Contracts (pmv2-contracts @ 4c7a54f)

TelegramOutbound (outbound)

OpsCommand (inbound publish)

OpsResult (inbound consume)

NATS Integration Points

Observability

Configuration

Build & Deploy

Coordination Hub

Future Work

Related

Graph View

Table of Contents

Backlinks