Compare commits

...

2 Commits

Author SHA1 Message Date
root
db88fde5c9 auto-repair: commit 2 uncommitted file(s) — 2026-05-25 2026-05-25 15:22:16 +00:00
Claude Opus 4.6
304e9f1bbb deps: lock gen_smtp 1.3.0 + ranch 2.2.0 for SMTP inbound receiver
mix.lock was not updated when gen_smtp was added to mix.exs in the
auto-repair commit (7706a40). Without this, `mix deps.get` would fail
or produce a different resolution at build time.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-23 06:05:31 +00:00
3 changed files with 306 additions and 1 deletions

View File

@ -68,7 +68,7 @@ defmodule Symbiont.Telepathy.SMTP do
port = Keyword.get(opts, :smtp_port, 25) port = Keyword.get(opts, :smtp_port, 25)
Logger.info("Telepathy SMTP: starting listener on port #{port}") Logger.info("Telepathy SMTP: starting listener on port #{port}")
:gen_smtp_server.start_link(__MODULE__, [ :gen_smtp_server.start(__MODULE__, [
[port: port, domain: @domain, address: {0, 0, 0, 0}] [port: port, domain: @domain, address: {0, 0, 0, 0}]
]) ])
end end

View File

@ -0,0 +1,303 @@
defmodule Symbiont.Telepathy.SMTP do
@moduledoc """
Muse Inbox SMTP receiver gen_smtp_server_session behaviour.
Listens on port 25 (configurable via `:smtp_port`), accepts mail addressed
to `muse@hydrascale.net` or `muse@cortex.hydrascale.net`, and hands off
each accepted message to `Symbiont.Telepathy.Pipeline` for async processing.
## Data flow
SMTP client
handle_RCPT/2 validate recipient
handle_DATA/4 parse MIME via :mimemail, log to inbox.jsonl,
spawn Pipeline task via Task.Supervisor
"250 OK" response returned to sender immediately
Processing (dispatch to Claude, send reply, mark read) happens asynchronously
under `Symbiont.TaskSupervisor` so the SMTP response is never delayed by
upstream latency. See `Symbiont.Telepathy.Pipeline` for the pipeline logic.
## Supervision
`Symbiont.Telepathy.Supervisor` starts this module as a `:supervisor`-type
child (gen_smtp_server is itself a supervisor internally).
## Inbox log
Every accepted message is appended to `/data/muse/inbox.jsonl` in the same
schema used by the Python `smtp_handler.py`, ensuring backward compatibility
with any tooling that reads that file. The `processed` flag is flipped to
`true` by `Pipeline` after successful dispatch.
"""
@behaviour :gen_smtp_server_session
require Logger
@accepted_recipients ["muse@hydrascale.net", "muse@cortex.hydrascale.net"]
@inbox_log "/data/muse/inbox.jsonl"
@domain "cortex.hydrascale.net"
defstruct from: nil, rcpt_to: []
# ---------------------------------------------------------------------------
# Supervision interface
# ---------------------------------------------------------------------------
@doc """
Returns a child spec suitable for `Supervisor.start_link/2`.
Passes `:smtp_port` through to `start_link/1`.
"""
def child_spec(opts) do
%{
id: __MODULE__,
start: {__MODULE__, :start_link, [opts]},
type: :supervisor,
restart: :permanent
}
end
@doc """
Starts the gen_smtp_server listener process.
`opts` must include `:smtp_port` (integer). Falls back to 25 if absent.
"""
def start_link(opts) do
port = Keyword.get(opts, :smtp_port, 25)
Logger.info("Telepathy SMTP: starting listener on port #{port}")
:gen_smtp_server.start_link(__MODULE__, [
[port: port, domain: @domain, address: {0, 0, 0, 0}]
])
end
# ---------------------------------------------------------------------------
# gen_smtp_server_session callbacks
# ---------------------------------------------------------------------------
@impl :gen_smtp_server_session
def init(_hostname, _session_count, _peer_info, _options) do
banner = "#{@domain} ESMTP Muse Inbox ready"
{:ok, banner, %__MODULE__{}}
end
@impl :gen_smtp_server_session
def handle_HELO(_hostname, state) do
{:ok, state}
end
@impl :gen_smtp_server_session
def handle_EHLO(_hostname, extensions, state) do
# Return the default extension list unchanged. SIZE and 8BITMIME are
# included by gen_smtp by default; we don't advertise AUTH.
{:ok, extensions, state}
end
@impl :gen_smtp_server_session
def handle_MAIL(from, state) do
{:ok, %{state | from: from}}
end
@impl :gen_smtp_server_session
def handle_MAIL_extension(_extension, state) do
{:ok, state}
end
@impl :gen_smtp_server_session
def handle_RCPT(to, state) do
addr =
to
|> to_string()
|> String.trim()
|> String.trim_leading("<")
|> String.trim_trailing(">")
|> String.downcase()
if addr in @accepted_recipients do
{:ok, %{state | rcpt_to: [to | state.rcpt_to]}}
else
Logger.info("Telepathy SMTP: rejected recipient #{addr}")
{:error, "550 Recipient #{addr} not accepted", state}
end
end
@impl :gen_smtp_server_session
def handle_RCPT_extension(_extension, state) do
{:ok, state}
end
@doc """
Core DATA handler. Called once the entire message has been received.
1. Parses the raw SMTP payload via `:mimemail.decode/1`.
2. Appends a raw entry (processed: false) to `inbox.jsonl`.
3. Fires a `Symbiont.Telepathy.Pipeline.process/4` task under
`Symbiont.TaskSupervisor` returns 250 immediately without waiting.
"""
@impl :gen_smtp_server_session
def handle_DATA(from, to, data, state) do
case parse_email(data) do
{:ok, parsed} ->
entry = %{
"timestamp" => DateTime.utc_now() |> DateTime.to_iso8601(),
"from" => parsed.from,
"to" => Enum.join(to, ", "),
"subject" => parsed.subject,
"message_id" => parsed.message_id,
"body" => String.slice(parsed.body, 0, 5000),
"processed" => false
}
append_inbox_log(entry)
Task.Supervisor.start_child(Symbiont.TaskSupervisor, fn ->
Symbiont.Telepathy.Pipeline.process(
parsed.from,
parsed.subject,
parsed.body,
parsed.message_id
)
end)
ref = :crypto.strong_rand_bytes(4) |> Base.encode16(case: :lower)
Logger.info("Telepathy SMTP: accepted #{ref} from #{from} (#{parsed.subject})")
{:ok, ref, state}
{:error, reason} ->
Logger.error("Telepathy SMTP: parse error from #{from}: #{inspect(reason)}")
{:error, "451 Error processing message", state}
end
end
@impl :gen_smtp_server_session
def handle_RSET(state) do
{:ok, %{state | from: nil, rcpt_to: []}}
end
@impl :gen_smtp_server_session
def handle_VRFY(_address, state) do
{:error, "252 Cannot VRFY user, but will accept message and attempt delivery", state}
end
@impl :gen_smtp_server_session
def handle_other(_verb, _args, state) do
{"500 Command unrecognized", state}
end
@impl :gen_smtp_server_session
def handle_AUTH(_type, _username, _password, state) do
{:error, state}
end
@impl :gen_smtp_server_session
def handle_STARTTLS(state) do
{:ok, state}
end
@impl :gen_smtp_server_session
def code_change(_old_vsn, state, _extra) do
{:ok, state}
end
@impl :gen_smtp_server_session
def terminate(reason, state) do
if reason not in [:normal, :shutdown] do
Logger.debug("Telepathy SMTP: session terminated (#{inspect(reason)})")
end
{:ok, state}
end
# ---------------------------------------------------------------------------
# Private: MIME parsing via :mimemail
# ---------------------------------------------------------------------------
defp parse_email(data) do
try do
{type, subtype, headers, _params, body} = :mimemail.decode(data)
from = find_header(headers, "From") || "unknown"
subject = find_header(headers, "Subject") || "(no subject)"
message_id = find_header(headers, "Message-ID") || ""
text_body = extract_body(type, subtype, headers, body)
{:ok, %{from: from, subject: subject, message_id: message_id, body: text_body}}
rescue
e -> {:error, Exception.message(e)}
catch
kind, value -> {:error, {kind, value}}
end
end
# Plain text — use directly.
defp extract_body("text", "plain", _headers, body) when is_binary(body) do
String.trim(body)
end
# HTML with no plain-text alternative — return a placeholder matching the
# Python smtp_handler.py behaviour.
defp extract_body("text", "html", headers, _body) do
from = find_header(headers, "From") || "sender"
"[HTML content from #{from}]"
end
# Multipart — walk parts, prefer text/plain, fall back to HTML placeholder.
defp extract_body("multipart", _subtype, _headers, parts) when is_list(parts) do
plain = find_part(parts, "text", "plain")
html = find_part(parts, "text", "html")
cond do
plain != nil ->
{_, _, _, _, body} = plain
String.trim(body)
html != nil ->
{_, _, part_headers, _, _} = html
from = find_header(part_headers, "From") || "sender"
"[HTML content from #{from}]"
true ->
""
end
end
defp extract_body(_type, _subtype, _headers, body) when is_binary(body) do
String.trim(body)
end
defp extract_body(_type, _subtype, _headers, _body), do: ""
defp find_part(parts, type, subtype) do
Enum.find(parts, fn
{^type, ^subtype, _, _, _} -> true
_ -> false
end)
end
defp find_header(headers, name) do
name_lower = String.downcase(name)
Enum.find_value(headers, nil, fn
{n, v} ->
n_str = to_string(n)
if String.downcase(n_str) == name_lower, do: to_string(v), else: nil
_ ->
nil
end)
end
# ---------------------------------------------------------------------------
# Private: inbox.jsonl logging
# ---------------------------------------------------------------------------
defp append_inbox_log(entry) do
File.mkdir_p!(Path.dirname(@inbox_log))
File.write!(@inbox_log, Jason.encode!(entry) <> "\n", [:append])
rescue
e -> Logger.warning("Telepathy SMTP: inbox log write failed: #{Exception.message(e)}")
end
end

