feat(v1.1.7-email-outbound): SMTP send/send_html

Outbound email reachable from scripts as email::send(#{...}) (plain
text) and email::send_html(#{...}) (multipart text + HTML). Backed by a
lettre SMTP relay configured from PICLOUD_SMTP_HOST/PORT/USER/PASSWORD/
TLS/TIMEOUT_SECS; if HOST/USER/PASSWORD aren't all set the service runs
in disabled mode (every send throws NotConfigured, warned at startup).

- EmailService trait + OutboundEmail DTO (picloud-shared);
  EmailServiceImpl + EmailTransport seam + lettre transport
  (manager-core), wired into the Services bundle and Rhai engine.
- Capability::AppEmailSend (→ script:write); seven-scope commitment held.
- Required-field + RFC5322-ish address validation; 25 MB per-message cap
  (PICLOUD_EMAIL_MAX_MESSAGE_BYTES). reply_to defaults to from.
- Per-call connection (pooling deferred to v1.2); no per-app from
  validation (operator's SMTP/SPF/DKIM concern).

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

crates/executor-core/tests/sdk_email.rs

@@ -0,0 +1,159 @@
+//! `email::` SDK bridge integration tests — runs a real Rhai engine
+//! against a recording `EmailService`. Verifies the Rhai map → DTO
+//! plumbing (address coercion, the text-only vs multipart split). The
+//! SMTP transport, validation, and authz are unit-tested at the service
+//! layer in `manager-core::email_service`.
+
+use std::collections::BTreeMap;
+use std::sync::{Arc, Mutex};
+
+use async_trait::async_trait;
+use picloud_executor_core::{Engine, ExecRequest, InvocationType, Limits};
+use picloud_shared::{
+    AppId, EmailError, EmailService, ExecutionId, NoopDeadLetterService, NoopDocsService,
+    NoopEventEmitter, NoopHttpService, NoopKvService, NoopModuleSource, OutboundEmail, RequestId,
+    ScriptId, ScriptSandbox, SdkCallCx, Services,
+};
+use serde_json::Value;
+
+#[derive(Default)]
+struct RecordingEmail {
+    sent: Mutex<Vec<OutboundEmail>>,
+}
+
+#[async_trait]
+impl EmailService for RecordingEmail {
+    async fn send(&self, _cx: &SdkCallCx, email: OutboundEmail) -> Result<(), EmailError> {
+        self.sent.lock().unwrap().push(email);
+        Ok(())
+    }
+}
+
+fn engine_with(rec: Arc<RecordingEmail>) -> Arc<Engine> {
+    let services = Services::new(
+        Arc::new(NoopKvService),
+        Arc::new(NoopDocsService),
+        Arc::new(NoopDeadLetterService),
+        Arc::new(NoopEventEmitter),
+        Arc::new(NoopModuleSource),
+        Arc::new(NoopHttpService),
+        Arc::new(picloud_shared::NoopFilesService),
+        Arc::new(picloud_shared::NoopPubsubService),
+        Arc::new(picloud_shared::NoopSecretsService),
+        rec,
+    );
+    Arc::new(Engine::new(Limits::default(), services))
+}
+
+fn baseline_request(app_id: AppId) -> ExecRequest {
+    let execution_id = ExecutionId::new();
+    ExecRequest {
+        execution_id,
+        request_id: RequestId::new(),
+        script_id: ScriptId::new(),
+        script_name: "email-test".into(),
+        invocation_type: InvocationType::Http,
+        path: "/email-test".into(),
+        headers: BTreeMap::new(),
+        body: Value::Null,
+        params: BTreeMap::new(),
+        query: BTreeMap::new(),
+        rest: String::new(),
+        sandbox_overrides: ScriptSandbox::default(),
+        app_id,
+        principal: None,
+        trigger_depth: 0,
+        root_execution_id: execution_id,
+        is_dead_letter_handler: false,
+        event: None,
+    }
+}
+
+async fn run(engine: Arc<Engine>, src: &str) -> Result<(), ()> {
+    let src = src.to_string();
+    let app = AppId::new();
+    tokio::task::spawn_blocking(move || engine.execute(&src, baseline_request(app)))
+        .await
+        .expect("spawn_blocking")
+        .map(|_| ())
+        .map_err(|_| ())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn send_parses_single_recipient_text() {
+    let rec = Arc::new(RecordingEmail::default());
+    let engine = engine_with(rec.clone());
+    run(
+        engine,
+        r#"
+        email::send(#{
+            to: "alice@example.com",
+            from: "alerts@myapp.com",
+            subject: "Build complete",
+            text: "done"
+        });
+        #{ ok: true }
+    "#,
+    )
+    .await
+    .unwrap();
+    let g = rec.sent.lock().unwrap();
+    let e = g.last().unwrap();
+    assert_eq!(e.to, vec!["alice@example.com".to_string()]);
+    assert_eq!(e.from, "alerts@myapp.com");
+    assert_eq!(e.subject, "Build complete");
+    assert_eq!(e.text.as_deref(), Some("done"));
+    // email::send forces text-only even if html were present.
+    assert!(e.html.is_none());
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn send_html_carries_both_parts_and_lists() {
+    let rec = Arc::new(RecordingEmail::default());
+    let engine = engine_with(rec.clone());
+    run(
+        engine,
+        r#"
+        email::send_html(#{
+            to: ["alice@x.com", "bob@y.com"],
+            cc: ["dave@z.com"],
+            bcc: ["audit@myapp.com"],
+            from: "alerts@myapp.com",
+            reply_to: "support@myapp.com",
+            subject: "hi",
+            text: "plain",
+            html: "<p>rich</p>"
+        });
+        #{ ok: true }
+    "#,
+    )
+    .await
+    .unwrap();
+    let g = rec.sent.lock().unwrap();
+    let e = g.last().unwrap();
+    assert_eq!(
+        e.to,
+        vec!["alice@x.com".to_string(), "bob@y.com".to_string()]
+    );
+    assert_eq!(e.cc, vec!["dave@z.com".to_string()]);
+    assert_eq!(e.bcc, vec!["audit@myapp.com".to_string()]);
+    assert_eq!(e.reply_to.as_deref(), Some("support@myapp.com"));
+    assert_eq!(e.text.as_deref(), Some("plain"));
+    assert_eq!(e.html.as_deref(), Some("<p>rich</p>"));
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn send_html_without_html_throws() {
+    let rec = Arc::new(RecordingEmail::default());
+    let engine = engine_with(rec.clone());
+    let res = run(
+        engine,
+        r#"
+        email::send_html(#{ to: "a@b.com", from: "c@d.com", subject: "x", text: "y" });
+        #{ ok: true }
+    "#,
+    )
+    .await;
+    assert!(res.is_err(), "send_html without html must throw");
+    assert!(rec.sent.lock().unwrap().is_empty());
+}