feat: initial Telepathy Elixir port — GenServer + JMAP fetch loop

Adds the Telepathy email communication layer as an OTP subtree:

- Telepathy.JMAP: GenServer that discovers Fastmail JMAP session,
  periodically polls inbox for new emails, and provides send_email/3.
  Uses :httpc (no external HTTP dep). Deduplicates via seen_ids MapSet
  seeded from existing MessageStore entries.

- Telepathy.MessageStore: JSONL-backed message persistence, compatible
  with the existing Python messages.jsonl format. Supports store, unread,
  mark_read, list, count operations.

- Telepathy.Supervisor: rest_for_one supervisor (MessageStore then JMAP).

- Conditionally enabled via config (:telepathy, :enabled). Disabled in
  test env. Can be disabled at runtime via TELEPATHY_ENABLED=false.

Validated against live Fastmail: session discovery, inbox fetch (20
emails), message storage all working.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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
 

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

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

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 = [
+    telepathy_cfg = Application.get_env(:symbiont, :telepathy, [])
-      {Task.Supervisor, name: Symbiont.TaskSupervisor},
+    telepathy_enabled = Keyword.get(telepathy_cfg, :enabled, true)
-      {Symbiont.Ledger, data_dir: data_dir},
+    poll_ms = Keyword.get(telepathy_cfg, :poll_interval_ms, 60_000)
-      {Symbiont.Queue, data_dir: data_dir},
+
-      {Symbiont.Engram, db_path: engram_db},
+    telepathy_children =
-      {Symbiont.Heartbeat, []},
+      if telepathy_enabled do
-      bandit_spec
+        [{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)

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

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

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