auto-repair: commit 7 uncommitted file(s) — 2026-05-23
This commit is contained in:
root
parent
cd553883ed
commit
7706a4047c
@ -10,7 +10,8 @@ config :symbiont,
|
||||
|
||||
config :symbiont, :telepathy,
|
||||
enabled: true,
|
||||
poll_interval_ms: 60_000
|
||||
poll_interval_ms: 60_000,
|
||||
smtp_port: 25
|
||||
|
||||
config :logger,
|
||||
level: :info
|
||||
|
||||
@ -14,4 +14,8 @@ if config_env() != :test do
|
||||
if System.get_env("TELEPATHY_ENABLED") == "false" do
|
||||
config :symbiont, :telepathy, enabled: false
|
||||
end
|
||||
|
||||
if smtp_port = System.get_env("SMTP_PORT") do
|
||||
config :symbiont, :telepathy, smtp_port: String.to_integer(smtp_port)
|
||||
end
|
||||
end
|
||||
|
||||
@ -11,7 +11,8 @@ defmodule Symbiont.Application do
|
||||
├── Symbiont.Heartbeat — periodic health checks + queue processing
|
||||
├── Symbiont.Telepathy.Supervisor — email communication layer (if enabled)
|
||||
│ ├── Symbiont.Telepathy.MessageStore — JSONL-backed message persistence
|
||||
│ └── Symbiont.Telepathy.JMAP — Fastmail JMAP client + polling
|
||||
│ ├── Symbiont.Telepathy.JMAP — Fastmail JMAP client + polling
|
||||
│ └── Symbiont.Telepathy.SMTP — gen_smtp listener on port 25
|
||||
└── Bandit (Symbiont.API) — HTTP API
|
||||
|
||||
In test mode, the supervisor starts empty — tests manage their own processes.
|
||||
@ -65,10 +66,11 @@ defmodule Symbiont.Application do
|
||||
telepathy_cfg = Application.get_env(:symbiont, :telepathy, [])
|
||||
telepathy_enabled = Keyword.get(telepathy_cfg, :enabled, true)
|
||||
poll_ms = Keyword.get(telepathy_cfg, :poll_interval_ms, 60_000)
|
||||
smtp_port = Keyword.get(telepathy_cfg, :smtp_port, 25)
|
||||
|
||||
telepathy_children =
|
||||
if telepathy_enabled do
|
||||
[{Symbiont.Telepathy.Supervisor, poll_interval_ms: poll_ms}]
|
||||
[{Symbiont.Telepathy.Supervisor, poll_interval_ms: poll_ms, smtp_port: smtp_port}]
|
||||
else
|
||||
[]
|
||||
end
|
||||
|
||||
309
lib/symbiont/telepathy/pipeline.ex
Normal file
309
lib/symbiont/telepathy/pipeline.ex
Normal file
@ -0,0 +1,309 @@
|
||||
defmodule Symbiont.Telepathy.Pipeline do
|
||||
@moduledoc """
|
||||
Inbound email processing pipeline.
|
||||
|
||||
Called from `Symbiont.Telepathy.SMTP.handle_DATA/4` via a
|
||||
`Task.Supervisor` task. Runs entirely asynchronously — the SMTP "250 OK"
|
||||
has already been sent before this module is invoked.
|
||||
|
||||
## Pipeline steps
|
||||
|
||||
1. **Store** — Persist the inbound email in `MessageStore` (messages.jsonl).
|
||||
2. **Dispatch** — POST to the Symbiont `/task` endpoint with the full email
|
||||
context and a prompt asking Muse to respond. Retries on transient errors
|
||||
using an exponential backoff schedule of 10s → 30s → 90s (4 total attempts).
|
||||
3. **Reply** — Send the response via `Symbiont.Telepathy.JMAP.send_email/3`.
|
||||
4. **Mark read** — Flip `processed: true` in `inbox.jsonl` and `read: true`
|
||||
in `MessageStore`.
|
||||
|
||||
## Error handling
|
||||
|
||||
Transient failures (connection errors, HTTP 5xx) are retried. Permanent
|
||||
failures (HTTP 4xx, empty replies) stop retrying immediately. After all
|
||||
attempts are exhausted, the `inbox.jsonl` entry is left with `processed: false`
|
||||
so it can be manually re-dispatched.
|
||||
|
||||
## Prompt fidelity
|
||||
|
||||
The prompt sent to `/task` is identical to the one used by the Python
|
||||
`smtp_handler.py`, including the optional `context.md` prefix. This ensures
|
||||
consistent Muse behaviour during the Python-to-Elixir migration period.
|
||||
"""
|
||||
|
||||
require Logger
|
||||
|
||||
@symbiont_url "http://127.0.0.1:8111"
|
||||
@inbox_log "/data/muse/inbox.jsonl"
|
||||
@context_path "/data/muse/context.md"
|
||||
|
||||
# Retry schedule in milliseconds: attempt 1 → immediate, 2 → 10s, 3 → 30s, 4 → 90s
|
||||
@backoff_schedule_ms [10_000, 30_000, 90_000]
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public API
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@doc """
|
||||
Run the full inbound email pipeline.
|
||||
|
||||
Called by `Symbiont.Telepathy.SMTP` under `Symbiont.TaskSupervisor`.
|
||||
Returns `:ok` on success, `:error` on terminal failure.
|
||||
"""
|
||||
def process(from_addr, subject, body, message_id) do
|
||||
Logger.info("Pipeline: processing email from #{from_addr} (#{subject})")
|
||||
|
||||
# Step 1: store in MessageStore
|
||||
telepathy_id = store_message(from_addr, subject, body)
|
||||
|
||||
# Step 2: dispatch to Symbiont with retry
|
||||
case dispatch_with_retry(from_addr, subject, body, message_id) do
|
||||
{:ok, reply_text} ->
|
||||
send_and_mark(from_addr, subject, reply_text, message_id, telepathy_id)
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.error(
|
||||
"Pipeline: dispatch exhausted for #{from_addr} " <>
|
||||
"(subject=#{inspect(subject)}): #{inspect(reason)}"
|
||||
)
|
||||
|
||||
:error
|
||||
end
|
||||
end
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Step 1: MessageStore
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
defp store_message(from_addr, subject, body) do
|
||||
msg = %{
|
||||
"content" => "From: #{from_addr}\nSubject: #{subject}\n\n#{String.slice(body, 0, 2000)}",
|
||||
"source" => "email-in",
|
||||
"subject" => "[Inbound] #{subject}"
|
||||
}
|
||||
|
||||
case Symbiont.Telepathy.MessageStore.store(msg) do
|
||||
{:ok, id} -> id
|
||||
_ -> nil
|
||||
end
|
||||
end
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Step 2: Symbiont dispatch with backoff retry
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
defp dispatch_with_retry(from_addr, subject, body, message_id) do
|
||||
do_dispatch(from_addr, subject, body, message_id, @backoff_schedule_ms, 1)
|
||||
end
|
||||
|
||||
# No more backoff slots — last attempt.
|
||||
defp do_dispatch(from_addr, subject, body, message_id, [], attempt) do
|
||||
case dispatch_to_symbiont(from_addr, subject, body) do
|
||||
{:ok, reply} ->
|
||||
{:ok, reply}
|
||||
|
||||
{_, detail} ->
|
||||
Logger.error(
|
||||
"Pipeline: giving up after #{attempt} attempt(s) for #{from_addr} " <>
|
||||
"(message_id=#{inspect(message_id)}): #{detail}"
|
||||
)
|
||||
|
||||
{:error, detail}
|
||||
end
|
||||
end
|
||||
|
||||
defp do_dispatch(from_addr, subject, body, message_id, [delay | rest], attempt) do
|
||||
case dispatch_to_symbiont(from_addr, subject, body) do
|
||||
{:ok, reply} ->
|
||||
{:ok, reply}
|
||||
|
||||
{:permanent, detail} ->
|
||||
Logger.error(
|
||||
"Pipeline: permanent failure for #{from_addr} " <>
|
||||
"(message_id=#{inspect(message_id)}): #{detail}"
|
||||
)
|
||||
|
||||
{:error, :permanent}
|
||||
|
||||
{:transient, detail} ->
|
||||
Logger.warning(
|
||||
"Pipeline: transient failure (attempt #{attempt}) for #{from_addr}: " <>
|
||||
"#{detail}. Retrying in #{div(delay, 1000)}s..."
|
||||
)
|
||||
|
||||
Process.sleep(delay)
|
||||
do_dispatch(from_addr, subject, body, message_id, rest, attempt + 1)
|
||||
end
|
||||
end
|
||||
|
||||
defp dispatch_to_symbiont(from_addr, subject, body) do
|
||||
context = load_context()
|
||||
prompt = build_prompt(context, from_addr, subject, body)
|
||||
payload = Jason.encode!(%{"task" => prompt, "force_tier" => "sonnet"})
|
||||
|
||||
url = ~c"#{@symbiont_url}/task"
|
||||
headers = [{~c"Content-Type", ~c"application/json"}]
|
||||
|
||||
case :httpc.request(
|
||||
:post,
|
||||
{url, headers, ~c"application/json", payload},
|
||||
[{:timeout, 600_000}, {:connect_timeout, 10_000}],
|
||||
[]
|
||||
) do
|
||||
{:ok, {{_, 200, _}, _, resp_body}} ->
|
||||
parse_symbiont_response(resp_body)
|
||||
|
||||
{:ok, {{_, status, _}, _, resp_body}} when status >= 500 ->
|
||||
detail = resp_body |> IO.iodata_to_binary() |> String.slice(0, 200)
|
||||
{:transient, "HTTP #{status}: #{detail}"}
|
||||
|
||||
{:ok, {{_, status, _}, _, resp_body}} ->
|
||||
detail = resp_body |> IO.iodata_to_binary() |> String.slice(0, 200)
|
||||
{:permanent, "HTTP #{status}: #{detail}"}
|
||||
|
||||
{:error, {:failed_connect, _}} ->
|
||||
{:transient, "connection refused to #{@symbiont_url}"}
|
||||
|
||||
{:error, reason} ->
|
||||
{:transient, "httpc error: #{inspect(reason)}"}
|
||||
end
|
||||
end
|
||||
|
||||
defp parse_symbiont_response(resp_body) do
|
||||
case Jason.decode(IO.iodata_to_binary(resp_body)) do
|
||||
{:ok, %{"result" => reply}} when is_binary(reply) and reply != "" ->
|
||||
{:ok, String.trim(reply)}
|
||||
|
||||
{:ok, _} ->
|
||||
{:permanent, "empty or missing 'result' in Symbiont response"}
|
||||
|
||||
{:error, _} ->
|
||||
{:permanent, "invalid JSON from Symbiont"}
|
||||
end
|
||||
end
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Step 3 + 4: Reply and mark read
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
defp send_and_mark(from_addr, subject, reply_text, message_id, telepathy_id) do
|
||||
re_subject = if String.starts_with?(subject, "Re:"), do: subject, else: "Re: #{subject}"
|
||||
reply_to = extract_email_addr(from_addr)
|
||||
|
||||
case Symbiont.Telepathy.JMAP.send_email(reply_to, re_subject, reply_text) do
|
||||
{:ok, _email_id} ->
|
||||
Logger.info("Pipeline: reply sent to #{reply_to} (#{re_subject})")
|
||||
mark_inbox_processed(message_id)
|
||||
if telepathy_id, do: Symbiont.Telepathy.MessageStore.mark_read(telepathy_id)
|
||||
:ok
|
||||
|
||||
{:error, reason} ->
|
||||
Logger.error("Pipeline: JMAP send failed for #{from_addr}: #{inspect(reason)}")
|
||||
:error
|
||||
end
|
||||
end
|
||||
|
||||
# Extract bare email from "Display Name <addr@domain>" or plain "addr@domain".
|
||||
defp extract_email_addr(addr) do
|
||||
case Regex.run(~r/<([^>]+)>/, addr) do
|
||||
[_, email] -> String.trim(email)
|
||||
_ -> String.trim(addr)
|
||||
end
|
||||
end
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# inbox.jsonl: flip processed flag after successful dispatch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
defp mark_inbox_processed(message_id) when is_binary(message_id) and message_id != "" do
|
||||
case File.read(@inbox_log) do
|
||||
{:ok, content} ->
|
||||
lines =
|
||||
content
|
||||
|> String.split("\n", trim: true)
|
||||
|> Enum.map(&update_processed_flag(&1, message_id))
|
||||
|
||||
tmp = @inbox_log <> ".tmp"
|
||||
File.write!(tmp, Enum.join(lines, "\n") <> "\n")
|
||||
File.rename!(tmp, @inbox_log)
|
||||
Logger.info("Pipeline: marked inbox entry processed: #{message_id}")
|
||||
|
||||
{:error, _} ->
|
||||
:ok
|
||||
end
|
||||
rescue
|
||||
e -> Logger.warning("Pipeline: mark_inbox_processed failed: #{Exception.message(e)}")
|
||||
end
|
||||
|
||||
defp mark_inbox_processed(_), do: :ok
|
||||
|
||||
defp update_processed_flag(line, message_id) do
|
||||
case Jason.decode(line) do
|
||||
{:ok, entry} when is_map(entry) ->
|
||||
if entry["message_id"] == message_id and entry["processed"] == false do
|
||||
entry
|
||||
|> Map.put("processed", true)
|
||||
|> Map.put("processed_at", DateTime.utc_now() |> DateTime.to_iso8601())
|
||||
|> Jason.encode!()
|
||||
else
|
||||
line
|
||||
end
|
||||
|
||||
_ ->
|
||||
line
|
||||
end
|
||||
end
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Prompt construction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
defp load_context do
|
||||
case File.read(@context_path) do
|
||||
{:ok, content} -> content
|
||||
{:error, _} -> ""
|
||||
end
|
||||
end
|
||||
|
||||
defp build_prompt(context, from_addr, subject, body) do
|
||||
"""
|
||||
#{context}
|
||||
|
||||
---
|
||||
|
||||
You received an email from Michael.
|
||||
|
||||
From: #{from_addr}
|
||||
Subject: #{subject}
|
||||
|
||||
#{body}
|
||||
|
||||
---
|
||||
|
||||
You are Muse. You have full system access on cortex via your tools (Bash, Read,
|
||||
Edit, Write, etc.). The dispatcher runs you under --dangerously-skip-permissions,
|
||||
so you do NOT need to ask for permission — just act.
|
||||
|
||||
Decide whether this email is:
|
||||
|
||||
(A) A REQUEST TO DO WORK (imperative): "please commit X", "fix the Y bug",
|
||||
"check Z and tell me", "write a script that…", "investigate…", etc.
|
||||
→ DO THE WORK NOW using your tools. Then write a brief reply summarizing
|
||||
what you actually did (file paths touched, commits made, what you found).
|
||||
Be specific: include real command output, real diffs, real numbers.
|
||||
Do NOT promise to do it later — the only "later" that exists is the next
|
||||
time he emails you. Either it gets done in this session or it doesn't.
|
||||
|
||||
(B) A CONVERSATIONAL MESSAGE or QUESTION: just write a reply, grounded in
|
||||
real system state (run quick checks with your tools if useful).
|
||||
|
||||
For both, the reply should:
|
||||
- Be concise (2-4 short paragraphs is usually right; longer if results demand it)
|
||||
- Use real data from your tools, never invented numbers or fake commit hashes
|
||||
- Sign off as "Muse"
|
||||
- Contain NO preamble, NO markdown fences, NO meta-commentary about what you're
|
||||
about to do — just the email body itself
|
||||
|
||||
Return ONLY the reply text.
|
||||
"""
|
||||
end
|
||||
end
|
||||
303
lib/symbiont/telepathy/smtp.ex
Normal file
303
lib/symbiont/telepathy/smtp.ex
Normal 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
|
||||
@ -2,9 +2,13 @@ defmodule Symbiont.Telepathy.Supervisor do
|
||||
@moduledoc """
|
||||
Supervision subtree for Telepathy (email communication layer).
|
||||
|
||||
Children:
|
||||
- MessageStore — JSONL-backed message persistence
|
||||
Children (start order matters — rest_for_one strategy):
|
||||
- MessageStore — JSONL-backed message persistence (must start first)
|
||||
- JMAP — Fastmail JMAP client with periodic inbox polling
|
||||
- SMTP — gen_smtp listener on port 25 (inbound email receiver)
|
||||
|
||||
The `rest_for_one` strategy ensures that if MessageStore crashes, JMAP and
|
||||
SMTP are also restarted (they depend on MessageStore being available).
|
||||
"""
|
||||
use Supervisor
|
||||
|
||||
@ -16,10 +20,12 @@ defmodule Symbiont.Telepathy.Supervisor do
|
||||
def init(opts) do
|
||||
messages_path = Keyword.get(opts, :messages_path, "/data/telepathy/messages.jsonl")
|
||||
poll_interval_ms = Keyword.get(opts, :poll_interval_ms, 60_000)
|
||||
smtp_port = Keyword.get(opts, :smtp_port, 25)
|
||||
|
||||
children = [
|
||||
{Symbiont.Telepathy.MessageStore, path: messages_path},
|
||||
{Symbiont.Telepathy.JMAP, poll_interval_ms: poll_interval_ms}
|
||||
{Symbiont.Telepathy.JMAP, poll_interval_ms: poll_interval_ms},
|
||||
{Symbiont.Telepathy.SMTP, smtp_port: smtp_port}
|
||||
]
|
||||
|
||||
Supervisor.init(children, strategy: :rest_for_one)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user