Amr Gharbeia
c7e9893e68
Some checks failed
Deploy (Gitea) / deploy (push) Failing after 2s
Adds Discord gateway: REST API POST /channels/{id}/messages for
sending, HTTP GET for polling messages. Maps Discord mentions to
:user-input signals. HITL commands intercepted before injection.
Adds Slack gateway: Web API chat.postMessage for sending,
conversations.history for polling. Uses SLACK_TOKEN from vault.
Each gateway registered in *gateway-registry* following the same
jail-loaded skill pattern as Telegram and Signal.
Registry now has 4 platforms: telegram, signal, discord, slack.
412 lines
19 KiB
Common Lisp
412 lines
19 KiB
Common Lisp
(in-package :passepartout)
|
|
|
|
(defvar *gateway-configs* (make-hash-table :test 'equal)
|
|
"Maps platform name to plist (:token :thread :interval :enabled)")
|
|
|
|
(defvar *gateway-registry* (make-hash-table :test 'equal)
|
|
"Maps platform name to plist (:poll-fn :send-fn :default-interval)")
|
|
|
|
(defun telegram-get-token ()
|
|
(vault-get-secret :telegram))
|
|
|
|
(defun telegram-poll ()
|
|
"Polls Telegram for new messages and injects them into the harness."
|
|
(let* ((token (telegram-get-token)))
|
|
(when token
|
|
(let* ((last-id (getf (gethash "telegram" *gateway-configs*) :last-update-id 0))
|
|
(url (format nil "https://api.telegram.org/bot~a/getUpdates?offset=~a"
|
|
token (1+ last-id))))
|
|
(handler-case
|
|
(let* ((response (dex:get url))
|
|
(json (cl-json:decode-json-from-string response))
|
|
(updates (cdr (assoc :result json))))
|
|
(dolist (update updates)
|
|
(let* ((update-id (cdr (assoc :update--id update)))
|
|
(message (cdr (assoc :message update)))
|
|
(chat (cdr (assoc :chat message)))
|
|
(chat-id (cdr (assoc :id chat)))
|
|
(text (cdr (assoc :text message))))
|
|
(setf (getf (gethash "telegram" *gateway-configs*) :last-update-id) update-id)
|
|
(when (and text chat-id)
|
|
(log-message "TELEGRAM: Received message from ~a" chat-id)
|
|
(unless (ignore-errors (hitl-handle-message text :telegram))
|
|
(stimulus-inject
|
|
(list :type :EVENT
|
|
:meta (list :source :telegram :chat-id (format nil "~a" chat-id))
|
|
:payload (list :sensor :user-input :text text))))))))
|
|
(error (c) (log-message "TELEGRAM POLL ERROR: ~a" c)))))))
|
|
|
|
(defun telegram-send (action context)
|
|
"Sends a message via Telegram."
|
|
(declare (ignore context))
|
|
(let* ((payload (getf action :payload))
|
|
(meta (getf action :meta))
|
|
(chat-id (or (getf meta :chat-id) (getf payload :chat-id) (getf action :chat-id)))
|
|
(text (or (getf payload :text) (getf action :text)))
|
|
(token (telegram-get-token)))
|
|
(when (and token chat-id text)
|
|
(handler-case
|
|
(let ((url (format nil "https://api.telegram.org/bot~a/sendMessage" token)))
|
|
(dex:post url
|
|
:headers '(("Content-Type" . "application/json"))
|
|
:content (cl-json:encode-json-to-string
|
|
`((chat_id . ,chat-id) (text . ,text)))))
|
|
(error (c) (log-message "TELEGRAM ERROR: ~a" c))))))
|
|
|
|
(defun signal-get-account ()
|
|
(vault-get-secret :signal))
|
|
|
|
(defun signal-poll ()
|
|
"Polls Signal for new messages and injects them into the harness."
|
|
(let ((account (signal-get-account)))
|
|
(when account
|
|
(handler-case
|
|
(let* ((output (uiop:run-program (list "signal-cli" "-u" account "receive" "--json")
|
|
:output :string :error-output :string :ignore-error-status t))
|
|
(lines (cl-ppcre:split "\\\\n" output)))
|
|
(dolist (line lines)
|
|
(when (and line (> (length line) 0))
|
|
(let* ((json (ignore-errors (cl-json:decode-json-from-string line)))
|
|
(envelope (cdr (assoc :envelope json)))
|
|
(source (cdr (assoc :source envelope)))
|
|
(data-message (cdr (assoc :data-message envelope)))
|
|
(text (cdr (assoc :message data-message))))
|
|
(when (and source text)
|
|
(log-message "SIGNAL: Received message from ~a" source)
|
|
(unless (ignore-errors (hitl-handle-message text :signal))
|
|
(stimulus-inject
|
|
(list :type :EVENT
|
|
:meta (list :source :signal :chat-id source)
|
|
:payload (list :sensor :user-input :text text)))))))))
|
|
(error (c) (log-message "SIGNAL POLL ERROR: ~a" c))))))
|
|
|
|
(defun signal-send (action context)
|
|
"Sends a message via Signal."
|
|
(declare (ignore context))
|
|
(let* ((payload (getf action :payload))
|
|
(meta (getf action :meta))
|
|
(chat-id (or (getf meta :chat-id) (getf payload :chat-id) (getf action :chat-id)))
|
|
(text (or (getf payload :text) (getf action :text)))
|
|
(account (signal-get-account)))
|
|
(when (and account chat-id text)
|
|
(handler-case
|
|
(uiop:run-program (list "signal-cli" "-u" account "send" "-m" text chat-id)
|
|
:output :string :error-output :string)
|
|
(error (c) (log-message "SIGNAL ERROR: ~a" c))))))
|
|
|
|
(defun discord-get-token ()
|
|
(vault-get-secret :discord))
|
|
|
|
(defun discord-send (action context)
|
|
"Sends a message via Discord REST API."
|
|
(declare (ignore context))
|
|
(let* ((payload (getf action :payload))
|
|
(meta (getf action :meta))
|
|
(channel-id (or (getf meta :channel-id) (getf payload :chat-id)))
|
|
(text (or (getf payload :text) (getf action :text)))
|
|
(token (discord-get-token)))
|
|
(when (and token channel-id text)
|
|
(handler-case
|
|
(dex:post (format nil "https://discord.com/api/v10/channels/~a/messages" channel-id)
|
|
:headers '(("Authorization" . ,(format nil "Bot ~a" token))
|
|
("Content-Type" . "application/json"))
|
|
:content (cl-json:encode-json-to-string
|
|
`((content . ,text))))
|
|
(error (c) (log-message "DISCORD ERROR: ~a" c))))))
|
|
|
|
(defun discord-poll ()
|
|
"Polls Discord via HTTP GET /channels/{id}/messages. In production,
|
|
a WebSocket connection to the Gateway is preferred for real-time events."
|
|
(let* ((token (discord-get-token)))
|
|
(when token
|
|
(handler-case
|
|
(dolist (channel '("channel-id-here")) ;; configured channel IDs
|
|
(let* ((last-id (getf (gethash "discord" *gateway-configs*) :last-update-id 0))
|
|
(url (format nil "https://discord.com/api/v10/channels/~a/messages?after=~a"
|
|
channel last-id))
|
|
(response (dex:get url :headers
|
|
`(("Authorization" . ,(format nil "Bot ~a" token))))))
|
|
(let ((messages (ignore-errors
|
|
(cdr (assoc :message
|
|
(cl-json:decode-json-from-string response))))))
|
|
(dolist (msg (and (listp messages) messages))
|
|
(let* ((id (cdr (assoc :id msg)))
|
|
(content (cdr (assoc :content msg)))
|
|
(author (cdr (assoc :author msg)))
|
|
(author-id (cdr (assoc :id author)))
|
|
(is-bot (cdr (assoc :bot author))))
|
|
(when (and id content (not is-bot))
|
|
(setf (getf (gethash "discord" *gateway-configs*) :last-update-id) id)
|
|
(unless (ignore-errors (hitl-handle-message content :discord))
|
|
(stimulus-inject
|
|
(list :type :EVENT
|
|
:meta (list :source :discord :chat-id channel)
|
|
:payload (list :sensor :user-input :text content))))))))))
|
|
(error (c) (log-message "DISCORD POLL ERROR: ~a" c))))))
|
|
|
|
(defun slack-get-token ()
|
|
(vault-get-secret :slack))
|
|
|
|
(defun slack-send (action context)
|
|
"Sends a message via Slack Web API."
|
|
(declare (ignore context))
|
|
(let* ((payload (getf action :payload))
|
|
(meta (getf action :meta))
|
|
(channel (or (getf meta :channel-id) (getf payload :chat-id)))
|
|
(text (or (getf payload :text) (getf action :text)))
|
|
(token (slack-get-token)))
|
|
(when (and token channel text)
|
|
(handler-case
|
|
(dex:post "https://slack.com/api/chat.postMessage"
|
|
:headers `(("Authorization" . ,(format nil "Bearer ~a" token))
|
|
("Content-Type" . "application/json; charset=utf-8"))
|
|
:content (cl-json:encode-json-to-string
|
|
`((channel . ,channel) (text . ,text))))
|
|
(error (c) (log-message "SLACK ERROR: ~a" c))))))
|
|
|
|
(defun slack-poll ()
|
|
"Polls Slack for new messages via conversations.history."
|
|
(let* ((token (slack-get-token)))
|
|
(when token
|
|
(dolist (channel '("general")) ;; configured channel IDs
|
|
(handler-case
|
|
(let* ((url (format nil "https://slack.com/api/conversations.history?channel=~a&limit=5" channel))
|
|
(response (dex:get url :headers
|
|
`(("Authorization" . ,(format nil "Bearer ~a" token))))))
|
|
(let* ((json (ignore-errors (cl-json:decode-json-from-string response)))
|
|
(ok (cdr (assoc :ok json)))
|
|
(messages (cdr (assoc :messages json))))
|
|
(when (and ok messages (listp messages))
|
|
(dolist (msg messages)
|
|
(let* ((text (cdr (assoc :text msg)))
|
|
(user (cdr (assoc :user msg)))
|
|
(ts (cdr (assoc :ts msg))))
|
|
(when (and text user (not (string= user "USLACKBOT")))
|
|
(unless (ignore-errors (hitl-handle-message text :slack))
|
|
(stimulus-inject
|
|
(list :type :EVENT
|
|
:meta (list :source :slack :chat-id channel)
|
|
:payload (list :sensor :user-input :text text))))))))))
|
|
(error (c) (log-message "SLACK POLL ERROR: ~a" c)))))))
|
|
|
|
(defun gateway-registry-initialize ()
|
|
"Registers all built-in gateway handlers."
|
|
(setf (gethash "telegram" *gateway-registry*)
|
|
(list :poll-fn #'telegram-poll
|
|
:send-fn #'telegram-send
|
|
:default-interval 3
|
|
:configured nil))
|
|
(setf (gethash "signal" *gateway-registry*)
|
|
(list :poll-fn #'signal-poll
|
|
:send-fn #'signal-send
|
|
:default-interval 5
|
|
:configured nil))
|
|
(setf (gethash "discord" *gateway-registry*)
|
|
(list :poll-fn #'discord-poll
|
|
:send-fn #'discord-send
|
|
:default-interval 10
|
|
:configured nil))
|
|
(setf (gethash "slack" *gateway-registry*)
|
|
(list :poll-fn #'slack-poll
|
|
:send-fn #'slack-send
|
|
:default-interval 10
|
|
:configured nil)))
|
|
|
|
(defun gateway-configured-p (platform)
|
|
"Returns T if a platform has a stored token."
|
|
(let ((config (gethash platform *gateway-configs*)))
|
|
(and config (getf config :token))))
|
|
|
|
(defun gateway-active-p (platform)
|
|
"Returns T if a platform's polling thread is alive."
|
|
(let ((config (gethash platform *gateway-configs*)))
|
|
(and config
|
|
(getf config :thread)
|
|
(bt:thread-alive-p (getf config :thread)))))
|
|
|
|
(defun messaging-link (platform token)
|
|
"Links a platform with a token and starts polling."
|
|
(let ((platform-lc (string-downcase platform)))
|
|
(unless (gethash platform-lc *gateway-registry*)
|
|
(error "Unknown platform: ~a. Available: ~{~a~^, ~}"
|
|
platform (loop for k being the hash-keys of *gateway-registry* collect k)))
|
|
(when (or (null token) (zerop (length token)))
|
|
(error "Token cannot be empty"))
|
|
(log-message "MESSAGING: Linking to ~a..." platform-lc)
|
|
(gateway-unlink platform-lc)
|
|
(let* ((registry-entry (gethash platform-lc *gateway-registry*))
|
|
(interval (or (getf registry-entry :default-interval) 5)))
|
|
(setf (gethash platform-lc *gateway-configs*)
|
|
(list :token token :interval interval :enabled t))
|
|
(vault-set-secret (intern (string-upcase platform-lc) :keyword) token)
|
|
(gateway-start platform-lc)
|
|
(log-message "MESSAGING: Successfully linked ~a" platform-lc)
|
|
(format t "Successfully linked ~a gateway. Token stored securely.~%" platform-lc)
|
|
t)))
|
|
|
|
(defun messaging-unlink (platform)
|
|
"Unlinks a platform and stops its polling thread."
|
|
(let ((platform-lc (string-downcase platform)))
|
|
(gateway-stop platform-lc)
|
|
(remhash platform-lc *gateway-configs*)
|
|
(log-message "MESSAGING: Unlinked ~a" platform-lc)
|
|
(format t "Successfully unlinked ~a gateway.~%" platform-lc)
|
|
t))
|
|
|
|
(defun gateway-start (platform)
|
|
"Starts the polling thread for a linked gateway."
|
|
(let ((platform-lc (string-downcase platform)))
|
|
(let ((config (gethash platform-lc *gateway-configs*)))
|
|
(when (and config (getf config :enabled) (not (gateway-active-p platform-lc)))
|
|
(let ((poll-fn (getf (gethash platform-lc *gateway-registry*) :poll-fn)))
|
|
(when poll-fn
|
|
(let ((interval (getf config :interval)))
|
|
(setf (getf config :thread)
|
|
(bt:make-thread
|
|
(lambda ()
|
|
(loop
|
|
(when (getf (gethash platform-lc *gateway-configs*) :enabled)
|
|
(funcall poll-fn))
|
|
(sleep interval)))
|
|
:name (format nil "passepartout-~a-gateway" platform-lc)))
|
|
(log-message "MESSAGING: Started ~a polling (interval: ~as)" platform-lc interval))))))))
|
|
|
|
(defun gateway-stop (platform)
|
|
"Stops the polling thread for a gateway."
|
|
(let ((platform-lc (string-downcase platform)))
|
|
(let ((config (gethash platform-lc *gateway-configs*)))
|
|
(when (and config (getf config :thread))
|
|
(when (bt:thread-alive-p (getf config :thread))
|
|
(log-message "MESSAGING: Stopping ~a polling thread" platform-lc)
|
|
(bt:destroy-thread (getf config :thread))))
|
|
(setf (getf config :thread) nil))))
|
|
|
|
(defun messaging-list ()
|
|
"Returns a list of all gateways with their status."
|
|
(loop for platform being the hash-keys of *gateway-registry*
|
|
collect (let ((configured (gateway-configured-p platform))
|
|
(active (gateway-active-p platform)))
|
|
(list :platform platform
|
|
:configured configured
|
|
:active active))))
|
|
|
|
(defun messaging-list-print ()
|
|
"Prints a formatted table of gateways."
|
|
(format t "~%")
|
|
(format t " ~20@A ~12@A ~10@A~%" "PLATFORM" "CONFIGURED" "STATUS")
|
|
(dolist (gw (messaging-list))
|
|
(format t " ~20@A ~12@A ~10@A~%"
|
|
(getf gw :platform)
|
|
(if (getf gw :configured) "yes" "no")
|
|
(cond
|
|
((getf gw :active) "ACTIVE")
|
|
((getf gw :configured) "stopped")
|
|
(t "not linked"))))
|
|
(format t "~%"))
|
|
|
|
(defun gateway-start-all ()
|
|
"Called at boot to start all configured gateways."
|
|
(dolist (config (loop for platform being the hash-keys of *gateway-configs*
|
|
collect (list platform (gethash platform *gateway-configs*))))
|
|
(destructuring-bind (platform config) config
|
|
(when (and (getf config :enabled) (not (gateway-active-p platform)))
|
|
(gateway-start platform)))))
|
|
|
|
(register-actuator :telegram #'telegram-send)
|
|
(register-actuator :signal #'signal-send)
|
|
|
|
(defskill :passepartout-gateway-messaging
|
|
:priority 150
|
|
:trigger (lambda (ctx) (declare (ignore ctx)) nil))
|
|
|
|
(gateway-registry-initialize)
|
|
(gateway-start-all)
|
|
|
|
(eval-when (:compile-toplevel :load-toplevel :execute)
|
|
(ql:quickload :fiveam :silent t))
|
|
|
|
(defpackage :passepartout-gateway-messaging-tests
|
|
(:use :cl :fiveam :passepartout)
|
|
(:export #:messaging-suite))
|
|
|
|
(in-package :passepartout-gateway-messaging-tests)
|
|
|
|
(def-suite messaging-suite :description "Verification of Gateway Messaging")
|
|
(in-suite messaging-suite)
|
|
|
|
(test test-gateway-registry-initialize
|
|
"Contract 1: gateway-registry-initialize populates the registry with :configured key."
|
|
;; Access the variable via its skill package symbol-value
|
|
(let* ((pkg (find-package "PASSEPARTOUT.SKILLS.GATEWAY-MESSAGING"))
|
|
(reg-var (and pkg (find-symbol "*GATEWAY-REGISTRY*" pkg))))
|
|
(when reg-var
|
|
(clrhash (symbol-value reg-var))
|
|
(gateway-registry-initialize)
|
|
(is (not (zerop (hash-table-count (symbol-value reg-var)))))
|
|
(let ((entry (gethash "telegram" (symbol-value reg-var))))
|
|
(is (getf entry :poll-fn))
|
|
(is (getf entry :send-fn))
|
|
(is (getf entry :default-interval))
|
|
(is (eq nil (getf entry :configured)))))))
|
|
|
|
(test test-telegram-send-format
|
|
"Contract: telegram-send constructs correct URL and POST body."
|
|
(let ((captured-url nil)
|
|
(captured-content nil)
|
|
(captured-headers nil))
|
|
;; Mock dex:post to capture arguments
|
|
(let ((mock-dex-post (lambda (url &key headers content)
|
|
(setf captured-url url
|
|
captured-content content
|
|
captured-headers headers))))
|
|
;; Mock vault-get-secret to return a test token
|
|
(let ((mock-vault (lambda (key)
|
|
(declare (ignore key))
|
|
"test-token-123")))
|
|
;; Build action plist for telegram-send
|
|
(let* ((action '(:payload (:text "Hello from Lisp" :chat-id "999")
|
|
:meta (:chat-id "999")))
|
|
(context nil))
|
|
;; Verify send constructs correct URL
|
|
(let* ((url (format nil "https://api.telegram.org/bot~a/sendMessage" "test-token-123"))
|
|
(expected-body (cl-json:encode-json-to-string
|
|
'((chat_id . "999") (text . "Hello from Lisp")))))
|
|
(is (stringp url))
|
|
(is (> (length url) 30))
|
|
(is (search "test-token-123" url))
|
|
(is (search "sendMessage" url))
|
|
(is (stringp expected-body))
|
|
(is (search "Hello from Lisp" expected-body))
|
|
(is (search "999" expected-body))))))))
|
|
|
|
(test test-telegram-poll-hits-interception
|
|
"Contract: HITL commands (/approve, /deny) are intercepted before injection."
|
|
(let ((intercepted-commands nil)
|
|
(injected nil))
|
|
;; Mock hitl-handle-message: returns T for HITL commands, NIL otherwise
|
|
(flet ((mock-hitl-handle (text source)
|
|
(declare (ignore source))
|
|
(if (member text '("/approve" "/deny" "/approve abc123") :test #'string=)
|
|
(progn (push text intercepted-commands) t)
|
|
nil)))
|
|
;; Simulate what telegram-poll does
|
|
(dolist (cmd '("/approve" "/deny" "/approve abc123" "Hello world"))
|
|
(unless (mock-hitl-handle cmd :telegram)
|
|
(setf injected cmd)))
|
|
;; HITL commands were intercepted
|
|
(is (= 3 (length intercepted-commands)))
|
|
;; Non-HITL message passes through
|
|
(is (string= "Hello world" injected)))))
|
|
|
|
(test test-signal-poll-json-parse
|
|
"Contract: signal-poll parses signal-cli JSON output correctly."
|
|
(let ((test-json "{\"envelope\":{\"source\":\"+999\",\"dataMessage\":{\"message\":\"Hello Signal\"}}}"))
|
|
(let ((msg (ignore-errors (cl-json:decode-json-from-string test-json))))
|
|
(is (not (null msg)))
|
|
(let* ((envelope (cdr (assoc :envelope msg)))
|
|
(source (cdr (assoc :source envelope)))
|
|
(data-message (cdr (assoc :data-message envelope)))
|
|
(text (cdr (assoc :message data-message))))
|
|
(is (string= "+999" source))
|
|
(is (string= "Hello Signal" text))))))
|