View File

@ -4,11 +4,13 @@
"db_connection": {:hex, :db_connection, "2.9.0", "a6a97c5c958a2d7091a58a9be40caf41ab496b0701d21e1d1abff3fa27a7f371", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "17d502eacaf61829db98facf6f20808ed33da6ccf495354a41e64fe42f9c509c"}, "db_connection": {:hex, :db_connection, "2.9.0", "a6a97c5c958a2d7091a58a9be40caf41ab496b0701d21e1d1abff3fa27a7f371", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "17d502eacaf61829db98facf6f20808ed33da6ccf495354a41e64fe42f9c509c"},
"elixir_make": {:hex, :elixir_make, "0.9.0", "6484b3cd8c0cee58f09f05ecaf1a140a8c97670671a6a0e7ab4dc326c3109726", [:mix], [], "hexpm", "db23d4fd8b757462ad02f8aa73431a426fe6671c80b200d9710caf3d1dd0ffdb"}, "elixir_make": {:hex, :elixir_make, "0.9.0", "6484b3cd8c0cee58f09f05ecaf1a140a8c97670671a6a0e7ab4dc326c3109726", [:mix], [], "hexpm", "db23d4fd8b757462ad02f8aa73431a426fe6671c80b200d9710caf3d1dd0ffdb"},
"exqlite": {:hex, :exqlite, "0.35.0", "90741471945db42b66cd8ca3149af317f00c22c769cc6b06e8b0a08c5924aae5", [:make, :mix], [{:cc_precompiler, "~> 0.1", [hex: :cc_precompiler, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:elixir_make, "~> 0.8", [hex: :elixir_make, repo: "hexpm", optional: false]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "a009e303767a28443e546ac8aab2539429f605e9acdc38bd43f3b13f1568bca9"}, "exqlite": {:hex, :exqlite, "0.35.0", "90741471945db42b66cd8ca3149af317f00c22c769cc6b06e8b0a08c5924aae5", [:make, :mix], [{:cc_precompiler, "~> 0.1", [hex: :cc_precompiler, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:elixir_make, "~> 0.8", [hex: :elixir_make, repo: "hexpm", optional: false]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "a009e303767a28443e546ac8aab2539429f605e9acdc38bd43f3b13f1568bca9"},
"gen_smtp": {:hex, :gen_smtp, "1.3.0", "62c3d91f0dcf6ce9db71bcb6881d7ad0d1d834c7f38c13fa8e952f4104a8442e", [:rebar3], [{:ranch, ">= 1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "0b73fbf069864ecbce02fe653b16d3f35fd889d0fdd4e14527675565c39d84e6"},
"hpax": {:hex, :hpax, "1.0.3", "ed67ef51ad4df91e75cc6a1494f851850c0bd98ebc0be6e81b026e765ee535aa", [:mix], [], "hexpm", "8eab6e1cfa8d5918c2ce4ba43588e894af35dbd8e91e6e55c817bca5847df34a"}, "hpax": {:hex, :hpax, "1.0.3", "ed67ef51ad4df91e75cc6a1494f851850c0bd98ebc0be6e81b026e765ee535aa", [:mix], [], "hexpm", "8eab6e1cfa8d5918c2ce4ba43588e894af35dbd8e91e6e55c817bca5847df34a"},
"jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"}, "jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"},
"mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"}, "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"},
"plug": {:hex, :plug, "1.19.1", "09bac17ae7a001a68ae393658aa23c7e38782be5c5c00c80be82901262c394c0", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "560a0017a8f6d5d30146916862aaf9300b7280063651dd7e532b8be168511e62"}, "plug": {:hex, :plug, "1.19.1", "09bac17ae7a001a68ae393658aa23c7e38782be5c5c00c80be82901262c394c0", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "560a0017a8f6d5d30146916862aaf9300b7280063651dd7e532b8be168511e62"},
"plug_crypto": {:hex, :plug_crypto, "2.1.1", "19bda8184399cb24afa10be734f84a16ea0a2bc65054e23a62bb10f06bc89491", [:mix], [], "hexpm", "6470bce6ffe41c8bd497612ffde1a7e4af67f36a15eea5f921af71cf3e11247c"}, "plug_crypto": {:hex, :plug_crypto, "2.1.1", "19bda8184399cb24afa10be734f84a16ea0a2bc65054e23a62bb10f06bc89491", [:mix], [], "hexpm", "6470bce6ffe41c8bd497612ffde1a7e4af67f36a15eea5f921af71cf3e11247c"},
"ranch": {:hex, :ranch, "2.2.0", "25528f82bc8d7c6152c57666ca99ec716510fe0925cb188172f41ce93117b1b0", [:make, :rebar3], [], "hexpm", "fa0b99a1780c80218a4197a59ea8d3bdae32fbff7e88527d7d8a4787eff4f8e7"},
"telemetry": {:hex, :telemetry, "1.4.1", "ab6de178e2b29b58e8256b92b382ea3f590a47152ca3651ea857a6cae05ac423", [:rebar3], [], "hexpm", "2172e05a27531d3d31dd9782841065c50dd5c3c7699d95266b2edd54c2dafa1c"}, "telemetry": {:hex, :telemetry, "1.4.1", "ab6de178e2b29b58e8256b92b382ea3f590a47152ca3651ea857a6cae05ac423", [:rebar3], [], "hexpm", "2172e05a27531d3d31dd9782841065c50dd5c3c7699d95266b2edd54c2dafa1c"},
"thousand_island": {:hex, :thousand_island, "1.4.3", "2158209580f633be38d43ec4e3ce0a01079592b9657afff9080d5d8ca149a3af", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "6e4ce09b0fd761a58594d02814d40f77daff460c48a7354a15ab353bb998ea0b"}, "thousand_island": {:hex, :thousand_island, "1.4.3", "2158209580f633be38d43ec4e3ce0a01079592b9657afff9080d5d8ca149a3af", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "6e4ce09b0fd761a58594d02814d40f77daff460c48a7354a15ab353bb998ea0b"},
"websock": {:hex, :websock, "0.5.3", "2f69a6ebe810328555b6fe5c831a851f485e303a7c8ce6c5f675abeb20ebdadc", [:mix], [], "hexpm", "6105453d7fac22c712ad66fab1d45abdf049868f253cf719b625151460b8b453"}, "websock": {:hex, :websock, "0.5.3", "2f69a6ebe810328555b6fe5c831a851f485e303a7c8ce6c5f675abeb20ebdadc", [:mix], [], "hexpm", "6105453d7fac22c712ad66fab1d45abdf049868f253cf719b625151460b8b453"},