feat(email): native SMTP-listener ingress (A3)

An opt-in receive-only SMTP listener (PICLOUD_SMTP_BIND) so an MX can point
straight at PiCloud. It speaks minimal SMTP (HELO/EHLO/MAIL/RCPT/DATA/RSET/
NOOP/QUIT), resolves each RCPT TO mailbox to the app-owned email trigger that
claims it, and inserts an Email outbox row the dispatcher fires — the same tail
as the HMAC webhook, unchanged.

- migration 0076: email_trigger_details.inbound_address (case-insensitively
  unique among app triggers) + TriggerRepo::email_inbound_target_by_address /
  SmtpInboundTarget; the interactive create-email API accepts inbound_address.
- crates/picloud/src/smtp.rs: a small tokio accept loop + session state machine
  + a testable deliver() core + a minimal RFC-5322 header/body split. DATA is
  size-capped with dot-unstuffing; no AUTH/STARTTLS (TLS terminates upstream —
  the recipient address + per-app isolation are the boundary).
- spawned in run_server alongside axum::serve, sharing the pool, on the same
  shutdown signal.

Pinned by a picloud integration test (deliver → one Email outbox row for a
known mailbox, none for an unknown one) + smtp.rs unit tests (address parse,
header/body split). Multipart/MIME decoding is a documented follow-up.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

CLAUDE.md

@@ -134,6 +134,7 @@ Environment variables consumed by the `picloud` binary:
 | `PICLOUD_DB_MAX_CONNECTIONS` | `32` | Postgres pool size. Matched to `PICLOUD_MAX_CONCURRENT_EXECUTIONS` so the data plane can't starve background workers. |
 | `PICLOUD_SESSION_TTL_HOURS` | `24` | Sliding-window admin session lifetime (idle timeout). |
 | `PICLOUD_SESSION_ABSOLUTE_TTL_HOURS` | `720` (30 days) | Absolute hard cap on an admin session's lifetime (audit 2026-07-11 C1). The sliding `touch` bump is clamped at `login_time + this`, so even a continuously-used or stolen-but-warm admin token self-expires. Mirrors the data-plane app-user session cap. |
+| `PICLOUD_SMTP_BIND` | — (off) | A3 native SMTP ingress. When set (e.g. `0.0.0.0:2525`), a receive-only SMTP listener binds alongside the HTTP server: it resolves each `RCPT TO` mailbox to the app-owned email trigger that claims it (`email_trigger_details.inbound_address`, unique) and writes an `Email` outbox row the dispatcher fires — same tail as the HMAC webhook. **No AUTH/STARTTLS at the listener** (TLS terminates upstream — Caddy/relay); the recipient address + per-app isolation are the boundary. `DATA` is size-capped. Unset = the webhook path only. Distinct from the outbound relay `PICLOUD_SMTP_*` vars (`SmtpConfig`). |
 | `PICLOUD_INTERCEPTOR_TIMEOUT_MS` | `5000` (5s) | §9.4 M5 default per-interceptor wall-clock timeout when a `[[interceptors]]` marker sets no `timeout_ms`. Bounds ONE hook run so a runaway guard (`loop {}`) is denied (fail closed) rather than hanging the guarded write. The effective deadline is always tightened to at most the caller's remaining budget — a hook can never EXTEND its caller's deadline. |
 | `PICLOUD_OUTBOX_CLAIM_TIMEOUT_SEC` | `600` (10 min) | How long a dispatcher may hold an OUTBOX claim before the reclaim task returns the row to the queue. **Without this a crash or restart mid-dispatch stranded every in-flight row permanently** — the dispatcher claims a row and only clears the claim on success (delete) or failure (reschedule), and `claim_due` selects `claimed_at IS NULL`, so a process that died in between left rows nothing would ever pick up again. Every other claim-based store (queue, group queue, workflow steps) already had a reclaimer; the outbox was the gap, and since it is the universal trigger path, the loss covered kv/docs/files/cron/pubsub/email/`invoke_async`/dead-letter alike. The default is deliberately generous: a script may run 300s, so a claim older than twice that is abandoned rather than slow. A reclaim does **not** bump `attempt_count` (the handler never ran — same reasoning as the transient queue `release`). Runs on the existing `PICLOUD_QUEUE_RECLAIM_INTERVAL_MS` ticker. |
 | `PICLOUD_REALTIME_BROADCAST_CAPACITY` | `64` | Per-channel SSE broadcast buffer depth (a slow consumer sees oldest events dropped). |
