From db88fde5c9db805f88ce4f0debdc491868f53fe1 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 25 May 2026 15:22:16 +0000 Subject: [PATCH] =?UTF-8?q?auto-repair:=20commit=202=20uncommitted=20file(?= =?UTF-8?q?s)=20=E2=80=94=202026-05-25?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lib/symbiont/telepathy/smtp.ex | 2 +- lib/symbiont/telepathy/smtp.ex.bak | 303 +++++++++++++++++++++++++++++ 2 files changed, 304 insertions(+), 1 deletion(-) create mode 100644 lib/symbiont/telepathy/smtp.ex.bak diff --git a/lib/symbiont/telepathy/smtp.ex b/lib/symbiont/telepathy/smtp.ex index 72fb691..6c5da5c 100644 --- a/lib/symbiont/telepathy/smtp.ex +++ b/lib/symbiont/telepathy/smtp.ex @@ -68,7 +68,7 @@ defmodule Symbiont.Telepathy.SMTP do port = Keyword.get(opts, :smtp_port, 25) 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}] ]) end diff --git a/lib/symbiont/telepathy/smtp.ex.bak b/lib/symbiont/telepathy/smtp.ex.bak new file mode 100644 index 0000000..72fb691 --- /dev/null +++ b/lib/symbiont/telepathy/smtp.ex.bak @@ -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