diff --git a/config/config.exs b/config/config.exs index 538e8d0..aaf86cf 100644 --- a/config/config.exs +++ b/config/config.exs @@ -8,6 +8,10 @@ config :symbiont, default_tier: :haiku, claude_cli: "claude" +config :symbiont, :telepathy, + enabled: true, + poll_interval_ms: 60_000 + config :logger, level: :info diff --git a/config/runtime.exs b/config/runtime.exs index 7f9a281..287bded 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -7,3 +7,7 @@ end if data_dir = System.get_env("SYMBIONT_DATA_DIR") do config :symbiont, data_dir: data_dir end + +if System.get_env("TELEPATHY_ENABLED") == "false" do + config :symbiont, :telepathy, enabled: false +end diff --git a/config/test.exs b/config/test.exs index 639ebe3..6fe1960 100644 --- a/config/test.exs +++ b/config/test.exs @@ -6,5 +6,8 @@ config :symbiont, heartbeat_interval_ms: :timer.hours(24), claude_cli: "echo" +config :symbiont, :telepathy, + enabled: false + config :logger, level: :warning diff --git a/lib/symbiont/application.ex b/lib/symbiont/application.ex index d25a346..48dd5ac 100644 --- a/lib/symbiont/application.ex +++ b/lib/symbiont/application.ex @@ -9,6 +9,9 @@ defmodule Symbiont.Application do ├── Symbiont.Queue — persistent task queue ├── Symbiont.Engram — cross-session memory (SQLite) ├── 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 └── Bandit (Symbiont.API) — HTTP API In test mode, the supervisor starts empty — tests manage their own processes. @@ -59,14 +62,25 @@ defmodule Symbiont.Application do shutdown: @bandit_child_shutdown_ms ) - children = [ - {Task.Supervisor, name: Symbiont.TaskSupervisor}, - {Symbiont.Ledger, data_dir: data_dir}, - {Symbiont.Queue, data_dir: data_dir}, - {Symbiont.Engram, db_path: engram_db}, - {Symbiont.Heartbeat, []}, - bandit_spec - ] + 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) + + telepathy_children = + if telepathy_enabled do + [{Symbiont.Telepathy.Supervisor, poll_interval_ms: poll_ms}] + else + [] + end + + children = + [ + {Task.Supervisor, name: Symbiont.TaskSupervisor}, + {Symbiont.Ledger, data_dir: data_dir}, + {Symbiont.Queue, data_dir: data_dir}, + {Symbiont.Engram, db_path: engram_db}, + {Symbiont.Heartbeat, []} + ] ++ telepathy_children ++ [bandit_spec] opts = [strategy: :rest_for_one, name: Symbiont.Supervisor] Supervisor.start_link(children, opts) diff --git a/lib/symbiont/telepathy/jmap.ex b/lib/symbiont/telepathy/jmap.ex new file mode 100644 index 0000000..1d0d70b --- /dev/null +++ b/lib/symbiont/telepathy/jmap.ex @@ -0,0 +1,484 @@ +defmodule Symbiont.Telepathy.JMAP do + @moduledoc """ + JMAP client GenServer for Fastmail (RFC 8620/8621). + + Discovers the JMAP session, then periodically polls for new inbound emails. + New emails are stored in the MessageStore. Also provides send_email/3 for + outbound mail. + + Uses :httpc from :inets (no external HTTP dependency). + """ + use GenServer + require Logger + + @jmap_session_url "https://api.fastmail.com/jmap/session" + @core_cap "urn:ietf:params:jmap:core" + @mail_cap "urn:ietf:params:jmap:mail" + @submit_cap "urn:ietf:params:jmap:submission" + + @default_poll_interval_ms 60_000 + @config_path "/data/telepathy/config.json" + + defstruct [ + :token, + :api_url, + :account_id, + :identity_id, + :from_email, + :drafts_id, + :inbox_id, + :poll_interval_ms, + seen_ids: MapSet.new() + ] + + # -- Client API -- + + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + def send_email(to, subject, body) do + GenServer.call(__MODULE__, {:send_email, to, subject, body}, 30_000) + end + + def session_info do + GenServer.call(__MODULE__, :session_info) + end + + def fetch_now do + send(__MODULE__, :poll) + :ok + end + + # -- Server -- + + @impl true + def init(opts) do + poll_ms = Keyword.get(opts, :poll_interval_ms, @default_poll_interval_ms) + token = load_token() + + state = %__MODULE__{ + token: token, + poll_interval_ms: poll_ms + } + + send(self(), :discover_session) + {:ok, state} + end + + @impl true + def handle_info(:discover_session, state) do + case discover_session(state.token) do + {:ok, session} -> + Logger.info("Telepathy JMAP session discovered: account=#{session.account_id}") + + new_state = %{ + state + | api_url: session.api_url, + account_id: session.account_id, + identity_id: session.identity_id, + from_email: session.from_email, + drafts_id: session.drafts_id, + inbox_id: session.inbox_id + } + + seen = seed_seen_ids() + schedule_poll(new_state.poll_interval_ms) + {:noreply, %{new_state | seen_ids: seen}} + + {:error, reason} -> + Logger.error("Telepathy JMAP session discovery failed: #{inspect(reason)}") + Process.send_after(self(), :discover_session, 10_000) + {:noreply, state} + end + end + + @impl true + def handle_info(:poll, %{api_url: nil} = state) do + schedule_poll(state.poll_interval_ms) + {:noreply, state} + end + + @impl true + def handle_info(:poll, state) do + new_state = + case fetch_inbox_emails(state) do + {:ok, emails} -> + {new_count, updated_seen} = store_new_emails(emails, state.seen_ids) + + if new_count > 0, + do: Logger.info("Telepathy: fetched #{new_count} new email(s)") + + %{state | seen_ids: updated_seen} + + {:error, reason} -> + Logger.warning("Telepathy JMAP poll failed: #{inspect(reason)}") + state + end + + schedule_poll(new_state.poll_interval_ms) + {:noreply, new_state} + end + + @impl true + def handle_call({:send_email, to, subject, body}, _from, state) do + result = do_send_email(state, to, subject, body) + {:reply, result, state} + end + + @impl true + def handle_call(:session_info, _from, state) do + info = %{ + api_url: state.api_url, + account_id: state.account_id, + identity_id: state.identity_id, + from_email: state.from_email, + inbox_id: state.inbox_id, + seen_count: MapSet.size(state.seen_ids) + } + + {:reply, info, state} + end + + # -- Private: Session Discovery -- + + defp discover_session(token) do + with {:ok, session_resp} <- jmap_get(@jmap_session_url, token), + api_url <- session_resp["apiUrl"], + account_id <- get_in(session_resp, ["primaryAccounts", @mail_cap]), + {:ok, details} <- fetch_identity_and_mailboxes(api_url, account_id, token) do + {:ok, Map.merge(%{api_url: api_url, account_id: account_id}, details)} + end + end + + defp fetch_identity_and_mailboxes(api_url, account_id, token) do + method_calls = [ + ["Identity/get", %{"accountId" => account_id, "ids" => nil}, "id0"], + ["Mailbox/get", %{"accountId" => account_id, "ids" => nil}, "mb0"] + ] + + case jmap_call(api_url, token, method_calls) do + {:ok, responses} -> + identity = extract_identity(responses) + {drafts_id, inbox_id} = extract_mailbox_ids(responses) + + {:ok, + %{ + identity_id: identity[:id], + from_email: identity[:email] || "mdwyer@michaelmdwyer.com", + drafts_id: drafts_id, + inbox_id: inbox_id + }} + + error -> + error + end + end + + defp extract_identity(responses) do + target = "mdwyer@michaelmdwyer.com" + + Enum.find_value(responses, %{id: nil, email: nil}, fn + ["Identity/get", data, _] -> + identities = data["list"] || [] + + match = + Enum.find(identities, fn i -> + String.downcase(i["email"] || "") == target + end) || List.first(identities) + + if match, do: %{id: match["id"], email: match["email"]}, else: %{id: nil, email: nil} + + _ -> + nil + end) + end + + defp extract_mailbox_ids(responses) do + Enum.find_value(responses, {nil, nil}, fn + ["Mailbox/get", data, _] -> + mailboxes = data["list"] || [] + drafts = Enum.find(mailboxes, &(&1["role"] == "drafts")) + inbox = Enum.find(mailboxes, &(&1["role"] == "inbox")) + {drafts && drafts["id"], inbox && inbox["id"]} + + _ -> + nil + end) + end + + # -- Private: Fetch Loop -- + + defp fetch_inbox_emails(state) do + method_calls = [ + [ + "Email/query", + %{ + "accountId" => state.account_id, + "filter" => %{"inMailbox" => state.inbox_id}, + "sort" => [%{"property" => "receivedAt", "isAscending" => false}], + "limit" => 20 + }, + "q0" + ], + [ + "Email/get", + %{ + "accountId" => state.account_id, + "#ids" => %{ + "resultOf" => "q0", + "name" => "Email/query", + "path" => "/ids" + }, + "properties" => [ + "id", + "from", + "to", + "subject", + "receivedAt", + "textBody", + "bodyValues", + "preview" + ], + "fetchTextBodyValues" => true + }, + "g0" + ] + ] + + case jmap_call(state.api_url, state.token, method_calls) do + {:ok, responses} -> + emails = + Enum.find_value(responses, [], fn + ["Email/get", data, _] -> data["list"] || [] + _ -> nil + end) || [] + + {:ok, emails} + + error -> + error + end + end + + defp store_new_emails(emails, seen_ids) do + new_emails = Enum.reject(emails, &MapSet.member?(seen_ids, &1["id"])) + + Enum.each(new_emails, fn email -> + body = extract_body(email) + from = extract_from(email) + subject = email["subject"] || "(no subject)" + + msg = %{ + "source" => "email-in", + "subject" => subject, + "content" => "[From: #{from}]\n#{body}", + "jmap_id" => email["id"], + "received_at" => email["receivedAt"] + } + + Symbiont.Telepathy.MessageStore.store(msg) + end) + + new_seen = + new_emails + |> Enum.map(& &1["id"]) + |> Enum.reduce(seen_ids, &MapSet.put(&2, &1)) + + {length(new_emails), new_seen} + end + + defp extract_body(email) do + body_values = email["bodyValues"] || %{} + text_body = email["textBody"] || [] + + case text_body do + [%{"partId" => part_id} | _] -> + get_in(body_values, [part_id, "value"]) || email["preview"] || "" + + _ -> + email["preview"] || "" + end + end + + defp extract_from(email) do + case email["from"] do + [%{"name" => name, "email" => addr} | _] when is_binary(name) and name != "" -> + "#{name} <#{addr}>" + + [%{"email" => addr} | _] -> + addr + + _ -> + "unknown" + end + end + + # -- Private: Send Email -- + + defp do_send_email(state, to, subject, body) do + method_calls = [ + [ + "Email/set", + %{ + "accountId" => state.account_id, + "create" => %{ + "e1" => %{ + "mailboxIds" => %{state.drafts_id => true}, + "from" => [%{"name" => "Muse", "email" => state.from_email}], + "to" => [%{"email" => to}], + "subject" => subject, + "replyTo" => [%{"name" => "Muse", "email" => "muse@hydrascale.net"}], + "keywords" => %{"$draft" => true}, + "bodyStructure" => %{"type" => "text/plain", "partId" => "body"}, + "bodyValues" => %{"body" => %{"value" => body, "charset" => "utf-8"}} + } + } + }, + "create-email" + ], + [ + "EmailSubmission/set", + %{ + "accountId" => state.account_id, + "create" => %{ + "s1" => %{ + "identityId" => state.identity_id, + "emailId" => "#e1" + } + } + }, + "send" + ] + ] + + case jmap_call(state.api_url, state.token, method_calls) do + {:ok, responses} -> + parse_send_response(responses) + + {:error, reason} -> + {:error, reason} + end + end + + defp parse_send_response(responses) do + Enum.reduce(responses, %{email_id: nil, sent: false, errors: []}, fn + ["Email/set", data, _], acc -> + case get_in(data, ["created", "e1", "id"]) do + nil -> + error = get_in(data, ["notCreated", "e1"]) + %{acc | errors: ["Email create: #{inspect(error)}" | acc.errors]} + + id -> + %{acc | email_id: id} + end + + ["EmailSubmission/set", data, _], acc -> + if get_in(data, ["created", "s1"]) do + %{acc | sent: true} + else + error = get_in(data, ["notCreated", "s1"]) + %{acc | errors: ["Submission: #{inspect(error)}" | acc.errors]} + end + + _, acc -> + acc + end) + |> then(fn + %{sent: true, email_id: id} -> {:ok, id} + %{errors: errors} -> {:error, Enum.join(errors, "; ")} + end) + end + + # -- Private: HTTP Helpers (using :httpc) -- + + defp jmap_get(url, token) do + headers = [{~c"Authorization", ~c"Bearer #{token}"}] + + case :httpc.request(:get, {String.to_charlist(url), headers}, [{:timeout, 15_000}], []) do + {:ok, {{_, 200, _}, _, body}} -> + Jason.decode(IO.iodata_to_binary(body)) + + {:ok, {{_, status, _}, _, body}} -> + {:error, {:http, status, IO.iodata_to_binary(body)}} + + {:error, reason} -> + {:error, reason} + end + end + + defp jmap_call(api_url, token, method_calls) do + payload = + Jason.encode!(%{ + "using" => [@core_cap, @mail_cap, @submit_cap], + "methodCalls" => method_calls + }) + + headers = [ + {~c"Authorization", ~c"Bearer #{token}"}, + {~c"Content-Type", ~c"application/json"} + ] + + case :httpc.request( + :post, + {String.to_charlist(api_url), headers, ~c"application/json", payload}, + [{:timeout, 30_000}], + [] + ) do + {:ok, {{_, 200, _}, _, body}} -> + case Jason.decode(IO.iodata_to_binary(body)) do + {:ok, %{"methodResponses" => responses}} -> {:ok, responses} + {:ok, other} -> {:error, {:unexpected_response, other}} + {:error, _} = err -> err + end + + {:ok, {{_, status, _}, _, body}} -> + {:error, {:http, status, IO.iodata_to_binary(body)}} + + {:error, reason} -> + {:error, reason} + end + end + + defp schedule_poll(interval_ms) do + Process.send_after(self(), :poll, interval_ms) + end + + defp load_token do + case System.get_env("FASTMAIL_API_TOKEN") do + nil -> load_token_from_config() + "" -> load_token_from_config() + token -> token + end + end + + defp load_token_from_config do + case File.read(@config_path) do + {:ok, content} -> + case Jason.decode(content) do + {:ok, %{"fastmail_token" => token}} + when is_binary(token) and token != "" and token != "PLACEHOLDER" -> + token + + _ -> + raise "No valid fastmail_token in #{@config_path}" + end + + {:error, _} -> + raise "Cannot read #{@config_path} — set FASTMAIL_API_TOKEN env var" + end + end + + defp seed_seen_ids do + case Symbiont.Telepathy.MessageStore.list(limit: 10_000) do + messages when is_list(messages) -> + messages + |> Enum.map(& &1["jmap_id"]) + |> Enum.reject(&is_nil/1) + |> MapSet.new() + + _ -> + MapSet.new() + end + end +end diff --git a/lib/symbiont/telepathy/message_store.ex b/lib/symbiont/telepathy/message_store.ex new file mode 100644 index 0000000..2fc7fe3 --- /dev/null +++ b/lib/symbiont/telepathy/message_store.ex @@ -0,0 +1,126 @@ +defmodule Symbiont.Telepathy.MessageStore do + @moduledoc """ + Persistent message store backed by JSONL, compatible with the Python + Telepathy messages.jsonl format. + + Message schema: + %{"id" => hex12, "timestamp" => iso8601, "source" => string, + "subject" => string, "content" => string, "read" => boolean} + """ + use GenServer + require Logger + + defstruct [:path, messages: []] + + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + def store(message) do + GenServer.call(__MODULE__, {:store, message}) + end + + def unread do + GenServer.call(__MODULE__, :unread) + end + + def mark_read(msg_id) do + GenServer.call(__MODULE__, {:mark_read, msg_id}) + end + + def list(opts \\ []) do + GenServer.call(__MODULE__, {:list, opts}) + end + + def count do + GenServer.call(__MODULE__, :count) + end + + # -- Server -- + + @impl true + def init(opts) do + path = Keyword.get(opts, :path, "/data/telepathy/messages.jsonl") + File.mkdir_p!(Path.dirname(path)) + unless File.exists?(path), do: File.write!(path, "") + + messages = load(path) + Logger.info("Telepathy MessageStore loaded: #{length(messages)} messages") + {:ok, %__MODULE__{path: path, messages: messages}} + end + + @impl true + def handle_call({:store, msg}, _from, state) do + entry = + Map.merge( + %{ + "id" => generate_id(), + "timestamp" => DateTime.utc_now() |> DateTime.to_iso8601(), + "read" => false + }, + msg + ) + + new_messages = state.messages ++ [entry] + append!(state.path, entry) + Logger.info("Telepathy message stored: #{entry["id"]} from #{entry["source"]}") + {:reply, {:ok, entry["id"]}, %{state | messages: new_messages}} + end + + @impl true + def handle_call(:unread, _from, state) do + unread = Enum.filter(state.messages, &(not &1["read"])) + {:reply, unread, state} + end + + @impl true + def handle_call({:mark_read, msg_id}, _from, state) do + new_messages = + Enum.map(state.messages, fn msg -> + if msg["id"] == msg_id, do: Map.put(msg, "read", true), else: msg + end) + + persist!(state.path, new_messages) + {:reply, :ok, %{state | messages: new_messages}} + end + + @impl true + def handle_call({:list, opts}, _from, state) do + limit = Keyword.get(opts, :limit, 50) + msgs = Enum.take(state.messages, -limit) + {:reply, msgs, state} + end + + @impl true + def handle_call(:count, _from, state) do + {:reply, length(state.messages), state} + end + + # -- Private -- + + defp load(path) do + path + |> File.stream!() + |> Stream.reject(&(&1 in ["", "\n"])) + |> Enum.map(fn line -> + case Jason.decode(String.trim(line)) do + {:ok, msg} -> msg + {:error, _} -> nil + end + end) + |> Enum.reject(&is_nil/1) + end + + defp append!(path, entry) do + File.write!(path, Jason.encode!(entry) <> "\n", [:append]) + end + + defp persist!(path, messages) do + content = Enum.map_join(messages, fn m -> Jason.encode!(m) <> "\n" end) + File.write!(path, content) + end + + defp generate_id do + :crypto.strong_rand_bytes(6) |> Base.encode16(case: :lower) + end +end diff --git a/lib/symbiont/telepathy/supervisor.ex b/lib/symbiont/telepathy/supervisor.ex new file mode 100644 index 0000000..ab5203a --- /dev/null +++ b/lib/symbiont/telepathy/supervisor.ex @@ -0,0 +1,27 @@ +defmodule Symbiont.Telepathy.Supervisor do + @moduledoc """ + Supervision subtree for Telepathy (email communication layer). + + Children: + - MessageStore — JSONL-backed message persistence + - JMAP — Fastmail JMAP client with periodic inbox polling + """ + use Supervisor + + def start_link(opts) do + Supervisor.start_link(__MODULE__, opts, name: __MODULE__) + end + + @impl true + 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) + + children = [ + {Symbiont.Telepathy.MessageStore, path: messages_path}, + {Symbiont.Telepathy.JMAP, poll_interval_ms: poll_interval_ms} + ] + + Supervisor.init(children, strategy: :rest_for_one) + end +end