feat: E2E #2 (Stash) gap remediation + S6 hardening

Closes the gaps and the one security finding from the second end-to-end
CLI test (E2E_STASH_REPORT.md), plus the H1 boot-regression found while
re-reviewing those fixes.

Security
- S6: reserved-path validation (`check_reserved`) now case-folds before
  comparing, so `/API/v2/x`, `/HEALTHZ`, `/Admin/x` are rejected like
  their lowercase forms. Request-time matching stays case-sensitive.
- S10: "public route != public data" callout in sdk-shape.md (script_gate
  skips authz when the principal is anonymous).

Observability / features
- G1: trigger executions now write `execution_logs`. Migration 0043 adds
  a `source` column (CHECK mirrors ExecutionSource/OutboxSourceKind,
  DEFAULT 'http' backfills history); a shared `build_execution_log` helper
  in executor-core; dispatcher logging for outbox triggers + queue
  consumers (skips sync-HTTP rows the orchestrator already logs). `pic
  logs` gains a source column + `--source` filter.
- G5: dev-only in-memory email capture under PICLOUD_DEV_MODE with no SMTP
  (email::send succeeds locally), readable at GET /api/v1/admin/dev/emails
  (Owner/Admin only; route mounted only in capture mode).
- G6: generalized the Rhai in-place-mutation footgun note (trim/replace/
  make_upper/make_lower/crop/truncate/pad return ()).
- G2/G3/G4 (CLI): `pic members`, `pic files`, `pic queues`, read-only
  `pic kv` (+ new kv_api.rs); `pic deploy --timeout/--memory/--kind/
  --sandbox`; first-class `pic triggers create-{docs,files,pubsub,queue,
  email}` wrappers. All new client path segments percent-encoded via seg().

H1 regression fix (found in re-review)
- The S6 change also runs in `compile_routes`, which compiles every stored
  route at boot and on each route CRUD. A single stored route the new
  validation rejects (creatable while the S6 gap existed) made the whole
  compile Err and aborted startup. `compile_routes` is now lenient: it
  skips an un-compilable row with a warning instead of bricking boot
  (route creation still validates separately). Migration 0044 sweeps
  pre-existing reserved-path routes on upgrade (WHERE mirrors
  check_reserved exactly). Added regression tests for both.

Verified: cargo fmt, clippy --all-targets --all-features -D warnings, the
schema_snapshot test, and the new S6/lenient-compile unit tests all pass;
boot-resilience and G1/G5 confirmed live.

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

crates/manager-core/migrations/0043_execution_logs_source.sql