@@ -156,6 +157,6 @@ Environment variables consumed by the `picloud` binary:
 
 ## Out of MVP
 
-This section captured the original MVP cut. Most of it has since **shipped in v1.1.x**: queue triggers, cron triggers, inbound email (`email:receive`, HMAC-webhook model), KV / docs / email / users / HTTP SDKs, function-to-function `invoke()`, and secrets are all live (blueprint §12 Phase 4 table). The **Workflows** track (DAG + nested workflows) has since **shipped** (migrations `0071`/`0072`). The **§9.4 service-interceptor** track has since shipped in full (migrations `0073`–`0075`; before/after phases, data-transform, all six services, per-interceptor timeout, resolve cache, `pic interceptors ls`). A read-only **metrics/observability dashboard** has since shipped (A2): `GET /api/v1/admin/apps/{id}/metrics?window=` aggregates the existing `execution_logs` table (counts, error rate, latency avg/p50/p95, an hourly series) via `ExecutionLogRepository::summarize_for_app` + `metrics_api`, surfaced as the dashboard's per-app **Metrics** tab — no hot-path instrumentation. **Still deferred:** a raw SMTP-listener ingress, and **multi-node cluster mode**. Don't pre-build for them — but don't make decisions that close the door on them either.
+This section captured the original MVP cut. Most of it has since **shipped in v1.1.x**: queue triggers, cron triggers, inbound email (`email:receive`, HMAC-webhook model), KV / docs / email / users / HTTP SDKs, function-to-function `invoke()`, and secrets are all live (blueprint §12 Phase 4 table). The **Workflows** track (DAG + nested workflows) has since **shipped** (migrations `0071`/`0072`). The **§9.4 service-interceptor** track has since shipped in full (migrations `0073`–`0075`; before/after phases, data-transform, all six services, per-interceptor timeout, resolve cache, `pic interceptors ls`). A read-only **metrics/observability dashboard** has since shipped (A2): `GET /api/v1/admin/apps/{id}/metrics?window=` aggregates the existing `execution_logs` table (counts, error rate, latency avg/p50/p95, an hourly series) via `ExecutionLogRepository::summarize_for_app` + `metrics_api`, surfaced as the dashboard's per-app **Metrics** tab — no hot-path instrumentation. **Still deferred:** **multi-node cluster mode**. Don't pre-build for them — but don't make decisions that close the door on them either.
 
 **Pulled forward to Phase 3 (pre-v1.1):** admin auth, multi-app scoping. The general cross-app **export/import** sharing model stays at v1.3+; note that v1.2 §11.6 shipped a narrower form — **group-owned shared collections** (KV/docs/files/topics/queues) let apps in one subtree share data through the owning group, with the ancestor-chain walk as the isolation boundary. See blueprint §11.5 + design-doc §11.6.

crates/manager-core/migrations/0076_email_inbound_address.sql

@@ -0,0 +1,13 @@
+-- A3 raw SMTP ingress — an inbound email ADDRESS an email trigger listens on.
+--
+-- The HMAC-webhook path (v1.1.7) addresses a trigger by its UUID in the URL. A
+-- native SMTP listener instead receives `RCPT TO:<address>` and must resolve
+-- that address to the bound trigger. `inbound_address` is the mailbox the
+-- trigger claims; NULL for a webhook-only trigger. Case-insensitively unique
+-- among app-owned triggers so an address resolves to exactly one script.
+ALTER TABLE email_trigger_details
+    ADD COLUMN inbound_address TEXT;
+
+CREATE UNIQUE INDEX email_trigger_details_inbound_address_uidx
+    ON email_trigger_details (lower(inbound_address))
+    WHERE inbound_address IS NOT NULL;

crates/manager-core/src/lib.rs

@@ -269,7 +269,8 @@ pub use trigger_repo::{
     collection_matches, CreateDeadLetterTrigger, CreateDocsTrigger, CreateEmailTrigger,
     CreateFilesTrigger, CreateKvTrigger, CreatePubsubTrigger, DeadLetterTriggerMatch,
     DocsTriggerMatch, EmailInboundTarget, FilesTriggerMatch, KvTriggerMatch, PostgresTriggerRepo,
-    Trigger, TriggerDetails, TriggerDispatchMode, TriggerKind, TriggerRepo, TriggerRepoError,
+    SmtpInboundTarget, Trigger, TriggerDetails, TriggerDispatchMode, TriggerKind, TriggerRepo,
+    TriggerRepoError,
 };
 pub use triggers_api::{triggers_router, TriggersApiError, TriggersState};
 pub use users_admin_api::{app_users_router, AppUsersApiError, AppUsersState};

