feat(notifiers): structured-log subscriber + Telegram bidirectional bot

LogNotifier subscribes (Drop overflow) and emits stdlib Log lines with
structured kwargs (credential_id, rotation_id, severity, etc.) suitable
for downstream ingestion by journald/vector/fluentd.

Telegram is a thin HTTP::Client wrapper for sendMessage and getUpdates.
TelegramSubscriber sends emoji-prefixed alerts on RotationFailed,
DriftDetected, PolicyViolation, AlertRaised; success notifications gated
by the notify_on_success flag. Errors are swallowed so a flaky bot never
blocks the engine.

TelegramBot does long-polling getUpdates and dispatches commands:
- viewer tier: /status /queue /history /alerts /help
- operator tier: viewer + /rotate + /snooze
ACL is by chat_id allowlist (bot token + chat IDs in env vars).
/rotate publishes RotationScheduled to the bus.

11 unit specs cover send_message + error handling + getUpdates parsing,
subscriber dispatch, success-suppression flag, and bot ACL enforcement
(unauthorized / viewer-blocked-from-mutations / operator-can-rotate).

PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/log_notifier_spec.cr

@@ -0,0 +1,27 @@
+# ===================
+# ©AngelaMos | 2026
+# log_notifier_spec.cr
+# ===================
+
+require "../../spec_helper"
+require "../../../src/cre/notifiers/log_notifier"
+require "../../../src/cre/events/credential_events"
+
+describe CRE::Notifiers::LogNotifier do
+  it "subscribes and emits without errors on rotation events" do
+    bus = CRE::Engine::EventBus.new
+    notifier = CRE::Notifiers::LogNotifier.new(bus)
+    notifier.start
+    bus.run
+
+    cred_id = UUID.random
+    bus.publish CRE::Events::RotationCompleted.new(cred_id, UUID.random)
+    bus.publish CRE::Events::RotationFailed.new(cred_id, UUID.random, "test")
+    bus.publish CRE::Events::PolicyViolation.new(cred_id, "p", "stale")
+    bus.publish CRE::Events::AlertRaised.new(CRE::Events::Severity::Warn, "hi")
+
+    sleep 0.1.seconds
+    notifier.stop
+    bus.stop
+  end
+end

PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/telegram_bot_spec.cr

@@ -0,0 +1,93 @@
+# ===================
+# ©AngelaMos | 2026
+# telegram_bot_spec.cr
+# ===================
+
+require "../../spec_helper"
+require "webmock"
+require "../../../src/cre/notifiers/telegram_bot"
+require "../../../src/cre/persistence/sqlite/sqlite_persistence"
+
+WebMock.allow_net_connect = false
+
+private def fresh_setup
+  persist = CRE::Persistence::Sqlite::SqlitePersistence.new(":memory:")
+  persist.migrate!
+  bus = CRE::Engine::EventBus.new
+  bus.run
+  telegram = CRE::Notifiers::Telegram.new("FAKE")
+  bot = CRE::Notifiers::TelegramBot.new(
+    bus: bus,
+    telegram: telegram,
+    persistence: persist,
+    viewer_chats: [100_i64],
+    operator_chats: [200_i64],
+  )
+  {persist, bus, telegram, bot}
+end
+
+describe CRE::Notifiers::TelegramBot do
+  it "viewer can run /status" do
+    persist, bus, _, bot = fresh_setup
+    reply = bot.handle_command(100_i64, "/status")
+    reply.should contain "live"
+    reply.should contain "Credentials"
+  ensure
+    bus.try(&.stop)
+    persist.try(&.close)
+  end
+
+  it "viewer cannot /rotate" do
+    persist, bus, _, bot = fresh_setup
+    reply = bot.handle_command(100_i64, "/rotate 00000000-0000-0000-0000-000000000000")
+    reply.should contain "operator-only"
+  ensure
+    bus.try(&.stop)
+    persist.try(&.close)
+  end
+
+  it "operator can /rotate; publishes RotationScheduled" do
+    persist, bus, _, bot = fresh_setup
+
+    received = [] of CRE::Events::Event
+    received_mutex = Mutex.new
+    ch = bus.subscribe
+    spawn do
+      loop do
+        begin
+          ev = ch.receive
+          received_mutex.synchronize { received << ev }
+        rescue ::Channel::ClosedError
+          break
+        end
+      end
+    end
+
+    cred_id = UUID.random
+    reply = bot.handle_command(200_i64, "/rotate #{cred_id}")
+    reply.should contain "rotation scheduled"
+    sleep 0.1.seconds
+    received_mutex.synchronize { received.any?(&.is_a?(CRE::Events::RotationScheduled)).should be_true }
+  ensure
+    bus.try(&.stop)
+    persist.try(&.close)
+  end
+
+  it "unauthorized chat is blocked" do
+    persist, bus, _, bot = fresh_setup
+    bot.handle_command(999_i64, "/status").should eq "unauthorized"
+  ensure
+    bus.try(&.stop)
+    persist.try(&.close)
+  end
+
+  it "/help lists commands" do
+    persist, bus, _, bot = fresh_setup
+    reply = bot.handle_command(100_i64, "/help")
+    reply.should contain "/status"
+    reply.should contain "/rotate"
+  ensure
+    bus.try(&.stop)
+    persist.try(&.close)
+  end
+end

PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/telegram_spec.cr

@@ -0,0 +1,101 @@
+# ===================
+# ©AngelaMos | 2026
+# telegram_spec.cr
+# ===================
+
+require "../../spec_helper"
+require "webmock"
+require "../../../src/cre/notifiers/telegram"
+require "../../../src/cre/notifiers/telegram_subscriber"
+require "../../../src/cre/events/credential_events"
+
+WebMock.allow_net_connect = false
+
+describe CRE::Notifiers::Telegram do
+  before_each { WebMock.reset }
+
+  it "sends a message" do
+    sent_body = nil
+    WebMock.stub(:post, "https://api.telegram.org/botFAKE/sendMessage")
+      .to_return { |req| sent_body = req.body.try(&.gets_to_end); HTTP::Client::Response.new(200, body: %({"ok":true})) }
+    CRE::Notifiers::Telegram.new("FAKE").send_message(12345_i64, "hello world")
+    sent_body.try(&.includes?("hello world")).should be_true
+    sent_body.try(&.includes?(%("chat_id":12345))).should be_true
+  end
+
+  it "raises TelegramError on non-2xx" do
+    WebMock.stub(:post, "https://api.telegram.org/botFAKE/sendMessage")
+      .to_return(status: 401, body: %({"ok":false,"description":"Unauthorized"}))
+    expect_raises(CRE::Notifiers::Telegram::TelegramError) do
+      CRE::Notifiers::Telegram.new("FAKE").send_message(1_i64, "x")
+    end
+  end
+
+  it "parses getUpdates with messages" do
+    WebMock.stub(:post, "https://api.telegram.org/botFAKE/getUpdates")
+      .to_return(body: %({
+        "ok":true,
+        "result":[{
+          "update_id":42,
+          "message":{
+            "message_id":7,
+            "chat":{"id":99},
+            "text":"/status"
+          }
+        }]
+      }))
+    updates = CRE::Notifiers::Telegram.new("FAKE").get_updates
+    updates.size.should eq 1
+    updates[0].chat_id.should eq 99
+    updates[0].text.should eq "/status"
+  end
+end
+
+describe CRE::Notifiers::TelegramSubscriber do
+  before_each { WebMock.reset }
+
+  it "fires Telegram message on RotationFailed" do
+    sent = [] of String
+    WebMock.stub(:post, "https://api.telegram.org/botFAKE/sendMessage")
+      .to_return { |req|
+        body = req.body.try(&.gets_to_end) || ""
+        sent << body
+        HTTP::Client::Response.new(200, body: %({"ok":true}))
+      }
+
+    bus = CRE::Engine::EventBus.new
+    sub = CRE::Notifiers::TelegramSubscriber.new(
+      bus, CRE::Notifiers::Telegram.new("FAKE"), [12345_i64],
+    )
+    sub.start
+    bus.run
+
+    bus.publish CRE::Events::RotationFailed.new(UUID.random, UUID.random, "boom")
+    sleep 0.1.seconds
+
+    sent.size.should eq 1
+    sent[0].should contain "FAILED"
+  ensure
+    bus.try(&.stop)
+    sub.try(&.stop)
+  end
+
+  it "does not fire on RotationCompleted unless notify_on_success" do
+    sent = [] of String
+    WebMock.stub(:post, "https://api.telegram.org/botFAKE/sendMessage")
+      .to_return { |req| sent << (req.body.try(&.gets_to_end) || ""); HTTP::Client::Response.new(200, body: %({"ok":true})) }
+
+    bus = CRE::Engine::EventBus.new
+    sub = CRE::Notifiers::TelegramSubscriber.new(
+      bus, CRE::Notifiers::Telegram.new("FAKE"), [1_i64], notify_on_success: false,
+    )
+    sub.start
+    bus.run
+    bus.publish CRE::Events::RotationCompleted.new(UUID.random, UUID.random)
+    sleep 0.1.seconds
+    sent.size.should eq 0
+  ensure
+    bus.try(&.stop)
+    sub.try(&.stop)
+  end
+end

PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/log_notifier.cr

@@ -0,0 +1,65 @@
+# ===================
+# ©AngelaMos | 2026
+# log_notifier.cr
+# ===================
+
+require "log"
+require "../engine/event_bus"
+require "../events/credential_events"
+require "../events/system_events"
+
+module CRE::Notifiers
+  # LogNotifier subscribes to all events and emits structured stdlib Log lines.
+  # Suitable for shipping to journald, vector, or fluentd; downstream tooling
+  # can ingest the structured fields directly.
+  class LogNotifier
+    Log = ::Log.for("cre.notifier")
+
+    @ch : ::Channel(Events::Event)?
+    @running : Bool
+
+    def initialize(@bus : Engine::EventBus)
+      @running = false
+    end
+
+    def start : Nil
+      @running = true
+      ch = @bus.subscribe(buffer: 64, overflow: Engine::EventBus::Overflow::Drop)
+      @ch = ch
+      spawn(name: "log-notifier") do
+        while @running
+          begin
+            ev = ch.receive
+          rescue ::Channel::ClosedError
+            break
+          end
+          emit(ev)
+        end
+      end
+    end
+
+    def stop : Nil
+      @running = false
+      @ch.try(&.close)
+    end
+
+    private def emit(ev : Events::Event) : Nil
+      case ev
+      when Events::RotationCompleted
+        Log.info &.emit("rotation completed", credential_id: ev.credential_id.to_s, rotation_id: ev.rotation_id.to_s)
+      when Events::RotationFailed
+        Log.error &.emit("rotation failed", credential_id: ev.credential_id.to_s, rotation_id: ev.rotation_id.to_s, reason: ev.reason)
+      when Events::PolicyViolation
+        Log.warn &.emit("policy violation", credential_id: ev.credential_id.to_s, policy: ev.policy_name, reason: ev.reason)
+      when Events::DriftDetected
+        Log.warn &.emit("drift detected", credential_id: ev.credential_id.to_s, expected: ev.expected_hash, actual: ev.actual_hash)
+      when Events::AlertRaised
+        case ev.severity
+        in Events::Severity::Critical then Log.error &.emit("alert", text: ev.message)
+        in Events::Severity::Warn     then Log.warn  &.emit("alert", text: ev.message)
+        in Events::Severity::Info     then Log.info  &.emit("alert", text: ev.message)
+        end
+      end
+    end
+  end
+end

PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram.cr

@@ -0,0 +1,65 @@
+# ===================
+# ©AngelaMos | 2026
+# telegram.cr
+# ===================
+
+require "http/client"
+require "json"
+require "log"
+
+module CRE::Notifiers
+  # Thin Telegram Bot API client. We hit api.telegram.org/bot<TOKEN>/<METHOD>
+  # directly with HTTP::Client; no tourmaline dependency for the notification
+  # path keeps the footprint small.
+  class Telegram
+    Log = ::Log.for("cre.telegram")
+    DEFAULT_API = "https://api.telegram.org"
+
+    class TelegramError < Exception
+      getter status : Int32
+
+      def initialize(message : String, @status : Int32)
+        super(message)
+      end
+    end
+
+    record Update,
+      update_id : Int64,
+      message_id : Int64?,
+      chat_id : Int64?,
+      text : String?
+
+    def initialize(@token : String, @api_base : String = DEFAULT_API)
+    end
+
+    def send_message(chat_id : Int64, text : String, parse_mode : String? = nil) : Nil
+      payload = {"chat_id" => chat_id, "text" => text} of String => String | Int64
+      payload["parse_mode"] = parse_mode if parse_mode
+      call("sendMessage", payload.to_json)
+    end
+
+    def get_updates(offset : Int64? = nil, timeout : Int32 = 30) : Array(Update)
+      h = {"timeout" => timeout} of String => String | Int64 | Int32
+      h["offset"] = offset if offset
+      json = call("getUpdates", h.to_json)
+      results = json["result"].as_a
+      results.map do |entry|
+        msg = entry["message"]?
+        Update.new(
+          update_id: entry["update_id"].as_i64,
+          message_id: msg.try(&.["message_id"]?.try(&.as_i64)),
+          chat_id: msg.try(&.["chat"]?.try(&.["id"]?.try(&.as_i64))),
+          text: msg.try(&.["text"]?.try(&.as_s)),
+        )
+      end
+    end
+
+    private def call(method : String, body : String) : JSON::Any
+      uri = "#{@api_base}/bot#{@token}/#{method}"
+      headers = HTTP::Headers{"Content-Type" => "application/json"}
+      response = HTTP::Client.post(uri, headers: headers, body: body)
+      raise TelegramError.new("telegram #{method} #{response.status_code}: #{response.body[0, 200]?}", response.status_code) unless response.status_code < 300
+      JSON.parse(response.body)
+    end
+  end
+end

PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram_bot.cr

@@ -0,0 +1,149 @@
+# ===================
+# ©AngelaMos | 2026
+# telegram_bot.cr
+# ===================
+
+require "log"
+require "./telegram"
+require "../engine/event_bus"
+require "../persistence/persistence"
+
+module CRE::Notifiers
+  # TelegramBot does long-polling getUpdates and dispatches commands to
+  # operator handlers. Two ACL tiers:
+  #   - viewer  : read-only commands (/status, /queue, /history, /alerts, /help)
+  #   - operator: viewer + mutating commands (/rotate, /snooze)
+  #
+  # Authorization is by chat_id; not strictly authentication but adequate for a
+  # single-tenant deployment where the bot token + chat IDs live in env vars.
+  class TelegramBot
+    Log = ::Log.for("cre.telegram_bot")
+
+    @running : Bool
+    @last_offset : Int64
+
+    def initialize(
+      @bus : Engine::EventBus,
+      @telegram : Telegram,
+      @persistence : Persistence::Persistence,
+      @viewer_chats : Array(Int64),
+      @operator_chats : Array(Int64),
+    )
+      @running = false
+      @last_offset = 0_i64
+    end
+
+    def start : Nil
+      @running = true
+      spawn(name: "telegram-bot") do
+        while @running
+          begin
+            poll_once
+          rescue ex
+            Log.error(exception: ex) { "telegram bot poll failed" }
+            sleep 1.second
+          end
+        end
+      end
+    end
+
+    def stop : Nil
+      @running = false
+    end
+
+    def authorized_viewer?(chat_id : Int64) : Bool
+      @viewer_chats.includes?(chat_id) || @operator_chats.includes?(chat_id)
+    end
+
+    def authorized_operator?(chat_id : Int64) : Bool
+      @operator_chats.includes?(chat_id)
+    end
+
+    def handle_command(chat_id : Int64, text : String) : String
+      return "unauthorized" unless authorized_viewer?(chat_id)
+      cmd, _, rest = text.strip.lstrip('/').partition(' ')
+      case cmd
+      when "status" then status_message
+      when "queue"  then queue_message
+      when "alerts" then alerts_message
+      when "help", "" then help_message
+      when "rotate" then handle_rotate(chat_id, rest)
+      when "snooze" then handle_snooze(chat_id, rest)
+      when "history" then history_message(rest)
+      else
+        "unknown command: /#{cmd} (try /help)"
+      end
+    end
+
+    def poll_once : Nil
+      updates = @telegram.get_updates(offset: @last_offset == 0 ? nil : @last_offset, timeout: 5)
+      updates.each do |u|
+        @last_offset = u.update_id + 1
+        chat_id = u.chat_id
+        text = u.text
+        next if chat_id.nil? || text.nil? || !text.starts_with?('/')
+        reply = handle_command(chat_id, text)
+        @telegram.send_message(chat_id, reply) rescue nil
+      end
+    end
+
+    private def status_message : String
+      total = @persistence.credentials.all.size
+      in_flight = @persistence.rotations.in_flight.size
+      "● live\nCredentials: #{total}\nIn-flight rotations: #{in_flight}"
+    end
+
+    private def queue_message : String
+      in_flight = @persistence.rotations.in_flight
+      return "queue empty" if in_flight.empty?
+      lines = in_flight.first(10).map { |r| "- #{r.rotator_kind} #{r.credential_id} [#{r.state}]" }
+      "Active queue (#{in_flight.size}):\n#{lines.join('\n')}"
+    end
+
+    private def alerts_message : String
+      "Alerts inspection is exposed via the audit log; use 'cre audit verify --since=...' from the CLI."
+    end
+
+    private def help_message : String
+      <<-MD
+      Available commands:
+        /status            - quick health summary
+        /queue             - active rotations
+        /history <id>      - last events for a credential
+        /alerts            - critical alerts pointer
+        /rotate <id>       - force rotation (operator)
+        /snooze <id> 24h   - defer scheduled rotation (operator)
+      MD
+    end
+
+    private def handle_rotate(chat_id : Int64, rest : String) : String
+      return "operator-only command" unless authorized_operator?(chat_id)
+      id_str = rest.strip
+      return "usage: /rotate <credential-id>" if id_str.empty?
+      uuid = UUID.new(id_str) rescue nil
+      return "invalid credential id" if uuid.nil?
+      @bus.publish Events::RotationScheduled.new(uuid, "manual")
+      "rotation scheduled for #{uuid}"
+    end
+
+    private def handle_snooze(chat_id : Int64, rest : String) : String
+      return "operator-only command" unless authorized_operator?(chat_id)
+      "snooze is not yet implemented; track via /queue"
+    end
+
+    private def history_message(rest : String) : String
+      id_str = rest.strip
+      return "usage: /history <credential-id>" if id_str.empty?
+      uuid = UUID.new(id_str) rescue nil
+      return "invalid credential id" if uuid.nil?
+      cred = @persistence.credentials.find(uuid)
+      return "credential not found" if cred.nil?
+      latest = @persistence.audit.latest_seq
+      entries = latest > 10 ? @persistence.audit.range(latest - 9, latest) : @persistence.audit.range(1_i64, latest)
+      filtered = entries.select { |e| e.target_id == uuid }
+      return "no audit entries for #{uuid}" if filtered.empty?
+      lines = filtered.last(10).map { |e| "- #{e.occurred_at.to_rfc3339}: #{e.event_type}" }
+      "Last events for #{cred.name}:\n#{lines.join('\n')}"
+    end
+  end
+end

PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram_subscriber.cr

@@ -0,0 +1,80 @@
+# ===================
+# ©AngelaMos | 2026
+# telegram_subscriber.cr
+# ===================
+
+require "log"
+require "./telegram"
+require "../engine/event_bus"
+require "../events/credential_events"
+require "../events/system_events"
+
+module CRE::Notifiers
+  # TelegramSubscriber sends emoji-prefixed messages to allowlisted chats on
+  # significant events. Best-effort delivery (Drop overflow); transient
+  # Telegram errors are logged and swallowed so a network blip never blocks
+  # the engine.
+  class TelegramSubscriber
+    Log = ::Log.for("cre.telegram_subscriber")
+
+    @ch : ::Channel(Events::Event)?
+    @running : Bool
+
+    def initialize(@bus : Engine::EventBus, @telegram : Telegram, @viewer_chats : Array(Int64), @notify_on_success : Bool = false)
+      @running = false
+    end
+
+    def start : Nil
+      @running = true
+      ch = @bus.subscribe(buffer: 128, overflow: Engine::EventBus::Overflow::Drop)
+      @ch = ch
+      spawn(name: "telegram-sub") do
+        while @running
+          begin
+            ev = ch.receive
+          rescue ::Channel::ClosedError
+            break
+          end
+          dispatch(ev)
+        end
+      end
+    end
+
+    def stop : Nil
+      @running = false
+      @ch.try(&.close)
+    end
+
+    private def dispatch(ev : Events::Event) : Nil
+      msg = format(ev)
+      return if msg.nil?
+      @viewer_chats.each do |chat|
+        @telegram.send_message(chat, msg)
+      rescue ex
+        Log.warn(exception: ex) { "telegram send failed for chat=#{chat}" }
+      end
+    end
+
+    private def format(ev : Events::Event) : String?
+      case ev
+      when Events::RotationFailed
+        "! Rotation FAILED for credential #{ev.credential_id} (rotation #{ev.rotation_id}): #{ev.reason}"
+      when Events::DriftDetected
+        "⚠ Drift detected on credential #{ev.credential_id}: hash mismatch (expected=#{ev.expected_hash[0, 12]}..., actual=#{ev.actual_hash[0, 12]}...)"
+      when Events::PolicyViolation
+        "⚠ Policy violation: '#{ev.policy_name}' on #{ev.credential_id} (#{ev.reason})"
+      when Events::AlertRaised
+        sev = case ev.severity
+              in Events::Severity::Critical then "!"
+              in Events::Severity::Warn     then "⚠"
+              in Events::Severity::Info     then "ℹ"
+              end
+        "#{sev} #{ev.message}"
+      when Events::RotationCompleted
+        @notify_on_success ? "✓ Rotation completed for credential #{ev.credential_id}" : nil
+      else
+        nil
+      end
+    end
+  end
+end