310 lines
10 KiB
Elixir
310 lines
10 KiB
Elixir
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
|