@@ -0,0 +1,20 @@
+-- G1 (E2E #2 "Stash"): trigger executions were invisible to `pic logs`.
+--
+-- Only HTTP-route executions ever wrote an `execution_logs` row; queue,
+-- cron, dead-letter, and `invoke()` runs left no trace, so background
+-- workers were observable only via dead-letters (failures) or their own
+-- side effects. The dispatcher now logs every trigger run too — this
+-- column records which kind of event dispatched each execution so the
+-- logs surface can show, and filter by, the origin.
+--
+-- DEFAULT 'http' backfills every pre-existing row: before this change the
+-- only thing that logged was the HTTP path, so 'http' is correct history.
+-- The CHECK list mirrors `manager-core::OutboxSourceKind` /
+-- `shared::ExecutionSource`; keep all three in sync.
+
+ALTER TABLE execution_logs
+    ADD COLUMN source TEXT NOT NULL DEFAULT 'http'
+    CHECK (source IN (
+        'http', 'kv', 'docs', 'dead_letter', 'cron',
+        'files', 'pubsub', 'email', 'invoke', 'queue'
+    ));

crates/manager-core/migrations/0044_delete_reserved_path_routes.sql

@@ -0,0 +1,22 @@
+-- H1 (re-review of the S6 reserved-path fix): the reserved-prefix check is
+-- now case-insensitive, both at route creation AND when the route table is
+-- compiled at boot. Routes created before the fix — while validation was
+-- case-sensitive — could hold paths like `/API/v2/x`, `/Admin/x`, or
+-- `/HEALTHZ`. `compile_routes` now skips such rows with a warning instead of
+-- aborting startup (so an un-upgraded boot can't be bricked), but those
+-- routes violate the reserved namespace and can never be served safely, so
+-- sweep them here on upgrade.
+--
+-- Mirrors `orchestrator-core::routing::pattern::check_reserved` exactly,
+-- case-insensitively: a path is reserved if its lowercased form equals one
+-- of the bare names (`/api` `/admin` `/healthz` `/version`) or starts with
+-- one of the prefixes. Note `/api/` and `/admin/` reserve on the trailing
+-- slash, while `/healthz` and `/version` reserve on bare prefix — matching
+-- the RESERVED_PATH_PREFIXES list. (Idempotent: a no-op once swept.)
+
+DELETE FROM routes
+WHERE lower(path) IN ('/api', '/admin', '/healthz', '/version')
+   OR lower(path) LIKE '/api/%'
+   OR lower(path) LIKE '/admin/%'
+   OR lower(path) LIKE '/healthz%'
+   OR lower(path) LIKE '/version%';

crates/manager-core/src/api.rs

@@ -12,8 +12,8 @@ use axum::{
     Extension, Json, Router,
 };
 use picloud_shared::{
-    AppId, ExecutionLog, InstanceRole, Principal, Script, ScriptId, ScriptKind, ScriptSandbox,
-    ScriptValidator, ValidatedScript, ValidationError,
+    AppId, ExecutionLog, ExecutionSource, InstanceRole, Principal, Script, ScriptId, ScriptKind,
+    ScriptSandbox, ScriptValidator, ValidatedScript, ValidationError,
 };
 use serde::Deserialize;
 
@@ -385,6 +385,10 @@ pub struct LogsQuery {
     #[serde(default, rename = "offset")]
     #[allow(dead_code)]
     pub legacy_offset: Option<i64>,
+    /// Optional origin filter (`http`, `kv`, `cron`, `invoke`, …). Absent
+    /// → all sources. An unrecognized value is a 422 (see `list_logs`).
+    #[serde(default)]
+    pub source: Option<String>,
 }
 
 const fn default_limit() -> i64 {
@@ -411,7 +415,17 @@ async fn list_logs<R: ScriptRepository, L: ExecutionLogRepository>(
         .cursor
         .as_deref()
         .and_then(crate::repo::ExecutionLogCursor::decode);
-    let logs = state.logs.list_for_script(id, limit, cursor).await?;
+    let source = match q.source.as_deref() {
+        None | Some("" | "all") => None,
+        Some(s) => Some(
+            ExecutionSource::from_wire(s)
+                .ok_or_else(|| ApiError::BadRequest(format!("unknown log source: {s:?}")))?,
+        ),
+    };
+    let logs = state
+        .logs
+        .list_for_script(id, limit, cursor, source)
+        .await?;
     Ok(Json(logs))
 }
 
@@ -427,6 +441,9 @@ pub enum ApiError {
     #[error("app not found: {0}")]
     AppNotFound(String),
 
+    #[error("bad request: {0}")]
+    BadRequest(String),
+
     #[error("conflict: {0}")]
     Conflict(String),
 
@@ -459,11 +476,11 @@ impl IntoResponse for ApiError {
     fn into_response(self) -> Response {
         let (status, message) = match &self {
             Self::NotFound(_) => (StatusCode::NOT_FOUND, self.to_string()),
-            Self::AppNotFound(_) => (StatusCode::UNPROCESSABLE_ENTITY, self.to_string()),
+            Self::AppNotFound(_)
+            | Self::BadRequest(_)
+            | Self::Invalid(_)
+            | Self::Ceiling(_) => (StatusCode::UNPROCESSABLE_ENTITY, self.to_string()),
             Self::Conflict(_) => (StatusCode::CONFLICT, self.to_string()),
-            Self::Invalid(_) | Self::Ceiling(_) => {
-                (StatusCode::UNPROCESSABLE_ENTITY, self.to_string())
-            }
             Self::Forbidden => (StatusCode::FORBIDDEN, self.to_string()),
             Self::AuthzRepo(e) => {
                 tracing::error!(error = %e, "authz repo error");

crates/manager-core/src/dev_email_api.rs

@@ -0,0 +1,48 @@
+//! `GET /api/v1/admin/dev/emails` — dev-only inspection of mail captured
+//! by the in-memory email sink (G5).
+//!
+//! Mounted **only** when the email service is running in dev-capture mode
+//! (`PICLOUD_DEV_MODE=true` and no SMTP relay configured). In every other
+//! configuration the route does not exist, so there is no production
+//! surface here. Capture is instance-wide (the SMTP transport seam can't
+//! see a script's `app_id`), so the endpoint is instance-wide too and is
+//! restricted to instance Owners/Admins.
+
+use std::sync::Arc;
+
+use axum::extract::State;
+use axum::http::StatusCode;
+use axum::response::Json;
+use axum::routing::get;
+use axum::{Extension, Router};
+use picloud_shared::{InstanceRole, Principal};
+
+use crate::email_service::{CapturedEmail, DevEmailSink};
+
+#[derive(Clone)]
+pub struct DevEmailState {
+    pub sink: Arc<DevEmailSink>,
+}
+
+/// Build the dev-email router. Callers mount this only when dev-capture
+/// mode is active (i.e. they hold a `Some(sink)`).
+pub fn dev_emails_router(state: DevEmailState) -> Router {
+    Router::new()
+        .route("/dev/emails", get(list_dev_emails))
+        .with_state(state)
+}
+
+async fn list_dev_emails(
+    Extension(principal): Extension<Principal>,
+    State(state): State<DevEmailState>,
+) -> Result<Json<Vec<CapturedEmail>>, StatusCode> {
+    // Instance-wide data → require an instance Owner/Admin. A Member
+    // (app-scoped) principal has no business reading every app's mail.
+    if !matches!(
+        principal.instance_role,
+        InstanceRole::Owner | InstanceRole::Admin
+    ) {
+        return Err(StatusCode::FORBIDDEN);
+    }
+    Ok(Json(state.sink.snapshot()))
+}

crates/manager-core/src/dispatcher.rs

@@ -24,11 +24,14 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use chrono::{DateTime, Utc};
-use picloud_executor_core::{ExecError, ExecRequest, ExecResponse, InvocationType};
+use picloud_executor_core::{
+    build_execution_log, ExecError, ExecRequest, ExecResponse, InvocationType,
+};
 use picloud_orchestrator_core::{ExecutionGate, ExecutorClient};
 use picloud_shared::{
-    DeadLetterId, ExecResponseSummary, ExecutionId, HttpDispatchPayload, InboxDeliveryOutcome,
-    InboxFailureKind, InboxResolver, InboxResult, RequestId, ScriptId, ScriptSandbox, TriggerEvent,
+    DeadLetterId, ExecResponseSummary, ExecutionId, ExecutionLogSink, ExecutionSource,
+    HttpDispatchPayload, InboxDeliveryOutcome, InboxFailureKind, InboxResolver, InboxResult,
+    RequestId, ScriptId, ScriptSandbox, TriggerEvent,
 };
 use rand::Rng;
 use uuid::Uuid;
@@ -53,6 +56,11 @@ pub struct Dispatcher {
     pub principals: Arc<dyn PrincipalResolver>,
     pub executor: Arc<dyn ExecutorClient>,
     pub gate: Arc<ExecutionGate>,
+    /// G1: records an `execution_logs` row for every trigger run so
+    /// background workers (queue / cron / dead-letter / invoke) show up
+    /// in `pic logs`, not just synchronous HTTP. Same sink the
+    /// orchestrator's data plane writes through.
+    pub log_sink: Arc<dyn ExecutionLogSink>,
     pub inbox: Arc<dyn InboxResolver>,
     /// v1.1.9. Reads `queue_messages` for the queue arm + the reclaim
     /// task. None in tests / harnesses that don't exercise queues.
@@ -149,6 +157,39 @@ fn async_exec_timeout_from_env() -> Duration {
 /// parallelism, not the dispatcher's serial-await.
 const QUEUE_DISPATCH_PARALLELISM: usize = 32;
 
+/// Map an outbox row's source kind to the execution-log `source`. The
+/// wire strings are identical, so this is total in practice; an unknown
+/// kind would default to `Http`.
+fn exec_source(row: &OutboxRow) -> ExecutionSource {
+    ExecutionSource::from_wire(row.source_kind.as_str()).unwrap_or_default()
+}
+
+/// G1: the slice of an `ExecRequest` needed to write an execution-log
+/// row, captured before the request is moved into the executor.
+struct ExecLogContext {
+    app_id: picloud_shared::AppId,
+    script_id: ScriptId,
+    request_id: RequestId,
+    path: String,
+    headers: std::collections::BTreeMap<String, String>,
+    body: serde_json::Value,
+    source: ExecutionSource,
+}
+
+impl ExecLogContext {
+    fn from_request(req: &ExecRequest, source: ExecutionSource) -> Self {
+        Self {
+            app_id: req.app_id,
+            script_id: req.script_id,
+            request_id: req.request_id,
+            path: req.path.clone(),
+            headers: req.headers.clone(),
+            body: req.body.clone(),
+            source,
+        }
+    }
+}
+
 impl Dispatcher {
     /// Spawn the dispatcher loop as a detached `tokio::task`. Also
     /// spawns the v1.1.9 queue visibility-timeout reclaim task. Both
@@ -376,6 +417,11 @@ impl Dispatcher {
             script_id: consumer.script_id,
             updated_at: script.updated_at,
         };
+        // G1: queue consumers dispatch outside the outbox, so they need
+        // their own log write. Source is always `queue`; no `reply_to`
+        // here, so there's no double-logging concern.
+        let log_cx = ExecLogContext::from_request(&exec_req, ExecutionSource::Queue);
+        let started = Utc::now();
         let outcome = self
             .executor
             .execute_with_identity(
@@ -385,8 +431,12 @@ impl Dispatcher {
                 async_exec_timeout_from_env(),
             )
             .await;
+        let finished = Utc::now();
         drop(permit);
 
+        self.record_execution_log(&log_cx, &outcome, started, finished)
+            .await;
+
         // Best-effort touch on last_fired_at (a failure here doesn't
         // change ack/nack behavior).
         if let Err(e) = self
@@ -585,18 +635,67 @@ impl Dispatcher {
             script_id: resolved.script_id,
             updated_at: resolved.script_updated_at,
         };
+        // G1: capture the request context before `exec_req` is consumed so
+        // we can write an execution-log row for this run (see below).
+        let log_cx = ExecLogContext::from_request(&exec_req, exec_source(&row));
+        let started = Utc::now();
         let outcome = self
             .executor
             .execute_with_identity(identity, &source, exec_req, async_exec_timeout_from_env())
             .await;
+        let finished = Utc::now();
         drop(permit);
 
+        // G1: persist an execution-log row so this run shows up in
+        // `pic logs`. Skip rows with a synchronous receiver (`reply_to`
+        // is set) — those are sync HTTP routes that the orchestrator's
+        // inbox path already logs, and logging here too would duplicate
+        // them. Trigger rows and fire-and-forget async HTTP (202) have no
+        // receiver, so the dispatcher is the only place that can log them.
+        if row.reply_to.is_none() {
+            self.record_execution_log(&log_cx, &outcome, started, finished)
+                .await;
+        }
+
         match outcome {
             Ok(resp) => self.handle_success(&row, &resolved, resp).await,
             Err(err) => self.handle_failure(&row, &resolved, err).await,
         }
     }
 
+    /// G1: best-effort persistence of an execution-log row for a
+    /// dispatcher-run script. A sink failure is logged but never changes
+    /// ack/nack/retry behavior — the audit trail is not on the hot path
+    /// of message-delivery correctness.
+    async fn record_execution_log(
+        &self,
+        cx: &ExecLogContext,
+        outcome: &Result<ExecResponse, ExecError>,
+        started: DateTime<Utc>,
+        finished: DateTime<Utc>,
+    ) {
+        let log = build_execution_log(
+            cx.app_id,
+            cx.script_id,
+            cx.request_id,
+            cx.path.clone(),
+            cx.headers.clone(),
+            cx.body.clone(),
+            cx.source,
+            outcome,
+            started,
+            finished,
+        );
+        if let Err(e) = self.log_sink.record(log).await {
+            tracing::warn!(
+                error = %e,
+                script_id = %cx.script_id,
+                source = cx.source.as_str(),
+                "failed to persist trigger execution log"
+            );
+        }
+    }
+
     async fn resolve_trigger(&self, row: &OutboxRow) -> Result<ResolvedTrigger, DispatcherError> {
         // For KV and DL kinds, the outbox carries `trigger_id`. Use it
         // to look up the trigger row, then resolve the script.

crates/manager-core/src/email_service.rs

@@ -248,6 +248,15 @@ fn non_empty_env(key: &str) -> Option<String> {
     std::env::var(key).ok().filter(|v| !v.trim().is_empty())
 }
 
+/// `PICLOUD_DEV_MODE=true` (case-insensitive). Matches the detection in
+/// `shared::crypto` so the dev email sink and the dev master key turn on
+/// together.
+fn dev_mode_enabled() -> bool {
+    std::env::var("PICLOUD_DEV_MODE")
+        .map(|v| v.trim().eq_ignore_ascii_case("true"))
+        .unwrap_or(false)
+}
+
 /// Internal transport seam so the service can be tested without a live
 /// SMTP server. The production impl is [`LettreEmailTransport`]; tests
 /// use a recording fake.
@@ -299,6 +308,91 @@ impl EmailTransport for LettreEmailTransport {
     }
 }
 
+/// G5: how many recently-captured dev emails the in-memory sink keeps.
+/// Old entries are evicted FIFO; this is a debugging aid, not storage.
+pub const DEV_EMAIL_CAPACITY: usize = 100;
+
+/// One email captured by the dev sink instead of being relayed. Serialized
+/// straight onto the dev-only inspection endpoint.
+#[derive(Clone, serde::Serialize)]
+pub struct CapturedEmail {
+    pub captured_at: chrono::DateTime<chrono::Utc>,
+    pub from: Option<String>,
+    pub to: Vec<String>,
+    /// The full RFC 5322 message (headers + body), exactly as it would
+    /// have hit the relay — enough to eyeball subject/body in dev.
+    pub raw: String,
+}
+
+/// In-memory ring buffer of captured dev emails. Shared (`Arc`) between
+/// the [`DevEmailTransport`] that writes and the dev endpoint that reads.
+pub struct DevEmailSink {
+    captured: std::sync::Mutex<std::collections::VecDeque<CapturedEmail>>,
+    capacity: usize,
+}
+
+impl DevEmailSink {
+    #[must_use]
+    pub fn new(capacity: usize) -> Self {
+        Self {
+            captured: std::sync::Mutex::new(std::collections::VecDeque::new()),
+            capacity: capacity.max(1),
+        }
+    }
+
+    fn push(&self, email: CapturedEmail) {
+        let mut q = self.captured.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        while q.len() >= self.capacity {
+            q.pop_front();
+        }
+        q.push_back(email);
+    }
+
+    /// Newest-first snapshot of the captured mail.
+    #[must_use]
+    pub fn snapshot(&self) -> Vec<CapturedEmail> {
+        let q = self.captured.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        q.iter().rev().cloned().collect()
+    }
+}
+
+/// Dev transport: instead of relaying, capture the message in memory and
+/// log it. Wired only when `PICLOUD_DEV_MODE=true` and no SMTP relay is
+/// configured, so `email::send` is exercisable locally without a relay.
+/// NEVER constructed in production (no dev mode → disabled mode instead).
+pub struct DevEmailTransport {
+    sink: Arc<DevEmailSink>,
+}
+
+impl DevEmailTransport {
+    #[must_use]
+    pub fn new(sink: Arc<DevEmailSink>) -> Self {
+        Self { sink }
+    }
+}
+
+#[async_trait]
+impl EmailTransport for DevEmailTransport {
+    async fn send(&self, message: &Message) -> Result<(), EmailError> {
+        let envelope = message.envelope();
+        let from = envelope.from().map(ToString::to_string);
+        let to: Vec<String> = envelope.to().iter().map(ToString::to_string).collect();
+        let raw = String::from_utf8_lossy(&message.formatted()).into_owned();
+        tracing::info!(
+            ?from,
+            ?to,
+            "email DEV CAPTURE: message captured in memory (not relayed)"
+        );
+        self.sink.push(CapturedEmail {
+            captured_at: chrono::Utc::now(),
+            from,
+            to,
+            raw,
+        });
+        Ok(())
+    }
+}
+
 pub struct EmailServiceImpl {
     /// `None` → disabled mode (every send returns `NotConfigured`).
     transport: Option<Arc<dyn EmailTransport>>,
@@ -328,27 +422,58 @@ impl EmailServiceImpl {
     /// — email is non-critical and must not block startup.
     #[must_use]
     pub fn from_env(authz: Arc<dyn AuthzRepo>) -> Self {
+        Self::from_env_with_dev_capture(authz).0
+    }
+
+    /// Like [`from_env`](Self::from_env), but in **dev mode with no SMTP
+    /// relay** it wires a [`DevEmailTransport`] that captures mail in
+    /// memory instead of returning `NotConfigured` — so `email::send` is
+    /// exercisable locally (G5). Returns the sink handle (`Some`) when
+    /// capture mode is active, so the caller can expose it via the
+    /// dev-only inspection endpoint.
+    ///
+    /// Production is unaffected: without `PICLOUD_DEV_MODE=true` an unset
+    /// relay still yields disabled mode (`NotConfigured`), never capture.
+    #[must_use]
+    pub fn from_env_with_dev_capture(
+        authz: Arc<dyn AuthzRepo>,
+    ) -> (Self, Option<Arc<DevEmailSink>>) {
         let config = EmailConfig::from_env();
-        let transport: Option<Arc<dyn EmailTransport>> = match SmtpConfig::from_env() {
+        match SmtpConfig::from_env() {
+            Some(cfg) => {
+                let transport: Option<Arc<dyn EmailTransport>> = match LettreEmailTransport::build(
+                    &cfg,
+                ) {
+                    Ok(t) => {
+                        tracing::info!(host = %cfg.host, port = cfg.port, "outbound email enabled");
+                        Some(Arc::new(t))
+                    }
+                    Err(e) => {
+                        tracing::error!(error = %e, "failed to build SMTP transport; email DISABLED");
+                        None
+                    }
+                };
+                (Self::new(transport, authz, config), None)
+            }
+            None if dev_mode_enabled() => {
+                tracing::warn!(
+                    "email DEV CAPTURE: PICLOUD_DEV_MODE=true and no SMTP relay configured — \
+                     email::send will SUCCEED and capture messages in memory (last {DEV_EMAIL_CAPACITY}, \
+                     readable at GET /api/v1/admin/dev/emails). NEVER use this in production."
+                );
+                let sink = Arc::new(DevEmailSink::new(DEV_EMAIL_CAPACITY));
+                let transport: Arc<dyn EmailTransport> =
+                    Arc::new(DevEmailTransport::new(sink.clone()));
+                (Self::new(Some(transport), authz, config), Some(sink))
+            }
             None => {
                 tracing::warn!(
                     "email is DISABLED: set PICLOUD_SMTP_HOST/USER/PASSWORD to enable \
                      email::send. Scripts calling email::send will get an error."
                 );
-                None
+                (Self::new(None, authz, config), None)
             }
-            Some(cfg) => match LettreEmailTransport::build(&cfg) {
-                Ok(t) => {
-                    tracing::info!(host = %cfg.host, port = cfg.port, "outbound email enabled");
-                    Some(Arc::new(t))
-                }
-                Err(e) => {
-                    tracing::error!(error = %e, "failed to build SMTP transport; email DISABLED");
-                    None
-                }
-            },
-        };
-        Self::new(transport, authz, config)
+        }
     }
 
     async fn check_send(&self, cx: &SdkCallCx) -> Result<(), EmailError> {

crates/manager-core/src/kv_api.rs

@@ -0,0 +1,162 @@
+//! `/api/v1/admin/apps/{id}/kv*` — read-only KV inspection (G2).
+//!
+//! Mirrors the minimal `files_api` / `queues_api` admin surface so the
+//! `pic kv` CLI (and a future dashboard tab) can browse stored keys
+//! without a script. **Read-only by design** — KV writes go through
+//! `kv::set` in scripts, which emit change events the trigger framework
+//! depends on; an admin write would bypass that and could break app
+//! invariants, so it is deliberately out of scope here (matching the
+//! read-only queues precedent).
+//!
+//! Two operations:
+//!   * `GET /apps/{id}/kv?collection=<c>&cursor=&limit=` — list keys in a
+//!     collection (cursor-paginated).
+//!   * `GET /apps/{id}/kv/{collection}/{key}` — fetch one value.
+//!
+//! Capability: `AppKvRead`, resolved against the app loaded from the
+//! path (same tier the SDK read path uses).
+
+use std::sync::Arc;
+
+use axum::extract::{Path, Query, State};
+use axum::response::{IntoResponse, Json, Response};
+use axum::routing::get;
+use axum::{Extension, Router};
+use picloud_shared::{AppId, Principal};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+
+use crate::app_repo::AppRepository;
+use crate::authz::{require, AuthzDenied, AuthzRepo, Capability};
+use crate::kv_repo::KvRepo;
+
+#[derive(Clone)]
+pub struct KvAdminState {
+    pub kv: Arc<dyn KvRepo>,
+    pub apps: Arc<dyn AppRepository>,
+    pub authz: Arc<dyn AuthzRepo>,
+}
+
+pub fn kv_admin_router(state: KvAdminState) -> Router {
+    Router::new()
+        .route("/apps/{app_id}/kv", get(list_keys))
+        .route("/apps/{app_id}/kv/{collection}/{key}", get(get_value))
+        .with_state(state)
+}
+
+#[derive(Debug, Deserialize)]
+pub struct ListKvQuery {
+    pub collection: String,
+    #[serde(default)]
+    pub cursor: Option<String>,
+    #[serde(default)]
+    pub limit: Option<u32>,
+}
+
+#[derive(Debug, Serialize)]
+struct GetValueResponse {
+    value: serde_json::Value,
+}
+
+/// Serialize mirror of `shared::KvListPage` (which is not `Serialize`).
+#[derive(Debug, Serialize)]
+struct ListKeysResponse {
+    keys: Vec<String>,
+    next_cursor: Option<String>,
+}
+
+async fn list_keys(
+    State(s): State<KvAdminState>,
+    Extension(principal): Extension<Principal>,
+    Path(id_or_slug): Path<String>,
+    Query(q): Query<ListKvQuery>,
+) -> Result<Json<ListKeysResponse>, KvApiError> {
+    let app_id = resolve_app(&*s.apps, &id_or_slug).await?;
+    require(s.authz.as_ref(), &principal, Capability::AppKvRead(app_id)).await?;
+    let page =
+        s.kv.list(
+            app_id,
+            &q.collection,
+            q.cursor.as_deref(),
+            q.limit.unwrap_or(0),
+        )
+        .await
+        .map_err(|e| KvApiError::Backend(e.to_string()))?;
+    Ok(Json(ListKeysResponse {
+        keys: page.keys,
+        next_cursor: page.next_cursor,
+    }))
+}
+
+async fn get_value(
+    State(s): State<KvAdminState>,
+    Extension(principal): Extension<Principal>,
+    Path((id_or_slug, collection, key)): Path<(String, String, String)>,
+) -> Result<Json<GetValueResponse>, KvApiError> {
+    let app_id = resolve_app(&*s.apps, &id_or_slug).await?;
+    require(s.authz.as_ref(), &principal, Capability::AppKvRead(app_id)).await?;
+    let value =
+        s.kv.get(app_id, &collection, &key)
+            .await
+            .map_err(|e| KvApiError::Backend(e.to_string()))?
+            .ok_or(KvApiError::NotFound)?;
+    Ok(Json(GetValueResponse { value }))
+}
+
+async fn resolve_app(apps: &dyn AppRepository, ident: &str) -> Result<AppId, KvApiError> {
+    crate::app_repo::resolve_app(apps, ident)
+        .await
+        .map_err(|e| KvApiError::Backend(e.to_string()))?
+        .map(|l| l.app.id)
+        .ok_or(KvApiError::AppNotFound)
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum KvApiError {
+    #[error("app not found")]
+    AppNotFound,
+    #[error("key not found")]
+    NotFound,
+    #[error("forbidden")]
+    Forbidden,
+    #[error("authorization repo error: {0}")]
+    AuthzRepo(String),
+    #[error("kv backend: {0}")]
+    Backend(String),
+}
+
+impl From<AuthzDenied> for KvApiError {
+    fn from(d: AuthzDenied) -> Self {
+        match d {
+            AuthzDenied::Denied => Self::Forbidden,
+            AuthzDenied::Repo(e) => Self::AuthzRepo(e.to_string()),
+        }
+    }
+}
+
+impl IntoResponse for KvApiError {
+    fn into_response(self) -> Response {
+        use axum::http::StatusCode;
+        let (status, body) = match &self {
+            Self::AppNotFound | Self::NotFound => {
+                (StatusCode::NOT_FOUND, json!({ "error": self.to_string() }))
+            }
+            Self::Forbidden => (StatusCode::FORBIDDEN, json!({ "error": self.to_string() })),
+            Self::AuthzRepo(e) => {
+                tracing::error!(error = %e, "kv admin authz error");
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    json!({ "error": "internal error" }),
+                )
+            }
+            Self::Backend(e) => {
+                tracing::error!(error = %e, "kv admin backend error");
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    json!({ "error": "internal error" }),
+                )
+            }
+        };
+        (status, Json(body)).into_response()
+    }
+}

crates/manager-core/src/lib.rs

@@ -33,6 +33,7 @@ pub mod cron_scheduler;
 pub mod dead_letter_repo;
 pub mod dead_letter_service;
 pub mod dead_letters_api;
+pub mod dev_email_api;
 pub mod dispatcher;
 pub mod docs_filter;
 pub mod docs_repo;
@@ -46,6 +47,7 @@ pub mod files_sweep;
 pub mod gc;
 pub mod http_service;
 pub mod invoke_service;
+pub mod kv_api;
 pub mod kv_repo;
 pub mod kv_service;
 pub mod log_sink;
@@ -143,6 +145,7 @@ pub use dead_letter_repo::{
 };
 pub use dead_letter_service::PostgresDeadLetterService;
 pub use dead_letters_api::{dead_letters_router, DeadLettersApiError, DeadLettersState};
+pub use dev_email_api::{dev_emails_router, DevEmailState};
 pub use dispatcher::{compute_backoff, Dispatcher, DispatcherError};
 pub use docs_repo::{DocsRepo, DocsRepoError, PostgresDocsRepo};
 pub use docs_service::DocsServiceImpl;
@@ -150,8 +153,8 @@ pub use email_inbound_api::{
     email_inbound_router, EmailInboundError, EmailInboundState, InboundNonceDedup,
 };
 pub use email_service::{
-    EmailConfig, EmailServiceImpl, EmailTransport, LettreEmailTransport, SmtpConfig, SmtpTls,
-    DEFAULT_EMAIL_MAX_MESSAGE_BYTES,
+    CapturedEmail, DevEmailSink, DevEmailTransport, EmailConfig, EmailServiceImpl, EmailTransport,
+    LettreEmailTransport, SmtpConfig, SmtpTls, DEFAULT_EMAIL_MAX_MESSAGE_BYTES,
 };
 pub use files_api::{files_admin_router, FilesAdminState};
 pub use files_repo::{FilesConfig, FilesRepo, FilesRepoError, FsFilesRepo};
@@ -159,6 +162,7 @@ pub use files_service::FilesServiceImpl;
 pub use files_sweep::{spawn_files_orphan_sweep, sweep_orphan_tmp_files, SweepStats};
 pub use gc::{spawn_abandoned_gc, spawn_app_user_token_gc, spawn_dead_letter_gc};
 pub use http_service::{HttpConfig, HttpServiceImpl};
+pub use kv_api::{kv_admin_router, KvAdminState};
 pub use kv_repo::{KvRepo, KvRepoError, PostgresKvRepo};
 pub use kv_service::KvServiceImpl;
 pub use log_sink::PostgresExecutionLogSink;

crates/manager-core/src/log_sink.rs

@@ -31,9 +31,9 @@ impl ExecutionLogSink for PostgresExecutionLogSink {
                 id, app_id, script_id, request_id, \
                 request_path, request_headers, request_body, \
                 response_code, response_body, \
-                logs, duration_ms, status, created_at \
+                logs, duration_ms, status, source, created_at \
              ) VALUES ( \
-                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13 \
+                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14 \
              )",
         )
         .bind(log.id)
@@ -48,6 +48,7 @@ impl ExecutionLogSink for PostgresExecutionLogSink {
         .bind(&log.script_logs)
         .bind(duration_ms)
         .bind(log.status.as_str())
+        .bind(log.source.as_str())
         .bind(log.created_at)
         .execute(&self.pool)
         .await

crates/manager-core/src/repo.rs

@@ -3,8 +3,8 @@ use std::collections::BTreeMap;
 use async_trait::async_trait;
 use picloud_orchestrator_core::{ResolverError, ScriptResolver};
 use picloud_shared::{
-    AdminUserId, AppId, ExecutionLog, ExecutionStatus, RequestId, Script, ScriptId, ScriptKind,
-    ScriptSandbox,
+    AdminUserId, AppId, ExecutionLog, ExecutionSource, ExecutionStatus, RequestId, Script,
+    ScriptId, ScriptKind, ScriptSandbox,
 };
 use sqlx::PgPool;
 
@@ -584,6 +584,7 @@ pub trait ExecutionLogRepository: Send + Sync {
         script_id: ScriptId,
         limit: i64,
         cursor: Option<ExecutionLogCursor>,
+        source: Option<ExecutionSource>,
     ) -> Result<Vec<ExecutionLog>, ScriptRepositoryError>;
 }
 
@@ -605,23 +606,30 @@ impl ExecutionLogRepository for PostgresExecutionLogRepository {
         script_id: ScriptId,
         limit: i64,
         cursor: Option<ExecutionLogCursor>,
+        source: Option<ExecutionSource>,
     ) -> Result<Vec<ExecutionLog>, ScriptRepositoryError> {
+        // The optional `source` filter is folded into one bind via
+        // `$N::text IS NULL OR source = $N` so we don't fan out into four
+        // query strings. `None` → the predicate is always true (no filter).
+        let source = source.map(ExecutionSource::as_str);
         let rows = match cursor {
             Some(c) => {
                 sqlx::query_as::<_, ExecutionLogRow>(
                     "SELECT id, app_id, script_id, request_id, \
                             request_path, request_headers, request_body, \
                             response_code, response_body, \
-                            logs, duration_ms, status, created_at \
+                            logs, duration_ms, status, source, created_at \
                      FROM execution_logs \
                      WHERE script_id = $1 \
                        AND (created_at, id) < ($2, $3) \
+                       AND ($4::text IS NULL OR source = $4) \
                      ORDER BY created_at DESC, id DESC \
-                     LIMIT $4",
+                     LIMIT $5",
                 )
                 .bind(script_id.into_inner())
                 .bind(c.created_at)
                 .bind(c.id)
+                .bind(source)
                 .bind(limit)
                 .fetch_all(&self.pool)
                 .await?
@@ -631,13 +639,15 @@ impl ExecutionLogRepository for PostgresExecutionLogRepository {
                     "SELECT id, app_id, script_id, request_id, \
                             request_path, request_headers, request_body, \
                             response_code, response_body, \
-                            logs, duration_ms, status, created_at \
+                            logs, duration_ms, status, source, created_at \
                      FROM execution_logs \
                      WHERE script_id = $1 \
+                       AND ($2::text IS NULL OR source = $2) \
                      ORDER BY created_at DESC, id DESC \
-                     LIMIT $2",
+                     LIMIT $3",
                 )
                 .bind(script_id.into_inner())
+                .bind(source)
                 .bind(limit)
                 .fetch_all(&self.pool)
                 .await?
@@ -662,6 +672,7 @@ struct ExecutionLogRow {
     logs: serde_json::Value,
     duration_ms: i32,
     status: String,
+    source: String,
     created_at: chrono::DateTime<chrono::Utc>,
 }
 
@@ -675,6 +686,9 @@ impl From<ExecutionLogRow> for ExecutionLog {
             "budget_exceeded" => ExecutionStatus::BudgetExceeded,
             _ => ExecutionStatus::Error,
         };
+        // Unknown values can't occur (CHECK constraint) but default to
+        // Http rather than panicking on a forward-compat surprise.
+        let source = ExecutionSource::from_wire(&r.source).unwrap_or_default();
         Self {
             id: r.id,
             app_id: r.app_id.into(),
@@ -688,6 +702,7 @@ impl From<ExecutionLogRow> for ExecutionLog {
             script_logs: r.logs,
             duration_ms: u64::try_from(r.duration_ms).unwrap_or(0),
             status,
+            source,
             created_at: r.created_at,
         }
     }

crates/manager-core/src/route_admin.rs

@@ -373,27 +373,56 @@ async fn refresh_table<RR: RouteRepository, SR: ScriptRepository>(
     state: &RouteAdminState<RR, SR>,
 ) -> Result<(), RouteApiError> {
     let rows = state.routes.list_all().await?;
-    let compiled = compile_routes(&rows)?;
+    let compiled = compile_routes(&rows);
     state.table.replace_all(compiled);
     Ok(())
 }
 
-pub fn compile_routes(rows: &[Route]) -> Result<Vec<CompiledRoute>, pattern::ParseError> {
+/// Compile stored route rows into the in-memory match table.
+///
+/// **Lenient by design (H1).** A row that fails to parse is *skipped with
+/// a warning*, not propagated as an error. The motivating case: a path
+/// that was valid when created but became reserved under a later, stricter
+/// validation (e.g. the case-insensitive reserved-prefix check) — but this
+/// also covers any other parse failure. A single un-compilable legacy row
+/// must never take down the entire data plane: this function runs at
+/// startup (where a hard error aborts boot) and on every table rebuild
+/// after a route edit (where it would fail an unrelated CRUD op). A skipped
+/// route simply doesn't match; the warning tells the operator to delete or
+/// fix it (and migration 0044 sweeps the reserved-path offenders on
+/// upgrade).
+#[must_use]
+pub fn compile_routes(rows: &[Route]) -> Vec<CompiledRoute> {
     rows.iter()
-        .map(|r| {
-            Ok(CompiledRoute {
-                route_id: r.id,
-                app_id: r.app_id,
-                script_id: r.script_id,
-                host: pattern::parse_host(r.host_kind, &r.host, r.host_param_name.as_deref())?,
-                path: pattern::parse_path(r.path_kind, &r.path)?,
-                method: r.method.clone(),
-                dispatch_mode: r.dispatch_mode,
-            })
+        .filter_map(|r| match compile_route(r) {
+            Ok(compiled) => Some(compiled),
+            Err(e) => {
+                tracing::warn!(
+                    route_id = %r.id,
+                    app_id = %r.app_id,
+                    path = %r.path,
+                    error = %e,
+                    "skipping un-compilable stored route — it will not match; \
+                     delete or fix it"
+                );
+                None
+            }
         })
         .collect()
 }
 
+fn compile_route(r: &Route) -> Result<CompiledRoute, pattern::ParseError> {
+    Ok(CompiledRoute {
+        route_id: r.id,
+        app_id: r.app_id,
+        script_id: r.script_id,
+        host: pattern::parse_host(r.host_kind, &r.host, r.host_param_name.as_deref())?,
+        path: pattern::parse_path(r.path_kind, &r.path)?,
+        method: r.method.clone(),
+        dispatch_mode: r.dispatch_mode,
+    })
+}
+
 /// Validate that a new route's (host_kind, host) is consistent with at
 /// least one of the parent app's domain claims. `HostKind::Any` is
 /// always permitted — it catches every host the app already owns.
@@ -577,3 +606,49 @@ impl IntoResponse for RouteApiError {
         (status, Json(body)).into_response()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use picloud_shared::DispatchMode;
+    use uuid::Uuid;
+
+    fn route_with_path(path: &str) -> Route {
+        Route {
+            id: Uuid::new_v4(),
+            app_id: AppId::from(Uuid::new_v4()),
+            script_id: ScriptId::from(Uuid::new_v4()),
+            host_kind: HostKind::Any,
+            host: String::new(),
+            host_param_name: None,
+            path_kind: PathKind::Exact,
+            path: path.to_string(),
+            method: None,
+            dispatch_mode: DispatchMode::default(),
+            created_at: chrono::Utc::now(),
+        }
+    }
+
+    #[test]
+    fn compile_routes_skips_uncompilable_rows_instead_of_failing() {
+        // H1 regression guard: a stored route whose path is now reserved
+        // (creatable before the case-insensitive reserved-prefix fix) must
+        // be skipped, not abort the whole compile — otherwise one legacy
+        // row bricks startup (`compile_routes` runs in `build_app`).
+        let good_a = route_with_path("/ok");
+        let bad = route_with_path("/API/v2/x"); // now reserved, case-insensitive
+        let good_b = route_with_path("/items");
+        let rows = vec![good_a.clone(), bad.clone(), good_b.clone()];
+
+        let compiled = compile_routes(&rows);
+
+        let ids: Vec<Uuid> = compiled.iter().map(|c| c.route_id).collect();
+        assert_eq!(compiled.len(), 2, "the reserved row must be dropped");
+        assert!(ids.contains(&good_a.id));
+        assert!(ids.contains(&good_b.id));
+        assert!(
+            !ids.contains(&bad.id),
+            "a reserved-path route must be skipped, never abort the compile"
+        );
+    }
+}

crates/manager-core/tests/expected_schema.txt

@@ -194,6 +194,7 @@ table: execution_logs
   status: text NOT NULL
   created_at: timestamp with time zone NOT NULL default=now()
   app_id: uuid NOT NULL
+  source: text NOT NULL default='http'::text
 
 table: files
   app_id: uuid NOT NULL
@@ -590,6 +591,7 @@ constraints on email_trigger_details:
   [PRIMARY KEY] email_trigger_details_pkey: PRIMARY KEY (trigger_id)
 
 constraints on execution_logs:
+  [CHECK] execution_logs_source_check: CHECK ((source = ANY (ARRAY['http'::text, 'kv'::text, 'docs'::text, 'dead_letter'::text, 'cron'::text, 'files'::text, 'pubsub'::text, 'email'::text, 'invoke'::text, 'queue'::text])))
   [CHECK] execution_logs_status_check: CHECK ((status = ANY (ARRAY['success'::text, 'error'::text, 'timeout'::text, 'budget_exceeded'::text])))
   [FOREIGN KEY] execution_logs_app_id_fk: FOREIGN KEY (app_id) REFERENCES apps(id) ON DELETE CASCADE
   [FOREIGN KEY] execution_logs_script_id_fkey: FOREIGN KEY (script_id) REFERENCES scripts(id) ON DELETE SET NULL
@@ -711,3 +713,5 @@ constraints on triggers:
   0040: execution logs keep history
   0041: dead letters composite idx
   0042: secrets envelope version
+  0043: execution logs source
+  0044: delete reserved path routes