crates/manager-core/src/trigger_repo.rs

@@ -316,6 +316,21 @@ pub struct CreateEmailTrigger {
     /// Envelope version of the sealed inbound secret (0 = legacy no-AAD, 1 =
     /// AAD-bound to the sealing owner). `0` when there is no secret.
     pub inbound_secret_version: i16,
+    /// A3: the mailbox this trigger receives at over the native SMTP listener.
+    /// `None` = webhook-only (addressed by trigger UUID in the URL).
+    pub inbound_address: Option<String>,
+    pub registered_by_principal: AdminUserId,
+}
+
+/// A3: the SMTP-ingress lookup result — the app-owned email trigger claiming an
+/// `RCPT TO` address, plus what an `Email` outbox row needs. No secret: the
+/// SMTP path has no HMAC (the mailbox address + per-app isolation are the
+/// boundary), unlike the webhook path.
+#[derive(Debug, Clone)]
+pub struct SmtpInboundTarget {
+    pub trigger_id: TriggerId,
+    pub app_id: AppId,
+    pub script_id: ScriptId,
     pub registered_by_principal: AdminUserId,
 }
 
@@ -475,6 +490,17 @@ pub trait TriggerRepo: Send + Sync {
         trigger_id: TriggerId,
     ) -> Result<Option<EmailInboundTarget>, TriggerRepoError>;
 
+    /// A3: resolve an inbound SMTP `RCPT TO` address to the app-owned email
+    /// trigger claiming it (case-insensitive). `None` when no enabled app
+    /// trigger claims the address. Defaults to `None` so non-Postgres impls
+    /// (tests) need not provide it.
+    async fn email_inbound_target_by_address(
+        &self,
+        _address: &str,
+    ) -> Result<Option<SmtpInboundTarget>, TriggerRepoError> {
+        Ok(None)
+    }
+
     async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Trigger>, TriggerRepoError>;
 
     /// Group-owned trigger TEMPLATES (§11 tail) declared directly at `group_id`.
@@ -1420,13 +1446,15 @@ impl TriggerRepo for PostgresTriggerRepo {
 
         sqlx::query(
             "INSERT INTO email_trigger_details \
-                (trigger_id, inbound_secret_encrypted, inbound_secret_nonce, inbound_secret_version) \
+                (trigger_id, inbound_secret_encrypted, inbound_secret_nonce, \
-             VALUES ($1, $2, $3, $4)",
+                 inbound_secret_version, inbound_address) \
+             VALUES ($1, $2, $3, $4, $5)",
         )
         .bind(parent.id)
         .bind(req.inbound_secret_encrypted.as_deref())
         .bind(req.inbound_secret_nonce.as_deref())
         .bind(req.inbound_secret_version)
+        .bind(req.inbound_address.as_deref())
         .execute(&mut *tx)
         .await?;
 
@@ -1493,6 +1521,33 @@ impl TriggerRepo for PostgresTriggerRepo {
         }))
     }
 
+    async fn email_inbound_target_by_address(
+        &self,
+        address: &str,
+    ) -> Result<Option<SmtpInboundTarget>, TriggerRepoError> {
+        // `t.app_id IS NOT NULL` excludes group templates (only their
+        // materialized per-app copies are invocable); `t.enabled` skips inert
+        // triggers. The address is matched case-insensitively (mailboxes are).
+        let row: Option<(Uuid, Uuid, Uuid, Uuid)> = sqlx::query_as(
+            "SELECT t.id, t.app_id, t.script_id, t.registered_by_principal \
+             FROM triggers t \
+             JOIN email_trigger_details d ON d.trigger_id = t.id \
+             WHERE lower(d.inbound_address) = lower($1) \
+               AND t.kind = 'email' AND t.app_id IS NOT NULL AND t.enabled",
+        )
+        .bind(address)
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(
+            row.map(|(tid, app_id, script_id, principal)| SmtpInboundTarget {
+                trigger_id: tid.into(),
+                app_id: app_id.into(),
+                script_id: script_id.into(),
+                registered_by_principal: principal.into(),
+            }),
+        )
+    }
+
     async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Trigger>, TriggerRepoError> {
         let parents: Vec<TriggerRow> = sqlx::query_as(
             "SELECT id, app_id, script_id, name, kind, enabled, sealed, shared, materialized_from, \

crates/manager-core/src/triggers_api.rs

@@ -576,6 +576,10 @@ struct CreateEmailTriggerRequest {
     /// (or absent) means the trigger accepts unsigned POSTs.
     #[serde(default)]
     inbound_secret: Option<String>,
+    /// A3: the mailbox this trigger receives at over the native SMTP listener
+    /// (`RCPT TO`). Absent = webhook-only. Case-insensitively unique.
+    #[serde(default)]
+    inbound_address: Option<String>,
 }
 
 async fn create_email_trigger(
@@ -628,6 +632,10 @@ async fn create_email_trigger(
         inbound_secret_encrypted,
         inbound_secret_nonce,
         inbound_secret_version: version,
+        inbound_address: input
+            .inbound_address
+            .map(|a| a.trim().to_string())
+            .filter(|a| !a.is_empty()),
         registered_by_principal: principal.user_id,
     };
     let created = s.triggers.create_email_trigger(app_id, req).await?;

crates/manager-core/tests/expected_schema.txt

@@ -183,6 +183,7 @@ table: email_trigger_details
   inbound_secret_encrypted: bytea NULL
   inbound_secret_nonce: bytea NULL
   inbound_secret_version: smallint NOT NULL default=0
+  inbound_address: text NULL
 
 table: execution_logs
   id: uuid NOT NULL default=gen_random_uuid()
@@ -614,6 +615,7 @@ indexes on docs_trigger_details:
   docs_trigger_details_pkey: public.docs_trigger_details USING btree (trigger_id)
 
 indexes on email_trigger_details:
+  email_trigger_details_inbound_address_uidx: public.email_trigger_details USING btree (lower(inbound_address)) WHERE (inbound_address IS NOT NULL)
   email_trigger_details_pkey: public.email_trigger_details USING btree (trigger_id)
 
 indexes on execution_logs:
@@ -1151,3 +1153,4 @@ constraints on workflows:
   0073: interceptors
   0074: interceptor phase
   0075: interceptor timeout
+  0076: email inbound address

crates/picloud/Cargo.toml

@@ -34,6 +34,7 @@ thiserror.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true
 figment.workspace = true
+chrono.workspace = true
 
 [dev-dependencies]
 picloud-test-support.workspace = true

crates/picloud/src/lib.rs

@@ -3,6 +3,8 @@
 //! the listener. Tests use the same `build_app` against an
 //! ephemeral test database.
 
+pub mod smtp;
+
 use std::sync::Arc;
 use std::time::Duration;
 

crates/picloud/src/main.rs

@@ -8,12 +8,13 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Duration;
 
+use picloud::smtp::SmtpIngress;
 use picloud::{build_app, init_db, AuthDeps};
 use picloud_manager_core::{
     auth::{hash_password, validate_password_hash},
     bootstrap_first_admin, migrations, seed_hello_world_if_fresh, AdminSessionRepository,
-    AdminUserRepository, HelloWorldOutcome, PostgresAppRepository, PostgresRouteRepository,
+    AdminUserRepository, HelloWorldOutcome, PostgresAppRepository, PostgresOutboxRepo,
-    PostgresScriptRepository,
+    PostgresRouteRepository, PostgresScriptRepository, PostgresTriggerRepo,
 };
 use tracing_subscriber::EnvFilter;
 
@@ -74,6 +75,25 @@ async fn run_server() -> anyhow::Result<()> {
     // so a delayed sweep can't extend session lifetimes.
     spawn_session_pruner(auth.sessions.clone());
 
+    // A3: opt-in native SMTP ingress. Binds only when PICLOUD_SMTP_BIND is set;
+    // shares the pool, resolves each RCPT TO to an app-owned email trigger, and
+    // writes Email outbox rows the dispatcher fires (same tail as the webhook).
+    // Its own shutdown_signal() future resolves on the same ctrl-c/SIGTERM.
+    if let Ok(smtp_bind) = std::env::var("PICLOUD_SMTP_BIND") {
+        let smtp_addr: SocketAddr = smtp_bind
+            .parse()
+            .map_err(|e| anyhow::anyhow!("invalid PICLOUD_SMTP_BIND `{smtp_bind}`: {e}"))?;
+        let ingress = SmtpIngress::new(
+            Arc::new(PostgresTriggerRepo::new(pool.clone())),
+            Arc::new(PostgresOutboxRepo::new(pool.clone())),
+        );
+        tokio::spawn(async move {
+            if let Err(e) = ingress.serve(smtp_addr, shutdown_signal()).await {
+                tracing::error!(?e, "smtp ingress failed");
+            }
+        });
+    }
+
     let app = build_app(pool, auth, master_key).await?;
 
     let listener = tokio::net::TcpListener::bind(addr).await?;

crates/picloud/src/smtp.rs

@@ -0,0 +1,342 @@
+//! A3 — a minimal, receive-only native SMTP listener.
+//!
+//! The v1.1.7 inbound-email path is an HMAC-signed HTTP webhook. This adds a
+//! first-party SMTP listener so an MX can point straight at PiCloud: it speaks
+//! just enough SMTP to accept a message (`HELO`/`EHLO`, `MAIL`, `RCPT`, `DATA`,
+//! `RSET`, `NOOP`, `QUIT`), resolves each `RCPT TO` mailbox to the app-owned
+//! email trigger that claims it, and inserts an `Email` outbox row exactly as
+//! the webhook does — so the dispatcher fires the bound script unchanged.
+//!
+//! **Security model.** No AUTH and no STARTTLS at the listener (TLS terminates
+//! upstream — Caddy/stunnel or the sending relay); the recipient *address* plus
+//! the existing per-app isolation are the boundary (an address resolves to
+//! exactly one app-owned trigger). Opt-in: the listener binds only when
+//! `PICLOUD_SMTP_BIND` is set. `DATA` is hard-capped so a single session can't
+//! stream unbounded bytes.
+//!
+//! **Parsing.** A deliberately minimal RFC-5322 split — headers up to the first
+//! blank line (`From`/`Subject`/`Message-ID` extracted), the rest as the text
+//! body. No MIME/multipart decoding (a follow-up); an HTML-only or multipart
+//! message still delivers, with the raw body as `text`.
+
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use picloud_manager_core::{NewOutboxRow, OutboxRepo, OutboxSourceKind, TriggerRepo};
+use picloud_shared::TriggerEvent;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::net::{TcpListener, TcpStream};
+
+/// Default `DATA` ceiling (1 MiB) — matches the webhook body cap.
+const DEFAULT_MAX_MESSAGE_BYTES: usize = 1024 * 1024;
+
+/// The SMTP ingress: resolves recipients and writes outbox rows. Cheap to
+/// clone (two `Arc`s).
+#[derive(Clone)]
+pub struct SmtpIngress {
+    triggers: Arc<dyn TriggerRepo>,
+    outbox: Arc<dyn OutboxRepo>,
+    max_message_bytes: usize,
+}
+
+impl SmtpIngress {
+    #[must_use]
+    pub fn new(triggers: Arc<dyn TriggerRepo>, outbox: Arc<dyn OutboxRepo>) -> Self {
+        Self {
+            triggers,
+            outbox,
+            max_message_bytes: DEFAULT_MAX_MESSAGE_BYTES,
+        }
+    }
+
+    /// Is `address` a live mailbox (an enabled app-owned email trigger claims
+    /// it)? Used to accept/reject `RCPT TO` per the SMTP convention.
+    async fn mailbox_exists(&self, address: &str) -> bool {
+        matches!(
+            self.triggers.email_inbound_target_by_address(address).await,
+            Ok(Some(_))
+        )
+    }
+
+    /// Deliver one received message to every matching recipient: resolve the
+    /// address → trigger, build the `Email` event, insert an outbox row (the
+    /// dispatcher fires the script). Returns the number of rows written. Unknown
+    /// mailboxes and per-recipient failures are skipped (logged), never fatal to
+    /// the others.
+    pub async fn deliver(&self, recipients: &[String], raw: &[u8]) -> usize {
+        let msg = ParsedMessage::parse(raw);
+        let mut delivered = 0;
+        for rcpt in recipients {
+            let target = match self.triggers.email_inbound_target_by_address(rcpt).await {
+                Ok(Some(t)) => t,
+                Ok(None) => continue, // unknown mailbox — silently drop
+                Err(e) => {
+                    tracing::error!(?e, rcpt, "smtp: mailbox lookup failed");
+                    continue;
+                }
+            };
+            let event = TriggerEvent::Email {
+                from: msg.from.clone(),
+                to: recipients.to_vec(),
+                cc: Vec::new(),
+                subject: msg.subject.clone(),
+                text: Some(msg.body.clone()),
+                html: None,
+                received_at: chrono::Utc::now(),
+                message_id: msg.message_id.clone(),
+            };
+            let payload = match serde_json::to_value(&event) {
+                Ok(p) => p,
+                Err(e) => {
+                    tracing::error!(?e, "smtp: serialize email event failed");
+                    continue;
+                }
+            };
+            let row = NewOutboxRow {
+                app_id: target.app_id,
+                source_kind: OutboxSourceKind::Email,
+                trigger_id: Some(target.trigger_id),
+                script_id: Some(target.script_id),
+                reply_to: None,
+                payload,
+                origin_principal: Some(target.registered_by_principal),
+                trigger_depth: 1,
+                root_execution_id: None,
+            };
+            match self.outbox.insert(row).await {
+                Ok(_) => delivered += 1,
+                Err(e) => tracing::error!(?e, rcpt, "smtp: outbox insert failed"),
+            }
+        }
+        delivered
+    }
+
+    /// Bind `addr` and serve until `shutdown` resolves. Each connection is
+    /// handled on its own task; a connection error is logged, never fatal.
+    pub async fn serve(
+        self,
+        addr: SocketAddr,
+        shutdown: impl std::future::Future<Output = ()> + Send,
+    ) -> std::io::Result<()> {
+        let listener = TcpListener::bind(addr).await?;
+        tracing::info!(%addr, "smtp ingress listening");
+        tokio::pin!(shutdown);
+        loop {
+            tokio::select! {
+                () = &mut shutdown => break,
+                accepted = listener.accept() => {
+                    match accepted {
+                        Ok((stream, peer)) => {
+                            let me = self.clone();
+                            tokio::spawn(async move {
+                                if let Err(e) = me.handle_conn(stream).await {
+                                    tracing::debug!(?e, %peer, "smtp connection ended");
+                                }
+                            });
+                        }
+                        Err(e) => tracing::warn!(?e, "smtp accept failed"),
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+
+    /// Drive one SMTP session to completion.
+    async fn handle_conn(&self, stream: TcpStream) -> std::io::Result<()> {
+        let (read, mut write) = stream.into_split();
+        let mut reader = BufReader::new(read);
+        write.write_all(b"220 picloud ESMTP ready\r\n").await?;
+
+        let mut recipients: Vec<String> = Vec::new();
+        let mut line = String::new();
+        loop {
+            line.clear();
+            if reader.read_line(&mut line).await? == 0 {
+                break; // peer closed
+            }
+            let cmd = line.trim_end_matches(['\r', '\n']);
+            let verb = cmd
+                .split_whitespace()
+                .next()
+                .unwrap_or("")
+                .to_ascii_uppercase();
+            match verb.as_str() {
+                "HELO" | "EHLO" => write.write_all(b"250 picloud\r\n").await?,
+                "MAIL" | "NOOP" => write.write_all(b"250 OK\r\n").await?,
+                "RCPT" => match extract_address(cmd) {
+                    Some(addr) if self.mailbox_exists(&addr).await => {
+                        recipients.push(addr);
+                        write.write_all(b"250 OK\r\n").await?;
+                    }
+                    _ => {
+                        write.write_all(b"550 5.1.1 no such mailbox\r\n").await?;
+                    }
+                },
+                "DATA" => {
+                    if recipients.is_empty() {
+                        write
+                            .write_all(b"554 5.5.1 no valid recipients\r\n")
+                            .await?;
+                        continue;
+                    }
+                    write
+                        .write_all(b"354 End data with <CR><LF>.<CR><LF>\r\n")
+                        .await?;
+                    if let Some(raw) = read_data(&mut reader, self.max_message_bytes).await? {
+                        let n = self.deliver(&recipients, &raw).await;
+                        recipients.clear();
+                        write
+                            .write_all(format!("250 OK queued {n}\r\n").as_bytes())
+                            .await?;
+                    } else {
+                        recipients.clear();
+                        write.write_all(b"552 5.3.4 message too large\r\n").await?;
+                    }
+                }
+                "RSET" => {
+                    recipients.clear();
+                    write.write_all(b"250 OK\r\n").await?;
+                }
+                "QUIT" => {
+                    write.write_all(b"221 2.0.0 Bye\r\n").await?;
+                    break;
+                }
+                _ => {
+                    write
+                        .write_all(b"502 5.5.2 command not recognized\r\n")
+                        .await?;
+                }
+            }
+        }
+        Ok(())
+    }
+}
+
+/// Read the `DATA` payload: lines until a lone `.`, with dot-unstuffing (a line
+/// starting `..` becomes `.`). Bounded by a cumulative byte cap — over it, we
+/// drain to the terminator (to keep the socket in sync) and return `None` so the
+/// caller replies 552. (TLS/relay upstream bounds any single unterminated line.)
+async fn read_data<R: AsyncBufReadExt + Unpin>(
+    reader: &mut R,
+    max_bytes: usize,
+) -> std::io::Result<Option<Vec<u8>>> {
+    let mut data: Vec<u8> = Vec::new();
+    let mut total = 0usize;
+    let mut line = String::new();
+    loop {
+        line.clear();
+        let n = reader.read_line(&mut line).await?;
+        if n == 0 {
+            break; // connection dropped mid-DATA
+        }
+        total += n;
+        let trimmed = line.trim_end_matches(['\r', '\n']);
+        if trimmed == "." {
+            return Ok(Some(data));
+        }
+        if total > max_bytes {
+            drain_to_terminator(reader).await?;
+            return Ok(None);
+        }
+        // Dot-unstuffing (RFC 5321 §4.5.2).
+        let content = trimmed.strip_prefix('.').unwrap_or(trimmed);
+        data.extend_from_slice(content.as_bytes());
+        data.extend_from_slice(b"\r\n");
+    }
+    Ok(Some(data))
+}
+
+/// Drain remaining `DATA` lines up to the `.` terminator after an overflow, so
+/// the connection stays protocol-synchronized. Bounded by a line budget.
+async fn drain_to_terminator<R: AsyncBufReadExt + Unpin>(reader: &mut R) -> std::io::Result<()> {
+    let mut line = String::new();
+    for _ in 0..1_000_000 {
+        line.clear();
+        let n = reader.read_line(&mut line).await?;
+        if n == 0 || line.trim_end_matches(['\r', '\n']) == "." {
+            break;
+        }
+    }
+    Ok(())
+}
+
+/// Extract the address from a `RCPT TO:<addr>` (or `RCPT TO: addr`) command.
+fn extract_address(cmd: &str) -> Option<String> {
+    let after = cmd.split_once(':')?.1.trim();
+    let inner = after
+        .strip_prefix('<')
+        .and_then(|s| s.strip_suffix('>'))
+        .unwrap_or(after);
+    let addr = inner.trim();
+    (!addr.is_empty() && addr.contains('@')).then(|| addr.to_string())
+}
+
+/// The minimal parse of a received message.
+struct ParsedMessage {
+    from: String,
+    subject: String,
+    message_id: Option<String>,
+    body: String,
+}
+
+impl ParsedMessage {
+    fn parse(raw: &[u8]) -> Self {
+        let text = String::from_utf8_lossy(raw);
+        // Headers up to the first blank line; the rest is the body.
+        let (headers, body) = match text.split_once("\r\n\r\n") {
+            Some((h, b)) => (h, b),
+            None => text.split_once("\n\n").unwrap_or((&text, "")),
+        };
+        let mut from = String::new();
+        let mut subject = String::new();
+        let mut message_id = None;
+        for header_line in headers.lines() {
+            if let Some((name, value)) = header_line.split_once(':') {
+                let value = value.trim().to_string();
+                match name.trim().to_ascii_lowercase().as_str() {
+                    "from" => from = value,
+                    "subject" => subject = value,
+                    "message-id" => message_id = Some(value),
+                    _ => {}
+                }
+            }
+        }
+        Self {
+            from,
+            subject,
+            message_id,
+            body: body.to_string(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn extract_address_handles_angle_brackets_and_bare() {
+        assert_eq!(
+            extract_address("RCPT TO:<a@b.com>"),
+            Some("a@b.com".to_string())
+        );
+        assert_eq!(
+            extract_address("RCPT TO: c@d.com"),
+            Some("c@d.com".to_string())
+        );
+        assert_eq!(extract_address("RCPT TO:<>"), None);
+        assert_eq!(extract_address("RCPT TO:not-an-address"), None);
+    }
+
+    #[test]
+    fn parse_extracts_headers_and_body() {
+        let raw =
+            b"From: alice@example.com\r\nSubject: Hi\r\nMessage-ID: <1@x>\r\n\r\nhello world\r\n";
+        let m = ParsedMessage::parse(raw);
+        assert_eq!(m.from, "alice@example.com");
+        assert_eq!(m.subject, "Hi");
+        assert_eq!(m.message_id.as_deref(), Some("<1@x>"));
+        assert!(m.body.contains("hello world"));
+    }
+}

crates/picloud/tests/smtp_ingress.rs

@@ -0,0 +1,105 @@
+//! A3 — the native SMTP ingress delivery core. Seeds an app-owned email trigger
+//! with an inbound `RCPT TO` address, then drives `SmtpIngress::deliver`
+//! directly (no socket — deterministic) and asserts it writes exactly one
+//! `Email` outbox row for the known mailbox and none for an unknown one. The
+//! dispatcher tail (outbox → script) is covered by the existing dispatcher
+//! tests; this pins the resolve-address → outbox step the SMTP path adds.
+
+use std::sync::Arc;
+
+use picloud::smtp::SmtpIngress;
+use picloud_manager_core::{PostgresOutboxRepo, PostgresTriggerRepo};
+use sqlx::PgPool;
+use uuid::Uuid;
+
+mod common;
+
+async fn seed(pool: &PgPool, address: &str) -> Uuid {
+    let group: (Uuid,) =
+        sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id")
+            .bind(format!("smtp-g-{}", Uuid::new_v4()))
+            .fetch_one(pool)
+            .await
+            .expect("group");
+    let app: (Uuid,) =
+        sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id")
+            .bind(format!("smtp-a-{}", Uuid::new_v4()))
+            .bind(group.0)
+            .fetch_one(pool)
+            .await
+            .expect("app");
+    let admin: (Uuid,) = sqlx::query_as(
+        "INSERT INTO admin_users (username, password_hash) VALUES ($1, 'x') RETURNING id",
+    )
+    .bind(format!("smtp-u-{}", Uuid::new_v4()))
+    .fetch_one(pool)
+    .await
+    .expect("admin");
+    let script: (Uuid,) = sqlx::query_as(
+        "INSERT INTO scripts (name, source, app_id) VALUES ($1, 'x', $2) RETURNING id",
+    )
+    .bind(format!("smtp-s-{}", Uuid::new_v4()))
+    .bind(app.0)
+    .fetch_one(pool)
+    .await
+    .expect("script");
+    let trigger: (Uuid,) = sqlx::query_as(
+        "INSERT INTO triggers \
+            (app_id, script_id, kind, enabled, dispatch_mode, retry_max_attempts, \
+             retry_backoff, retry_base_ms, registered_by_principal, name) \
+         VALUES ($1, $2, 'email', TRUE, 'async', 3, 'exponential', 1000, $3, $4) RETURNING id",
+    )
+    .bind(app.0)
+    .bind(script.0)
+    .bind(admin.0)
+    .bind(format!("smtp-t-{}", Uuid::new_v4()))
+    .fetch_one(pool)
+    .await
+    .expect("trigger");
+    sqlx::query("INSERT INTO email_trigger_details (trigger_id, inbound_address) VALUES ($1, $2)")
+        .bind(trigger.0)
+        .bind(address)
+        .execute(pool)
+        .await
+        .expect("email details");
+    app.0
+}
+
+async fn email_outbox_count(pool: &PgPool, app_id: Uuid) -> i64 {
+    let row: (i64,) =
+        sqlx::query_as("SELECT count(*) FROM outbox WHERE app_id = $1 AND source_kind = 'email'")
+            .bind(app_id)
+            .fetch_one(pool)
+            .await
+            .expect("count");
+    row.0
+}
+
+#[tokio::test]
+async fn deliver_writes_one_outbox_row_for_a_known_mailbox() {
+    let Some(pool) = common::test_pool("smtp_ingress").await else {
+        return;
+    };
+    let address = "hooks@app.test";
+    let app_id = seed(&pool, address).await;
+
+    let ingress = SmtpIngress::new(
+        Arc::new(PostgresTriggerRepo::new(pool.clone())),
+        Arc::new(PostgresOutboxRepo::new(pool.clone())),
+    );
+    let raw = b"From: alice@example.com\r\nSubject: Hi there\r\n\r\nhello body\r\n";
+
+    // Known mailbox → one outbox row.
+    let delivered = ingress.deliver(&[address.to_string()], raw).await;
+    assert_eq!(delivered, 1, "a known mailbox must be delivered");
+    assert_eq!(email_outbox_count(&pool, app_id).await, 1);
+
+    // Unknown mailbox → nothing (the resolution IS the boundary).
+    let delivered = ingress.deliver(&["nobody@app.test".to_string()], raw).await;
+    assert_eq!(delivered, 0, "an unknown mailbox must not deliver");
+    assert_eq!(
+        email_outbox_count(&pool, app_id).await,
+        1,
+        "no extra outbox row for an unknown mailbox"
+    );
+}