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>

CLAUDE.md

@@ -117,7 +117,7 @@ Environment variables consumed by the `picloud` binary:
 | `PICLOUD_MAX_CONCURRENT_EXECUTIONS` | `32` | Global concurrency cap on data-plane script executions. Overflow returns HTTP 503 with `Retry-After: 1` immediately (no queue). |
 | `DATABASE_URL` | — | Required. Postgres connection string. |
 | `PICLOUD_SECRET_KEY` | — | Master encryption key (base64). Required at startup unless dev mode is acknowledged (below). |
-| `PICLOUD_DEV_MODE` | `false` | `true` enables local-dev conveniences. Without `PICLOUD_SECRET_KEY` it ALSO requires the acknowledgement var below — `PICLOUD_DEV_MODE=true` alone aborts at startup. |
+| `PICLOUD_DEV_MODE` | `false` | `true` enables local-dev conveniences. Without `PICLOUD_SECRET_KEY` it ALSO requires the acknowledgement var below — `PICLOUD_DEV_MODE=true` alone aborts at startup. Also: when no SMTP relay is configured, `email::send` switches from disabled (`NotConfigured`) to an **in-memory dev sink** — sends succeed and the last 100 messages are readable at `GET /api/v1/admin/dev/emails` (instance Owner/Admin only; route exists only in this mode). Never in production. |
 | `PICLOUD_DEV_INSECURE_KEY` | — | Set to the literal `i-understand-this-is-insecure` to let dev mode boot without `PICLOUD_SECRET_KEY`, using a deterministic, world-known dev master key. Never set in production — it would encrypt everything with a public value. |
 | `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 session lifetime. |

E2E_STASH_REPORT.md

@@ -0,0 +1,138 @@
+# E2E Test #2 — "Stash" (paste + file-drop) + security matrix
+
+**Date:** 2026-06-13 · **Build:** v1.1.9 · **Instance:** host `picloud` on `127.0.0.1:8099`
+(dev mode, `PICLOUD_FILES_MAX_FILE_SIZE_BYTES=1048576`, SSRF guard on, SMTP unset), Postgres in
+Docker. Driven entirely through the `pic` CLI; data-plane traffic via `curl`.
+
+## Goal
+
+Second end-to-end pass. The first test (To-Do, `E2E_TODO_REPORT.md`) covered users/docs/pubsub/
+cron/routes/domains/topics. This one builds a **new** app to exercise the *untested* half of the
+platform — **kv, files, queue, invoke, secrets, api-keys, dead-letters, async + two-param routes**
+— and then runs a **10-probe security matrix**. Findings: feature/CLI gaps + security verdicts.
+
+## Verdict
+
+**Every feature worked** end-to-end, and **every security control held except one Low-severity
+gap** (reserved-path validation is case-sensitive). Highlights of what's *good*: cross-app
+isolation is airtight, the SSRF guard blocks cloud-metadata, size/op caps fire before authz, and
+internal script errors are scrubbed from responses with a correlation id. Two notable
+**observability/ergonomics gaps** (trigger executions invisible to `pic logs`; the `.trim()`/`.replace()`
+return-`()` footgun) and the expected CLI-coverage gaps round it out.
+
+---
+
+## The app — "Stash" (built CLI-only, no raw admin curl)
+
+A pastebin with file attachments and a background word-count worker. 9 scripts, 6 routes, a queue
+consumer trigger, a dead-letter trigger, 2 API keys, 1 secret — all via `pic`.
+
+| Feature exercised | How | Result |
+|---|---|---|
+| **kv** | paste store, hit counter, `/stats` aggregates, event log | ✅ get/set/list round-trip |
+| **files** | `files::create` from base64 blob + `get`/`head`, two-param download route | ✅ bytes round-trip (`"hello attachment bytes"`, size 22) |
+| **queue** | `queue::enqueue("ingest")` + `create-from-json --kind queue` consumer | ✅ worker fired |
+| **invoke** | worker → `invoke("analyzer", #{text})` function-to-function | ✅ returns word count |
+| **secrets** | `pic secrets set` (stdin) + `secrets::get` to gate `/stats` | ✅ 401 w/o token, 200 with |
+| **async route** | `routes create --dispatch async` for `POST /events` | ✅ **202** + execution_id |
+| **dead-letters** | poison job → 3 attempts → DL row; `pic dead-letters replay` | ✅ captured + replayed (+9 words) |
+| **api-keys** | `pic api-keys mint --scope … [--app …]` | ✅ minted, scopes enforced (below) |
+| **two-param route** | `/pastes/:code/files/:fid` | ✅ both params captured |
+
+End-to-end stats after the run: `{total_pastes: 2, total_words: 13}` — proving queue → worker →
+invoke → kv all fired on real Postgres rows + on-disk file bytes.
+
+---
+
+## Security matrix (10 probes, all reproduced live)
+
+| # | Probe | Verdict | Evidence |
+|---|---|---|---|
+| **S1** | Cross-app isolation | ✅ HOLDS | A script in app `evil` doing `kv…list()` / `secrets::get("admin-token")` / `files…list()` returned `{paste_keys:[], attachment_count:0, stolen_secret:null, my_secret_names:[]}`. `app_id` derives from `cx.app_id`; no SDK call takes a script-passed app_id. |
+| **S2** | Secret confidentiality | ✅ HOLDS | `pic secrets ls` shows names only; the plaintext `sup3r-s3cret-admin` does **not** appear anywhere in the server log even though `/stats` reads it each request. |
+| **S3** | API-key scope + app-binding | ✅ HOLDS | `script:read` key → **403** on `POST /apps` and `GET /admins`, **200** on in-scope `GET /scripts`. `app:admin` key bound to `stash` → **403** on app `evil`. `instance:admin` + `--app` → **422**. Unknown scope `bogus:scope` → rejected by CLI. |
+| **S4** | Sandbox op-budget | ✅ HOLDS | `loop {}` → **HTTP 507 "execution exceeded operation budget"** in **0.30 s**; `/healthz` still `ok` (no hang). |
+| **S5** | Value/size caps | ✅ HOLDS | Oversized kv write rejected (HTTP 502). Notably the detail (`"Length of string too large"`) was **scrubbed from the response** and logged with `correlation_id` — good error hygiene. Caps are checked before authz, so anonymous callers can't DoS. |
+| **S6** | Reserved-path prefixes | ⚠️ **GAP (Low)** | `/api/x`, `/admin/x`, `/healthz`, `/version` → correctly **422**. But case variants `/API/v2/x`, `/Admin/x`, `/HEALTHZ` were **accepted**. See finding below. |
+| **S7** | Files path traversal | ✅ HOLDS | `files::collection("../../../etc")` → `"invalid collection name: must not contain '/', '\\', '..', or NUL"`; a non-UUID file id → not found. No FS escape. |
+| **S8** | docs filter injection | ✅ HOLDS | `docs::find(#{ "x'); DROP TABLE docs;--": 1 })` → ran safely, `count=0`; the `docs` table still exists (`to_regclass` = true). Values/paths are bound as `$N` params. |
+| **S9** | SSRF via `http::` | ✅ HOLDS | `http::get("http://127.0.0.1:8099/healthz")` → `"blocked by SSRF policy: loopback"`; `http://169.254.169.254/…` (cloud metadata) → `"blocked by SSRF policy: link-local"`. |
+| **S10** | Anonymous data-plane access | ℹ️ By design (caveat) | Every Stash route is public (no auth) yet freely uses kv/secrets/files/queue. `script_gate` returns `Ok` when `principal.is_none()` — the *script* is the gate. Correct per design, but a sharp threat-model edge (see note). |
+
+### S6 — reserved-path validation is case-sensitive (Low)
+**What:** route-creation rejects the exact reserved prefixes but accepts case variants:
+```
+pic routes create --path /api/x   -> 422 "path '/api/x' is reserved"
+pic routes create --path /API/v2/x -> Created route … (GET * /API/v2/x)   # accepted
+pic routes create --path /HEALTHZ  -> Created route …                      # accepted
+pic routes create --path /Admin/x  -> Created route …                      # accepted
+```
+**Current impact — Low, *not* a full bypass:** path matching is case-*sensitive*, so a request to
+the real lowercase `/api/v2/x` still 404s (the system namespace is safe); only the exact
+mixed-case path serves the script (`GET /API/v2/x` → 200, `/HEALTHZ` → 200). Real `/healthz`
+still returns `ok` from the top-level handler.
+**Why it still matters:** (1) inconsistent enforcement of a stated security boundary; (2) lets a
+tenant publish convincing look-alike paths (`/Admin/login`, `/API/v1/…`) for phishing or log
+confusion; (3) **fragile** — method matching is already case-insensitive (`matcher.rs:275`); if
+path matching is ever made case-insensitive too, this instantly becomes a real shadowing bypass
+of `/api/`, `/admin/`, `/healthz`. The validation should be the durable guard.
+**Fix:** normalize case before the reserved check in `orchestrator-core/src/routing/pattern.rs`
+(`check_reserved` ~line 110) — compare `raw.to_ascii_lowercase()` against the reserved list (or
+reject any case-insensitive match).
+
+### S10 — anonymous public scripts have full app data-plane access (design caveat)
+Not a bug, but worth surfacing for users: a *public* (unauthenticated) route's script can read/write
+**all** of the app's kv, docs, files, secrets, and queue — the platform only gates *authenticated*
+principals; for anonymous ingress the script itself must enforce access. A developer who assumes
+"this route is public" ≠ "this code can read every secret in my app" could over-expose data. Worth
+a prominent doc callout near the SDK auth model.
+
+---
+
+## Feature / CLI gaps
+
+- **G1 — trigger executions are invisible to `pic logs`.** After successful queue-worker and
+  analyzer runs, `pic logs <worker>` and `pic logs <analyzer>` were **empty**. Trigger-dispatched
+  executions (queue/cron/dead-letter/invoke) don't surface in the per-script log tail — the only
+  built-in visibility into worker activity is **dead-letters (failures only)** or the script's own
+  side effects. This is a real observability gap for background workloads. *Suggest: include
+  trigger executions in the logs surface, or add a `pic logs --trigger`/events view.*
+- **G2 — no `pic kv` / `files` / `queues` / `members` commands.** Data-plane stores are
+  script/HTTP-only (no admin CLI to inspect kv/files/queue contents; queues are read-only HTTP).
+  App **membership** (`/apps/{id}/members`) also has no CLI — multi-user app roles can't be managed
+  with `pic`. *(Confirmed absent via `pic <cmd> --help`.)*
+- **G3 — no per-script sandbox/timeout flags in `pic deploy`/`scripts`.** `timeout_seconds`,
+  `memory_limit_mb`, and sandbox overrides are only settable via the raw scripts API or instance
+  env (`PICLOUD_SANDBOX_MAX_*`). A developer can't cap a single script's runtime from the CLI.
+- **G4 — uneven trigger wrappers.** Only `kv`, `cron`, `dead-letter` have first-class
+  `pic triggers create-*` wrappers; `docs`/`files`/`pubsub`/`queue`/`email` require hand-built
+  `create-from-json --kind … --body '{…}'`. Works, but the developer must know each body schema.
+- **G5 — `email::send` unusable in dev.** Returns `"email is not configured: set
+  PICLOUD_SMTP_HOST/USER/PASSWORD"`. Expected (no silent drop), but the email feature can't be
+  exercised at all without an SMTP relay — worth a documented local-dev fake/sink.
+- **G6 — Rhai `.trim()` returns `()` (in-place mutation footgun).** Same family as `.replace()`
+  (already documented). `let t = text.trim()` sets `t` to unit; my analyzer dead-lettered with
+  `Function not found: split((), …)` until rewritten to `text.trim();` as a statement. *Suggest:
+  extend the existing stdlib footgun note to list `trim`/`to_upper`/`to_lower` etc., not just
+  `replace`.*
+
+## Things done especially well (no action)
+- Cross-app isolation has no script-controlled `app_id` anywhere — the boundary is structural.
+- SSRF guard covers loopback **and** link-local (cloud-metadata `169.254.169.254`), and blocks at
+  every redirect hop.
+- Size/operation caps are enforced **before** authz, so anonymous public scripts can't DoS the DB.
+- Script runtime errors are **scrubbed** from the HTTP response and logged with a `correlation_id`
+  — no internal detail leaks to the caller.
+- `deploy`/`apps create` honor `--output json` (the earlier F1 fix), so the whole build scripts
+  cleanly with captured ids.
+
+## Reproduction / teardown
+Scripts under `/tmp/stash/*.rhai`; apps `stash` + `evil` and their data persist in the dev
+Postgres. Server: `target/debug/picloud` on `:8099` (see Phase 0 env). Teardown: kill the host
+`picloud`; `docker compose down [-v]`. (Note: the S6 probe left harmless `/API/v2/x`, `/HEALTHZ`,
+`/Admin/x`, `/Api/y` routes on `stash` — they vanish with the app.)
+
+## Priority
+**S6** is the only code-level finding (Low — confirm-and-harden the reserved-path check). **G1**
+(trigger-execution observability) is the most impactful *developer-experience* gap for anyone
+running background workers. Everything else is documented-behavior or known CLI-coverage gaps.

crates/executor-core/src/lib.rs

@@ -19,5 +19,6 @@ pub use module_resolver::{
 };
 pub use sandbox::Limits;
 pub use types::{
-    ExecError, ExecRequest, ExecResponse, ExecStats, InvocationType, LogEntry, LogLevel,
+    build_execution_log, ExecError, ExecRequest, ExecResponse, ExecStats, InvocationType, LogEntry,
+    LogLevel,
 };

crates/executor-core/src/types.rs

@@ -2,10 +2,12 @@ use std::collections::BTreeMap;
 
 use chrono::{DateTime, Utc};
 use picloud_shared::{
-    AppId, ExecutionId, Principal, RequestId, ScriptId, ScriptSandbox, TriggerEvent,
+    AppId, ExecutionId, ExecutionLog, ExecutionSource, ExecutionStatus, Principal, RequestId,
+    ScriptId, ScriptSandbox, TriggerEvent,
 };
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
+use uuid::Uuid;
 
 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "lowercase")]
@@ -167,3 +169,77 @@ pub enum ExecError {
     #[error("execution declined: server at capacity (retry after {retry_after_secs}s)")]
     Overloaded { retry_after_secs: u32 },
 }
+
+/// Build an `ExecutionLog` row from one invocation's outcome.
+///
+/// Shared by every execution path so the log shape stays identical
+/// regardless of who ran the script: the orchestrator's sync/direct HTTP
+/// handlers pass `ExecutionSource::Http`, while the manager's trigger
+/// dispatcher passes the trigger's kind (`Kv`, `Cron`, `Invoke`, …). That
+/// `source` is what lets `pic logs` surface background runs that were
+/// previously invisible. `Overloaded` is never logged — admission is
+/// refused before a row exists — but it maps to `Error` defensively.
+#[allow(clippy::too_many_arguments)]
+#[must_use]
+pub fn build_execution_log(
+    app_id: AppId,
+    script_id: ScriptId,
+    request_id: RequestId,
+    request_path: String,
+    request_headers: BTreeMap<String, String>,
+    request_body: serde_json::Value,
+    source: ExecutionSource,
+    outcome: &Result<ExecResponse, ExecError>,
+    started: DateTime<Utc>,
+    finished: DateTime<Utc>,
+) -> ExecutionLog {
+    let duration_ms = u64::try_from(
+        finished
+            .signed_duration_since(started)
+            .num_milliseconds()
+            .max(0),
+    )
+    .unwrap_or(0);
+
+    let (status, response_code, response_body, script_logs) = match outcome {
+        Ok(resp) => {
+            let logs = serde_json::to_value(&resp.logs).unwrap_or(serde_json::Value::Array(vec![]));
+            (
+                ExecutionStatus::Success,
+                Some(resp.status_code),
+                Some(resp.body.clone()),
+                logs,
+            )
+        }
+        Err(e) => {
+            let status = match e {
+                ExecError::Timeout(_) => ExecutionStatus::Timeout,
+                ExecError::OperationBudgetExceeded => ExecutionStatus::BudgetExceeded,
+                _ => ExecutionStatus::Error,
+            };
+            (
+                status,
+                None,
+                Some(serde_json::json!({ "error": e.to_string() })),
+                serde_json::Value::Array(vec![]),
+            )
+        }
+    };
+
+    ExecutionLog {
+        id: Uuid::new_v4(),
+        app_id,
+        script_id,
+        request_id,
+        request_path,
+        request_headers,
+        request_body,
+        response_code,
+        response_body,
+        script_logs,
+        duration_ms,
+        status,
+        source,
+        created_at: started,
+    }
+}

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,
+    AppId, ExecutionLog, ExecutionSource, InstanceRole, Principal, Script, ScriptId, ScriptKind,
-    ScriptValidator, ValidatedScript, ValidationError,
+    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,
+    DeadLetterId, ExecResponseSummary, ExecutionId, ExecutionLogSink, ExecutionSource,
-    InboxFailureKind, InboxResolver, InboxResult, RequestId, ScriptId, ScriptSandbox, TriggerEvent,
+    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,
+    CapturedEmail, DevEmailSink, DevEmailTransport, EmailConfig, EmailServiceImpl, EmailTransport,
-    DEFAULT_EMAIL_MAX_MESSAGE_BYTES,
+    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,
+    AdminUserId, AppId, ExecutionLog, ExecutionSource, ExecutionStatus, RequestId, Script,
-    ScriptSandbox,
+    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| {
+        .filter_map(|r| match compile_route(r) {
-            Ok(CompiledRoute {
+            Ok(compiled) => Some(compiled),
-                route_id: r.id,
+            Err(e) => {
-                app_id: r.app_id,
+                tracing::warn!(
-                script_id: r.script_id,
+                    route_id = %r.id,
-                host: pattern::parse_host(r.host_kind, &r.host, r.host_param_name.as_deref())?,
+                    app_id = %r.app_id,
-                path: pattern::parse_path(r.path_kind, &r.path)?,
+                    path = %r.path,
-                method: r.method.clone(),
+                    error = %e,
-                dispatch_mode: r.dispatch_mode,
+                    "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

crates/orchestrator-core/src/api.rs

@@ -15,11 +15,13 @@ use axum::{
     Extension, Json, Router,
 };
 use chrono::Utc;
-use picloud_executor_core::{ExecError, ExecRequest, ExecResponse, InvocationType};
+use picloud_executor_core::{
+    build_execution_log, ExecError, ExecRequest, ExecResponse, InvocationType,
+};
 use picloud_shared::{
-    AppId, DispatchMode, ExecutionId, ExecutionLog, ExecutionLogSink, ExecutionStatus,
+    AppId, DispatchMode, ExecutionId, ExecutionLog, ExecutionLogSink, ExecutionSource,
-    HttpDispatchPayload, InboxFailureKind, InboxResult, NewHttpOutbox, OutboxWriter, Principal,
+    ExecutionStatus, HttpDispatchPayload, InboxFailureKind, InboxResult, NewHttpOutbox,
-    RequestId, ScriptId,
+    OutboxWriter, Principal, RequestId, ScriptId,
 };
 use serde_json::Value as Json_;
 use uuid::Uuid;
@@ -150,6 +152,7 @@ where
         request_path,
         request_headers,
         request_body,
+        ExecutionSource::Http,
         &outcome,
         started,
         finished,
@@ -530,6 +533,7 @@ fn build_inbox_execution_log(
         script_logs: Json_::Array(vec![]),
         duration_ms,
         status,
+        source: ExecutionSource::Http,
         created_at: started,
     }
 }
@@ -630,67 +634,9 @@ fn exec_response_to_http(resp: ExecResponse) -> Response {
     (status, http_headers, Json(resp.body)).into_response()
 }
 
-#[allow(clippy::too_many_arguments)]
+// `build_execution_log` moved to `picloud_executor_core` so the manager's
-fn build_execution_log(
+// trigger dispatcher can build identically-shaped rows (G1 — trigger
-    app_id: AppId,
+// executions are now logged with their `ExecutionSource`).
-    script_id: ScriptId,
-    request_id: RequestId,
-    request_path: String,
-    request_headers: BTreeMap<String, String>,
-    request_body: Json_,
-    outcome: &Result<ExecResponse, ExecError>,
-    started: chrono::DateTime<Utc>,
-    finished: chrono::DateTime<Utc>,
-) -> ExecutionLog {
-    let duration_ms = u64::try_from(
-        finished
-            .signed_duration_since(started)
-            .num_milliseconds()
-            .max(0),
-    )
-    .unwrap_or(0);
-
-    let (status, response_code, response_body, script_logs) = match outcome {
-        Ok(resp) => {
-            let logs = serde_json::to_value(&resp.logs).unwrap_or(Json_::Array(vec![]));
-            (
-                ExecutionStatus::Success,
-                Some(resp.status_code),
-                Some(resp.body.clone()),
-                logs,
-            )
-        }
-        Err(e) => {
-            let status = match e {
-                ExecError::Timeout(_) => ExecutionStatus::Timeout,
-                ExecError::OperationBudgetExceeded => ExecutionStatus::BudgetExceeded,
-                _ => ExecutionStatus::Error,
-            };
-            (
-                status,
-                None,
-                Some(serde_json::json!({ "error": e.to_string() })),
-                Json_::Array(vec![]),
-            )
-        }
-    };
-
-    ExecutionLog {
-        id: Uuid::new_v4(),
-        app_id,
-        script_id,
-        request_id,
-        request_path,
-        request_headers,
-        request_body,
-        response_code,
-        response_body,
-        script_logs,
-        duration_ms,
-        status,
-        created_at: started,
-    }
-}
 
 // ----------------------------------------------------------------------------
 // Errors

crates/orchestrator-core/src/routing/pattern.rs

@@ -108,8 +108,15 @@ pub fn parse_path(kind: PathKind, raw: &str) -> Result<PathPattern, ParseError>
 }
 
 fn check_reserved(raw: &str) -> Result<(), ParseError> {
+    // Case-fold before comparing: request-time path matching is case-sensitive,
+    // but the reserved namespace must be rejected regardless of case so a tenant
+    // can't publish look-alikes like `/Admin/login` or `/API/v2/x`. Method/host
+    // matching are already case-insensitive; this keeps the validation guard
+    // durable even if path matching ever follows.
+    let lowered = raw.to_ascii_lowercase();
     for r in RESERVED_PATH_PREFIXES {
-        if raw == r.trim_end_matches('/') || raw.starts_with(r) {
+        if lowered == r.trim_end_matches('/') || lowered.starts_with(r) {
+            // Preserve the caller's original case in the error for clarity.
             return Err(ParseError::ReservedPath(raw.to_string()));
         }
     }
@@ -456,6 +463,29 @@ mod tests {
         }
     }
 
+    #[test]
+    fn rejects_reserved_paths_case_insensitively() {
+        // Case variants of every reserved prefix must be rejected: the reserved
+        // namespace is a security boundary and must not be bypassable by casing.
+        for raw in [
+            "/API/v2/foo",
+            "/Api/v2/foo",
+            "/aPi/x",
+            "/Admin/dashboard",
+            "/ADMIN/x",
+            "/HEALTHZ",
+            "/HealthZ",
+            "/Version",
+            "/VERSION",
+        ] {
+            let e = parse_path(PathKind::Exact, raw).unwrap_err();
+            assert!(
+                matches!(e, ParseError::ReservedPath(_)),
+                "expected reserved for {raw:?}, got {e:?}"
+            );
+        }
+    }
+
     #[test]
     fn rejects_missing_leading_slash() {
         let e = parse_path(PathKind::Exact, "greet").unwrap_err();

crates/picloud-cli/src/client.rs

@@ -12,7 +12,7 @@ use chrono::{DateTime, Utc};
 use percent_encoding::{utf8_percent_encode, AsciiSet, CONTROLS};
 use picloud_shared::{
     AdminUserId, ApiKeyId, App, AppDomain, AppId, AppRole, AppUser, DispatchMode, ExecutionLog,
-    HostKind, InstanceRole, PathKind, Route, Scope, Script, ScriptId,
+    HostKind, InstanceRole, PathKind, Route, Scope, Script, ScriptId, ScriptKind, ScriptSandbox,
 };
 use reqwest::{header, Method, RequestBuilder, StatusCode};
 use serde::{Deserialize, Serialize};
@@ -171,10 +171,23 @@ impl Client {
     }
 
     /// `PUT /api/v1/admin/scripts/{id}` — matches the dashboard, which
-    /// uses PUT despite the field-level update semantics.
+    /// uses PUT despite the field-level update semantics. `cfg` carries
-    pub async fn scripts_update_source(&self, id: &str, source: &str) -> Result<Script> {
+    /// optional per-script runtime overrides (G3); unset fields are
+    /// omitted so they keep their stored value.
+    pub async fn scripts_update_source(
+        &self,
+        id: &str,
+        source: &str,
+        cfg: &ScriptConfig,
+    ) -> Result<Script> {
         let id = seg(id);
-        let body = UpdateScriptBody { source };
+        let body = UpdateScriptBody {
+            source,
+            timeout_seconds: cfg.timeout_seconds,
+            memory_limit_mb: cfg.memory_limit_mb,
+            kind: cfg.kind,
+            sandbox: cfg.sandbox,
+        };
         let resp = self
             .request(Method::PUT, &format!("/api/v1/admin/scripts/{id}"))
             .json(&body)
@@ -222,15 +235,19 @@ impl Client {
     }
 
     /// `GET /api/v1/admin/scripts/{id}/logs?limit=N`
-    pub async fn logs_list(&self, script_id: &str, limit: u32) -> Result<Vec<ExecutionLog>> {
+    pub async fn logs_list(
+        &self,
+        script_id: &str,
+        limit: u32,
+        source: Option<&str>,
+    ) -> Result<Vec<ExecutionLog>> {
         let script_id = seg(script_id);
-        let resp = self
+        let mut path = format!("/api/v1/admin/scripts/{script_id}/logs?limit={limit}");
-            .request(
+        if let Some(src) = source {
-                Method::GET,
+            path.push_str("&source=");
-                &format!("/api/v1/admin/scripts/{script_id}/logs?limit={limit}"),
+            path.push_str(&seg(src));
-            )
+        }
-            .send()
+        let resp = self.request(Method::GET, &path).send().await?;
-            .await?;
         decode(resp).await
     }
 
@@ -709,6 +726,180 @@ impl Client {
             .await?;
         decode(resp).await
     }
+
+    // --- App membership (G2) ----------------------------------------------
+
+    /// `GET /api/v1/admin/apps/{id_or_slug}/members`
+    pub async fn members_list(&self, app: &str) -> Result<Vec<AppMemberDto>> {
+        let app = seg(app);
+        let resp = self
+            .request(Method::GET, &format!("/api/v1/admin/apps/{app}/members"))
+            .send()
+            .await?;
+        decode(resp).await
+    }
+
+    /// `POST /api/v1/admin/apps/{id_or_slug}/members`
+    pub async fn members_grant(
+        &self,
+        app: &str,
+        user_id: &str,
+        role: AppRole,
+    ) -> Result<AppMemberDto> {
+        let app = seg(app);
+        let body = serde_json::json!({ "user_id": user_id, "role": role });
+        let resp = self
+            .request(Method::POST, &format!("/api/v1/admin/apps/{app}/members"))
+            .json(&body)
+            .send()
+            .await?;
+        decode(resp).await
+    }
+
+    /// `PATCH /api/v1/admin/apps/{id_or_slug}/members/{user_id}`
+    pub async fn members_set_role(
+        &self,
+        app: &str,
+        user_id: &str,
+        role: AppRole,
+    ) -> Result<AppMemberDto> {
+        let (app, user_id) = (seg(app), seg(user_id));
+        let body = serde_json::json!({ "role": role });
+        let resp = self
+            .request(
+                Method::PATCH,
+                &format!("/api/v1/admin/apps/{app}/members/{user_id}"),
+            )
+            .json(&body)
+            .send()
+            .await?;
+        decode(resp).await
+    }
+
+    /// `DELETE /api/v1/admin/apps/{id_or_slug}/members/{user_id}`
+    pub async fn members_remove(&self, app: &str, user_id: &str) -> Result<()> {
+        let (app, user_id) = (seg(app), seg(user_id));
+        let resp = self
+            .request(
+                Method::DELETE,
+                &format!("/api/v1/admin/apps/{app}/members/{user_id}"),
+            )
+            .send()
+            .await?;
+        decode_status(resp).await
+    }
+
+    // --- Files (G2, read-only admin surface) ------------------------------
+
+    /// `GET /api/v1/admin/apps/{id_or_slug}/files?collection=&limit=`
+    pub async fn files_list(
+        &self,
+        app: &str,
+        collection: &str,
+        limit: u32,
+    ) -> Result<ListFilesResponse> {
+        let app = seg(app);
+        let collection = seg(collection);
+        let resp = self
+            .request(
+                Method::GET,
+                &format!("/api/v1/admin/apps/{app}/files?collection={collection}&limit={limit}"),
+            )
+            .send()
+            .await?;
+        decode(resp).await
+    }
+
+    /// `GET /api/v1/admin/apps/{id_or_slug}/files/{collection}/{file_id}` —
+    /// streams the raw bytes (download). Returns the body verbatim.
+    pub async fn files_get_bytes(
+        &self,
+        app: &str,
+        collection: &str,
+        file_id: &str,
+    ) -> Result<Vec<u8>> {
+        let (app, collection, file_id) = (seg(app), seg(collection), seg(file_id));
+        let resp = self
+            .request(
+                Method::GET,
+                &format!("/api/v1/admin/apps/{app}/files/{collection}/{file_id}"),
+            )
+            .send()
+            .await?;
+        if resp.status().is_success() {
+            Ok(resp.bytes().await.context("reading file bytes")?.to_vec())
+        } else {
+            Err(server_error(resp).await)
+        }
+    }
+
+    /// `DELETE /api/v1/admin/apps/{id_or_slug}/files/{collection}/{file_id}`
+    pub async fn files_delete(&self, app: &str, collection: &str, file_id: &str) -> Result<()> {
+        let (app, collection, file_id) = (seg(app), seg(collection), seg(file_id));
+        let resp = self
+            .request(
+                Method::DELETE,
+                &format!("/api/v1/admin/apps/{app}/files/{collection}/{file_id}"),
+            )
+            .send()
+            .await?;
+        decode_status(resp).await
+    }
+
+    // --- KV (G2, read-only admin surface) ---------------------------------
+
+    /// `GET /api/v1/admin/apps/{id_or_slug}/kv?collection=&limit=`
+    pub async fn kv_list(&self, app: &str, collection: &str, limit: u32) -> Result<KvListPageDto> {
+        let app = seg(app);
+        let collection = seg(collection);
+        let resp = self
+            .request(
+                Method::GET,
+                &format!("/api/v1/admin/apps/{app}/kv?collection={collection}&limit={limit}"),
+            )
+            .send()
+            .await?;
+        decode(resp).await
+    }
+
+    /// `GET /api/v1/admin/apps/{id_or_slug}/kv/{collection}/{key}`
+    pub async fn kv_get(&self, app: &str, collection: &str, key: &str) -> Result<Value> {
+        let (app, collection, key) = (seg(app), seg(collection), seg(key));
+        let resp = self
+            .request(
+                Method::GET,
+                &format!("/api/v1/admin/apps/{app}/kv/{collection}/{key}"),
+            )
+            .send()
+            .await?;
+        let wrapped: KvGetResponse = decode(resp).await?;
+        Ok(wrapped.value)
+    }
+
+    // --- Queues (G2, read-only admin surface) -----------------------------
+
+    /// `GET /api/v1/admin/apps/{id_or_slug}/queues`
+    pub async fn queues_list(&self, app: &str) -> Result<Vec<QueueSummaryDto>> {
+        let app = seg(app);
+        let resp = self
+            .request(Method::GET, &format!("/api/v1/admin/apps/{app}/queues"))
+            .send()
+            .await?;
+        decode(resp).await
+    }
+
+    /// `GET /api/v1/admin/apps/{id_or_slug}/queues/{queue_name}`
+    pub async fn queue_get(&self, app: &str, queue_name: &str) -> Result<QueueDetailDto> {
+        let (app, queue_name) = (seg(app), seg(queue_name));
+        let resp = self
+            .request(
+                Method::GET,
+                &format!("/api/v1/admin/apps/{app}/queues/{queue_name}"),
+            )
+            .send()
+            .await?;
+        decode(resp).await
+    }
 }
 
 /// `POST /api/v1/admin/auth/login` — sits outside the `Client` because
@@ -957,6 +1148,17 @@ pub struct SecretItemDto {
     pub updated_at: DateTime<Utc>,
 }
 
+/// Per-script runtime config the CLI can now set (G3). All optional — an
+/// unset field is omitted so the server applies its own default (and the
+/// `PICLOUD_SANDBOX_MAX_*` admin ceilings still clamp overrides).
+#[derive(Debug, Default, Clone)]
+pub struct ScriptConfig {
+    pub timeout_seconds: Option<i32>,
+    pub memory_limit_mb: Option<i32>,
+    pub kind: Option<ScriptKind>,
+    pub sandbox: Option<ScriptSandbox>,
+}
+
 #[derive(Debug, Serialize)]
 pub struct CreateScriptBody<'a> {
     pub app_id: AppId,
@@ -964,11 +1166,104 @@ pub struct CreateScriptBody<'a> {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub description: Option<&'a str>,
     pub source: &'a str,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub timeout_seconds: Option<i32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub memory_limit_mb: Option<i32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub kind: Option<ScriptKind>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub sandbox: Option<ScriptSandbox>,
 }
 
 #[derive(Debug, Serialize)]
 struct UpdateScriptBody<'a> {
     source: &'a str,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    timeout_seconds: Option<i32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    memory_limit_mb: Option<i32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    kind: Option<ScriptKind>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    sandbox: Option<ScriptSandbox>,
+}
+
+// --- G2 response DTOs (deserialize-only mirrors of the server shapes) ---
+
+// `#[allow(dead_code)]` on a couple of fields below: these structs mirror
+// the full server response shape (so the surface is documented and future
+// columns are a one-line add), but the TSV/KvBlock renderers don't print
+// every field today.
+
+#[derive(Debug, Deserialize)]
+pub struct AppMemberDto {
+    pub user_id: AdminUserId,
+    pub username: String,
+    #[allow(dead_code)]
+    pub email: Option<String>,
+    pub instance_role: InstanceRole,
+    pub is_active: bool,
+    pub role: AppRole,
+    pub created_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct FileMetaDto {
+    pub id: String,
+    #[allow(dead_code)]
+    pub collection: String,
+    pub name: String,
+    pub content_type: String,
+    pub size: u64,
+    #[allow(dead_code)]
+    pub checksum: String,
+    #[allow(dead_code)]
+    pub created_at: String,
+    pub updated_at: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct ListFilesResponse {
+    pub files: Vec<FileMetaDto>,
+    pub next_cursor: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct KvListPageDto {
+    pub keys: Vec<String>,
+    pub next_cursor: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+struct KvGetResponse {
+    value: Value,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct QueueSummaryDto {
+    pub queue_name: String,
+    pub total: u64,
+    pub pending: u64,
+    pub claimed: u64,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct QueueConsumerDto {
+    pub trigger_id: String,
+    pub script_id: ScriptId,
+    pub script_name: String,
+    pub visibility_timeout_secs: u32,
+    pub last_fired_at: Option<DateTime<Utc>>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct QueueDetailDto {
+    pub queue_name: String,
+    pub total: u64,
+    pub pending: u64,
+    pub claimed: u64,
+    pub consumer: Option<QueueConsumerDto>,
 }
 
 #[derive(Debug, Serialize)]

crates/picloud-cli/src/cmds/files.rs

@@ -0,0 +1,71 @@
+//! `pic files ls | get | rm` — operator-facing files inspection (G2).
+//!
+//! Wraps the read-only `/api/v1/admin/apps/{id}/files*` surface. There is
+//! no `set`/`upload` — blob writes go through scripts (`files::create`);
+//! the admin surface is inspect + delete only, matching the dashboard.
+
+use std::io::Write;
+use std::path::Path;
+
+use anyhow::{Context, Result};
+
+use crate::client::Client;
+use crate::config;
+use crate::output::{OutputMode, Table};
+
+pub async fn ls(app: &str, collection: &str, limit: u32, mode: OutputMode) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let page = client.files_list(app, collection, limit).await?;
+    let mut table = Table::new(["id", "name", "content_type", "size", "updated_at"]);
+    for f in page.files {
+        table.row([
+            f.id,
+            f.name,
+            f.content_type,
+            f.size.to_string(),
+            f.updated_at,
+        ]);
+    }
+    table.print(mode);
+    if page.next_cursor.is_some() {
+        let _ = writeln!(
+            std::io::stderr(),
+            "(more results available — raise --limit to see them)"
+        );
+    }
+    Ok(())
+}
+
+/// Download a file's bytes. With `--out <path>` writes to disk; otherwise
+/// streams raw bytes to stdout (pipe to a file or `xxd`).
+pub async fn get(app: &str, collection: &str, file_id: &str, out: Option<&Path>) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let bytes = client.files_get_bytes(app, collection, file_id).await?;
+    match out {
+        Some(path) => {
+            std::fs::write(path, &bytes).with_context(|| format!("writing {}", path.display()))?;
+            let _ = writeln!(
+                std::io::stderr(),
+                "Wrote {} bytes to {}",
+                bytes.len(),
+                path.display()
+            );
+        }
+        None => {
+            std::io::stdout()
+                .write_all(&bytes)
+                .context("writing bytes to stdout")?;
+        }
+    }
+    Ok(())
+}
+
+pub async fn rm(app: &str, collection: &str, file_id: &str) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    client.files_delete(app, collection, file_id).await?;
+    println!("Deleted file {file_id} from {collection}");
+    Ok(())
+}

crates/picloud-cli/src/cmds/kv.rs

@@ -0,0 +1,42 @@
+//! `pic kv ls | get` — read-only KV inspection (G2).
+//!
+//! Wraps the read-only `/api/v1/admin/apps/{id}/kv*` surface. There is no
+//! `set`/`rm` on purpose: KV writes go through `kv::set` in scripts (which
+//! emit the change events triggers depend on); an admin write would bypass
+//! that, so the CLI stays read-only (matching the queues precedent).
+
+use std::io::Write;
+
+use anyhow::Result;
+
+use crate::client::Client;
+use crate::config;
+use crate::output::{OutputMode, Table};
+
+pub async fn ls(app: &str, collection: &str, limit: u32, mode: OutputMode) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let page = client.kv_list(app, collection, limit).await?;
+    let mut table = Table::new(["key"]);
+    for k in page.keys {
+        table.row([k]);
+    }
+    table.print(mode);
+    if page.next_cursor.is_some() {
+        let _ = writeln!(
+            std::io::stderr(),
+            "(more keys available — raise --limit to see them)"
+        );
+    }
+    Ok(())
+}
+
+pub async fn get(app: &str, collection: &str, key: &str) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let value = client.kv_get(app, collection, key).await?;
+    // Always emit the JSON value (pretty) so it pipes cleanly into jq.
+    let pretty = serde_json::to_string_pretty(&value).unwrap_or_else(|_| value.to_string());
+    println!("{pretty}");
+    Ok(())
+}

crates/picloud-cli/src/cmds/logs.rs

@@ -12,10 +12,15 @@ use crate::client::Client;
 use crate::config;
 use crate::output::{OutputMode, Table};
 
-pub async fn run(script_id: &str, limit: u32, mode: OutputMode) -> Result<()> {
+pub async fn run(
+    script_id: &str,
+    limit: u32,
+    source: Option<&str>,
+    mode: OutputMode,
+) -> Result<()> {
     let creds = config::resolve()?;
     let client = Client::from_creds(&creds)?;
-    let entries = client.logs_list(script_id, limit).await?;
+    let entries = client.logs_list(script_id, limit, source).await?;
     match mode {
         OutputMode::Tsv => render_tsv(&entries),
         OutputMode::Json => render_json(&entries),
@@ -24,11 +29,14 @@ pub async fn run(script_id: &str, limit: u32, mode: OutputMode) -> Result<()> {
 }
 
 fn render_tsv(entries: &[ExecutionLog]) {
-    let mut table = Table::new(["created_at", "status", "summary"]);
+    // `source` is shown so background runs (queue/cron/invoke/…) are
+    // distinguishable from HTTP at a glance — the whole point of G1.
+    let mut table = Table::new(["created_at", "source", "status", "summary"]);
     for e in entries {
         let summary = summarize(&e.response_body, &e.script_logs);
         table.row([
             e.created_at.to_rfc3339(),
+            e.source.as_str().to_string(),
             status_label(&e.status).to_string(),
             truncate(&summary, 120),
         ]);

crates/picloud-cli/src/cmds/members.rs

@@ -0,0 +1,86 @@
+//! `pic members ls | add | set | rm` — manage app membership (G2).
+//!
+//! Wraps `/api/v1/admin/apps/{id}/members*`. All gated on
+//! `AppAdmin(app)` server-side; Editors/Viewers get a 403.
+
+use anyhow::Result;
+use picloud_shared::AppRole;
+
+use crate::client::Client;
+use crate::config;
+use crate::output::{KvBlock, OutputMode, Table};
+
+pub async fn ls(app: &str, mode: OutputMode) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let members = client.members_list(app).await?;
+    let mut table = Table::new(["user_id", "username", "role", "instance_role", "active"]);
+    for m in members {
+        table.row([
+            m.user_id.to_string(),
+            m.username,
+            app_role_str(m.role).to_string(),
+            format!("{:?}", m.instance_role).to_lowercase(),
+            m.is_active.to_string(),
+        ]);
+    }
+    table.print(mode);
+    Ok(())
+}
+
+pub async fn add(app: &str, user_id: &str, role: &str, mode: OutputMode) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let m = client
+        .members_grant(app, user_id, parse_role(role)?)
+        .await?;
+    print_member(&m, mode);
+    Ok(())
+}
+
+pub async fn set(app: &str, user_id: &str, role: &str, mode: OutputMode) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let m = client
+        .members_set_role(app, user_id, parse_role(role)?)
+        .await?;
+    print_member(&m, mode);
+    Ok(())
+}
+
+pub async fn rm(app: &str, user_id: &str) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    client.members_remove(app, user_id).await?;
+    println!("Removed {user_id} from {app}");
+    Ok(())
+}
+
+fn print_member(m: &crate::client::AppMemberDto, mode: OutputMode) {
+    let mut block = KvBlock::new();
+    block
+        .field("user_id", m.user_id.to_string())
+        .field("username", m.username.clone())
+        .field("role", app_role_str(m.role))
+        .field("created_at", m.created_at.to_rfc3339());
+    block.print(mode);
+}
+
+fn parse_role(s: &str) -> Result<AppRole> {
+    match s.to_ascii_lowercase().as_str() {
+        "app_admin" | "admin" => Ok(AppRole::AppAdmin),
+        "editor" => Ok(AppRole::Editor),
+        "viewer" => Ok(AppRole::Viewer),
+        other => Err(anyhow::anyhow!(
+            "unknown role {other:?} (want app_admin | editor | viewer)"
+        )),
+    }
+}
+
+fn app_role_str(r: AppRole) -> &'static str {
+    match r {
+        AppRole::AppAdmin => "app_admin",
+        AppRole::Editor => "editor",
+        AppRole::Viewer => "viewer",
+    }
+}

crates/picloud-cli/src/cmds/mod.rs

@@ -3,9 +3,13 @@ pub mod api_keys;
 pub mod apps;
 pub mod apps_domains;
 pub mod dead_letters;
+pub mod files;
+pub mod kv;
 pub mod login;
 pub mod logout;
 pub mod logs;
+pub mod members;
+pub mod queues;
 pub mod routes;
 pub mod scripts;
 pub mod secrets;

crates/picloud-cli/src/cmds/queues.rs

@@ -0,0 +1,62 @@
+//! `pic queues ls | show` — read-only queue inspection (G2).
+//!
+//! Wraps the read-only `/api/v1/admin/apps/{id}/queues*` surface (no
+//! purge/requeue — that is v1.2). Gated on `AppLogRead` server-side.
+
+use anyhow::Result;
+
+use crate::client::Client;
+use crate::config;
+use crate::output::{KvBlock, OutputMode, Table};
+
+pub async fn ls(app: &str, mode: OutputMode) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let queues = client.queues_list(app).await?;
+    let mut table = Table::new(["queue", "total", "pending", "claimed"]);
+    for q in queues {
+        table.row([
+            q.queue_name,
+            q.total.to_string(),
+            q.pending.to_string(),
+            q.claimed.to_string(),
+        ]);
+    }
+    table.print(mode);
+    Ok(())
+}
+
+pub async fn show(app: &str, queue_name: &str, mode: OutputMode) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let q = client.queue_get(app, queue_name).await?;
+    let mut block = KvBlock::new();
+    block
+        .field("queue", q.queue_name)
+        .field("total", q.total.to_string())
+        .field("pending", q.pending.to_string())
+        .field("claimed", q.claimed.to_string());
+    match q.consumer {
+        Some(c) => {
+            block
+                .field("consumer_script", c.script_name)
+                .field("consumer_script_id", c.script_id.to_string())
+                .field("consumer_trigger_id", c.trigger_id)
+                .field(
+                    "visibility_timeout_secs",
+                    c.visibility_timeout_secs.to_string(),
+                )
+                .field(
+                    "last_fired_at",
+                    c.last_fired_at
+                        .map(|t| t.to_rfc3339())
+                        .unwrap_or_else(|| "-".to_string()),
+                );
+        }
+        None => {
+            block.field("consumer", "(none registered)");
+        }
+    }
+    block.print(mode);
+    Ok(())
+}

crates/picloud-cli/src/cmds/scripts.rs

@@ -8,7 +8,7 @@ use anyhow::{anyhow, Context, Result};
 use picloud_shared::AppId;
 use serde_json::Value;
 
-use crate::client::{Client, CreateScriptBody};
+use crate::client::{Client, CreateScriptBody, ScriptConfig};
 use crate::config;
 use crate::output::{KvBlock, OutputMode, Table};
 
@@ -62,6 +62,7 @@ pub async fn deploy(
     app_ident: &str,
     name_override: Option<&str>,
     description: Option<&str>,
+    cfg: &ScriptConfig,
     mode: OutputMode,
 ) -> Result<()> {
     let creds = config::resolve()?;
@@ -90,7 +91,7 @@ pub async fn deploy(
     let existing = client.scripts_list_by_app(app_ident).await?;
     let (script, action) = if let Some(s) = existing.into_iter().find(|s| s.name == name) {
         let updated = client
-            .scripts_update_source(&s.id.to_string(), &source)
+            .scripts_update_source(&s.id.to_string(), &source, cfg)
             .await?;
         (updated, "updated")
     } else {
@@ -99,6 +100,10 @@ pub async fn deploy(
             name: &name,
             description,
             source: &source,
+            timeout_seconds: cfg.timeout_seconds,
+            memory_limit_mb: cfg.memory_limit_mb,
+            kind: cfg.kind,
+            sandbox: cfg.sandbox,
         };
         (client.scripts_create(&body).await?, "created")
     };

crates/picloud-cli/src/cmds/triggers.rs

@@ -124,6 +124,114 @@ pub async fn create_dead_letter(
     Ok(())
 }
 
+/// docs/files share KV's `{collection_glob, ops?}` shape.
+async fn create_collection_trigger(
+    kind: &str,
+    app: &str,
+    script_id: &str,
+    collection_glob: &str,
+    ops: &[String],
+    dispatch: &str,
+    mode: OutputMode,
+) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let mut body = base_body(script_id, dispatch);
+    body["collection_glob"] = json!(collection_glob);
+    if !ops.is_empty() {
+        body["ops"] = json!(ops);
+    }
+    let created = client.triggers_create(app, kind, &body).await?;
+    print_created(&created, mode);
+    Ok(())
+}
+
+pub async fn create_docs(
+    app: &str,
+    script_id: &str,
+    collection_glob: &str,
+    ops: &[String],
+    dispatch: &str,
+    mode: OutputMode,
+) -> Result<()> {
+    create_collection_trigger("docs", app, script_id, collection_glob, ops, dispatch, mode).await
+}
+
+pub async fn create_files(
+    app: &str,
+    script_id: &str,
+    collection_glob: &str,
+    ops: &[String],
+    dispatch: &str,
+    mode: OutputMode,
+) -> Result<()> {
+    create_collection_trigger(
+        "files",
+        app,
+        script_id,
+        collection_glob,
+        ops,
+        dispatch,
+        mode,
+    )
+    .await
+}
+
+pub async fn create_pubsub(
+    app: &str,
+    script_id: &str,
+    topic_pattern: &str,
+    dispatch: &str,
+    mode: OutputMode,
+) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let mut body = base_body(script_id, dispatch);
+    body["topic_pattern"] = json!(topic_pattern);
+    let created = client.triggers_create(app, "pubsub", &body).await?;
+    print_created(&created, mode);
+    Ok(())
+}
+
+pub async fn create_queue(
+    app: &str,
+    script_id: &str,
+    queue_name: &str,
+    visibility_timeout_secs: Option<u32>,
+    dispatch: &str,
+    mode: OutputMode,
+) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    let mut body = base_body(script_id, dispatch);
+    body["queue_name"] = json!(queue_name);
+    if let Some(v) = visibility_timeout_secs {
+        body["visibility_timeout_secs"] = json!(v);
+    }
+    let created = client.triggers_create(app, "queue", &body).await?;
+    print_created(&created, mode);
+    Ok(())
+}
+
+pub async fn create_email(
+    app: &str,
+    script_id: &str,
+    inbound_secret: Option<&str>,
+    mode: OutputMode,
+) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    // Email triggers take no dispatch_mode (inbound webhook only) and an
+    // optional shared HMAC secret the provider signs POSTs with.
+    let mut body = json!({ "script_id": script_id });
+    if let Some(secret) = inbound_secret {
+        body["inbound_secret"] = json!(secret);
+    }
+    let created = client.triggers_create(app, "email", &body).await?;
+    print_created(&created, mode);
+    Ok(())
+}
+
 pub async fn create_from_json(app: &str, kind: &str, body: &str, mode: OutputMode) -> Result<()> {
     let creds = config::resolve()?;
     let client = Client::from_creds(&creds)?;

crates/picloud-cli/src/main.rs

@@ -128,6 +128,138 @@ enum Cmd {
         #[command(subcommand)]
         cmd: SecretsCmd,
     },
+
+    /// App membership — list members and grant / change / revoke their
+    /// per-app role (app_admin | editor | viewer).
+    Members {
+        #[command(subcommand)]
+        cmd: MembersCmd,
+    },
+
+    /// Files inspection — list a collection's blobs, download bytes, or
+    /// delete a file. Read + delete only; writes go through scripts.
+    Files {
+        #[command(subcommand)]
+        cmd: FilesCmd,
+    },
+
+    /// Queue inspection — list queues with depth counts, or drill into
+    /// one queue's stats + registered consumer. Read-only.
+    Queues {
+        #[command(subcommand)]
+        cmd: QueuesCmd,
+    },
+
+    /// KV inspection — list keys in a collection or fetch one value.
+    /// Read-only; writes go through `kv::set` in scripts.
+    Kv {
+        #[command(subcommand)]
+        cmd: KvCmd,
+    },
+}
+
+#[derive(Subcommand)]
+enum KvCmd {
+    /// List keys in a collection.
+    Ls {
+        #[arg(long)]
+        app: String,
+        #[arg(long)]
+        collection: String,
+        #[arg(long, default_value_t = 100)]
+        limit: u32,
+    },
+    /// Fetch one key's value (printed as JSON).
+    Get {
+        #[arg(long)]
+        app: String,
+        #[arg(long)]
+        collection: String,
+        key: String,
+    },
+}
+
+#[derive(Subcommand)]
+enum MembersCmd {
+    /// List app members.
+    Ls {
+        #[arg(long)]
+        app: String,
+    },
+    /// Grant a user a role on the app.
+    Add {
+        #[arg(long)]
+        app: String,
+        #[arg(long = "user")]
+        user_id: String,
+        /// `app_admin` | `editor` | `viewer`.
+        #[arg(long)]
+        role: String,
+    },
+    /// Change an existing member's role.
+    Set {
+        #[arg(long)]
+        app: String,
+        #[arg(long = "user")]
+        user_id: String,
+        #[arg(long)]
+        role: String,
+    },
+    /// Remove a member.
+    Rm {
+        #[arg(long)]
+        app: String,
+        #[arg(long = "user")]
+        user_id: String,
+    },
+}
+
+#[derive(Subcommand)]
+enum FilesCmd {
+    /// List files in a collection.
+    Ls {
+        #[arg(long)]
+        app: String,
+        #[arg(long)]
+        collection: String,
+        #[arg(long, default_value_t = 100)]
+        limit: u32,
+    },
+    /// Download a file's bytes (to `--out <path>` or stdout).
+    Get {
+        #[arg(long)]
+        app: String,
+        #[arg(long)]
+        collection: String,
+        #[arg(long = "id")]
+        file_id: String,
+        #[arg(long)]
+        out: Option<PathBuf>,
+    },
+    /// Delete a file.
+    Rm {
+        #[arg(long)]
+        app: String,
+        #[arg(long)]
+        collection: String,
+        #[arg(long = "id")]
+        file_id: String,
+    },
+}
+
+#[derive(Subcommand)]
+enum QueuesCmd {
+    /// List queues with depth counts.
+    Ls {
+        #[arg(long)]
+        app: String,
+    },
+    /// Show one queue's stats + consumer.
+    Show {
+        #[arg(long)]
+        app: String,
+        queue_name: String,
+    },
 }
 
 #[derive(Args)]
@@ -234,6 +366,82 @@ struct DeployArgs {
     name: Option<String>,
     #[arg(long)]
     description: Option<String>,
+    /// Per-script wall-clock timeout in seconds (overrides the instance
+    /// default; still clamped by `PICLOUD_SANDBOX_MAX_*` ceilings).
+    #[arg(long)]
+    timeout: Option<i32>,
+    /// Per-script memory ceiling in MB.
+    #[arg(long)]
+    memory: Option<i32>,
+    /// Script kind: `endpoint` (default) or `module` (importable, no route).
+    #[arg(long, value_enum)]
+    kind: Option<ScriptKindArg>,
+    /// Sandbox override as `key=value`, repeatable. Keys: `max_operations`,
+    /// `max_string_size`, `max_array_size`, `max_map_size`,
+    /// `max_call_levels`, `max_expr_depth`. E.g. `--sandbox max_operations=500000`.
+    #[arg(long = "sandbox", value_name = "KEY=VALUE")]
+    sandbox: Vec<String>,
+}
+
+#[derive(Copy, Clone, clap::ValueEnum)]
+enum ScriptKindArg {
+    Endpoint,
+    Module,
+}
+
+impl From<ScriptKindArg> for picloud_shared::ScriptKind {
+    fn from(v: ScriptKindArg) -> Self {
+        match v {
+            ScriptKindArg::Endpoint => Self::Endpoint,
+            ScriptKindArg::Module => Self::Module,
+        }
+    }
+}
+
+impl DeployArgs {
+    /// Fold the runtime-config flags into a `ScriptConfig`, parsing and
+    /// validating the `--sandbox key=value` pairs.
+    fn script_config(&self) -> anyhow::Result<client::ScriptConfig> {
+        let sandbox = parse_sandbox_overrides(&self.sandbox)?;
+        Ok(client::ScriptConfig {
+            timeout_seconds: self.timeout,
+            memory_limit_mb: self.memory,
+            kind: self.kind.map(Into::into),
+            sandbox,
+        })
+    }
+}
+
+/// Parse repeatable `--sandbox key=value` flags into a `ScriptSandbox`.
+/// Unknown keys and non-integer values are hard errors so a typo doesn't
+/// silently deploy an unrestricted script.
+fn parse_sandbox_overrides(
+    pairs: &[String],
+) -> anyhow::Result<Option<picloud_shared::ScriptSandbox>> {
+    use anyhow::{anyhow, Context};
+    if pairs.is_empty() {
+        return Ok(None);
+    }
+    let mut sb = picloud_shared::ScriptSandbox::default();
+    for raw in pairs {
+        let (key, val) = raw
+            .split_once('=')
+            .ok_or_else(|| anyhow!("--sandbox expects key=value, got {raw:?}"))?;
+        let n: u64 = val
+            .trim()
+            .parse()
+            .with_context(|| format!("--sandbox {key}: {val:?} is not a non-negative integer"))?;
+        match key.trim() {
+            "max_operations" => sb.max_operations = Some(n),
+            "max_string_size" => sb.max_string_size = Some(n),
+            "max_array_size" => sb.max_array_size = Some(n),
+            "max_map_size" => sb.max_map_size = Some(n),
+            "max_call_levels" => sb.max_call_levels = Some(n),
+            "max_expr_depth" => sb.max_expr_depth = Some(n),
+            other => return Err(anyhow!("unknown --sandbox key: {other:?}")),
+        }
+    }
+    Ok(Some(sb))
 }
 
 #[derive(Args)]
@@ -275,6 +483,11 @@ struct LogsArgs {
     script_id: String,
     #[arg(long, default_value_t = 50)]
     limit: u32,
+    /// Filter by execution origin: `http`, `kv`, `cron`, `queue`,
+    /// `invoke`, `dead_letter`, `docs`, `files`, `pubsub`, `email`.
+    /// Omit (or `all`) to show every source.
+    #[arg(long)]
+    source: Option<String>,
 }
 
 #[derive(Clone, Copy, ValueEnum)]
@@ -498,10 +711,89 @@ enum TriggersCmd {
         dispatch: DispatchModeArg,
     },
 
+    /// Create a docs trigger — fires on document mutations in matching
+    /// collections.
+    #[command(name = "create-docs")]
+    CreateDocs {
+        #[arg(long)]
+        app: String,
+        #[arg(long)]
+        script: String,
+        /// Glob over collection names (`*`, `posts`, `events_*`, …).
+        #[arg(long)]
+        collection: String,
+        /// Repeat to filter ops: `--op insert --op delete`. Empty fires on any.
+        #[arg(long = "op")]
+        ops: Vec<String>,
+        #[arg(long, value_enum, default_value_t = DispatchModeArg::Async)]
+        dispatch: DispatchModeArg,
+    },
+
+    /// Create a files trigger — fires on blob create/update/delete in
+    /// matching collections.
+    #[command(name = "create-files")]
+    CreateFiles {
+        #[arg(long)]
+        app: String,
+        #[arg(long)]
+        script: String,
+        #[arg(long)]
+        collection: String,
+        #[arg(long = "op")]
+        ops: Vec<String>,
+        #[arg(long, value_enum, default_value_t = DispatchModeArg::Async)]
+        dispatch: DispatchModeArg,
+    },
+
+    /// Create a pub/sub trigger — fires on messages published to topics
+    /// matching the pattern.
+    #[command(name = "create-pubsub")]
+    CreatePubsub {
+        #[arg(long)]
+        app: String,
+        #[arg(long)]
+        script: String,
+        /// Topic glob (`*`, `orders.*`, `user.signup`, …).
+        #[arg(long = "topic")]
+        topic_pattern: String,
+        #[arg(long, value_enum, default_value_t = DispatchModeArg::Async)]
+        dispatch: DispatchModeArg,
+    },
+
+    /// Create a queue consumer trigger — fires per message claimed off
+    /// the named queue.
+    #[command(name = "create-queue")]
+    CreateQueue {
+        #[arg(long)]
+        app: String,
+        #[arg(long)]
+        script: String,
+        #[arg(long = "queue")]
+        queue_name: String,
+        /// Per-message visibility timeout in seconds (claim lease).
+        #[arg(long = "visibility-timeout")]
+        visibility_timeout_secs: Option<u32>,
+        #[arg(long, value_enum, default_value_t = DispatchModeArg::Async)]
+        dispatch: DispatchModeArg,
+    },
+
+    /// Create an email trigger — fires on inbound mail POSTed to the
+    /// webhook receiver. No dispatch mode (inbound webhook only).
+    #[command(name = "create-email")]
+    CreateEmail {
+        #[arg(long)]
+        app: String,
+        #[arg(long)]
+        script: String,
+        /// Shared HMAC secret the provider signs inbound POSTs with.
+        /// Omit to accept unsigned POSTs.
+        #[arg(long = "inbound-secret")]
+        inbound_secret: Option<String>,
+    },
+
     /// Create a trigger of any kind from a raw JSON body — escape
-    /// hatch for kinds the CLI doesn't expose per-kind wrappers for
+    /// hatch for advanced retry/dispatch settings beyond the per-kind
-    /// (docs/files/pubsub/email/queue) and for advanced retry/dispatch
+    /// wrappers' defaults.
-    /// settings beyond the per-kind defaults.
     #[command(name = "create-from-json")]
     CreateFromJson {
         #[arg(long)]
@@ -790,16 +1082,20 @@ async fn main() -> ExitCode {
         } => cmds::scripts::ls(app.as_deref(), mode).await,
         Cmd::Scripts {
             cmd: ScriptsCmd::Deploy(args),
-        } => {
+        } => match args.script_config() {
-            cmds::scripts::deploy(
+            Ok(cfg) => {
-                &args.file,
+                cmds::scripts::deploy(
-                &args.app,
+                    &args.file,
-                args.name.as_deref(),
+                    &args.app,
-                args.description.as_deref(),
+                    args.name.as_deref(),
-                mode,
+                    args.description.as_deref(),
-            )
+                    &cfg,
-            .await
+                    mode,
-        }
+                )
+                .await
+            }
+            Err(e) => Err(e),
+        },
         Cmd::Scripts {
             cmd: ScriptsCmd::Invoke(args),
         } => cmds::scripts::invoke(&args.id, args.body.as_deref(), &args.headers).await,
@@ -821,20 +1117,28 @@ async fn main() -> ExitCode {
         Cmd::ApiKeys {
             cmd: ApiKeysCmd::Rm { id },
         } => cmds::api_keys::rm(&id).await,
-        Cmd::Logs(LogsArgs { script_id, limit }) => cmds::logs::run(&script_id, limit, mode).await,
+        Cmd::Logs(LogsArgs {
+            script_id,
+            limit,
+            source,
+        }) => cmds::logs::run(&script_id, limit, source.as_deref(), mode).await,
         Cmd::Invoke(args) => {
             cmds::scripts::invoke(&args.id, args.body.as_deref(), &args.headers).await
         }
-        Cmd::Deploy(args) => {
+        Cmd::Deploy(args) => match args.script_config() {
-            cmds::scripts::deploy(
+            Ok(cfg) => {
-                &args.file,
+                cmds::scripts::deploy(
-                &args.app,
+                    &args.file,
-                args.name.as_deref(),
+                    &args.app,
-                args.description.as_deref(),
+                    args.name.as_deref(),
-                mode,
+                    args.description.as_deref(),
-            )
+                    &cfg,
-            .await
+                    mode,
-        }
+                )
+                .await
+            }
+            Err(e) => Err(e),
+        },
         Cmd::Routes {
             cmd: RoutesCmd::Ls { script_id },
         } => cmds::routes::ls(&script_id, mode).await,
@@ -1004,6 +1308,92 @@ async fn main() -> ExitCode {
             )
             .await
         }
+        Cmd::Triggers {
+            cmd:
+                TriggersCmd::CreateDocs {
+                    app,
+                    script,
+                    collection,
+                    ops,
+                    dispatch,
+                },
+        } => {
+            cmds::triggers::create_docs(
+                &app,
+                &script,
+                &collection,
+                &ops,
+                dispatch_wire(dispatch),
+                mode,
+            )
+            .await
+        }
+        Cmd::Triggers {
+            cmd:
+                TriggersCmd::CreateFiles {
+                    app,
+                    script,
+                    collection,
+                    ops,
+                    dispatch,
+                },
+        } => {
+            cmds::triggers::create_files(
+                &app,
+                &script,
+                &collection,
+                &ops,
+                dispatch_wire(dispatch),
+                mode,
+            )
+            .await
+        }
+        Cmd::Triggers {
+            cmd:
+                TriggersCmd::CreatePubsub {
+                    app,
+                    script,
+                    topic_pattern,
+                    dispatch,
+                },
+        } => {
+            cmds::triggers::create_pubsub(
+                &app,
+                &script,
+                &topic_pattern,
+                dispatch_wire(dispatch),
+                mode,
+            )
+            .await
+        }
+        Cmd::Triggers {
+            cmd:
+                TriggersCmd::CreateQueue {
+                    app,
+                    script,
+                    queue_name,
+                    visibility_timeout_secs,
+                    dispatch,
+                },
+        } => {
+            cmds::triggers::create_queue(
+                &app,
+                &script,
+                &queue_name,
+                visibility_timeout_secs,
+                dispatch_wire(dispatch),
+                mode,
+            )
+            .await
+        }
+        Cmd::Triggers {
+            cmd:
+                TriggersCmd::CreateEmail {
+                    app,
+                    script,
+                    inbound_secret,
+                },
+        } => cmds::triggers::create_email(&app, &script, inbound_secret.as_deref(), mode).await,
         Cmd::Triggers {
             cmd: TriggersCmd::CreateFromJson { app, kind, body },
         } => cmds::triggers::create_from_json(&app, &kind, &body, mode).await,
@@ -1074,6 +1464,65 @@ async fn main() -> ExitCode {
         Cmd::Secrets {
             cmd: SecretsCmd::Rm { app, name },
         } => cmds::secrets::rm(&app, &name).await,
+        Cmd::Members {
+            cmd: MembersCmd::Ls { app },
+        } => cmds::members::ls(&app, mode).await,
+        Cmd::Members {
+            cmd: MembersCmd::Add { app, user_id, role },
+        } => cmds::members::add(&app, &user_id, &role, mode).await,
+        Cmd::Members {
+            cmd: MembersCmd::Set { app, user_id, role },
+        } => cmds::members::set(&app, &user_id, &role, mode).await,
+        Cmd::Members {
+            cmd: MembersCmd::Rm { app, user_id },
+        } => cmds::members::rm(&app, &user_id).await,
+        Cmd::Files {
+            cmd:
+                FilesCmd::Ls {
+                    app,
+                    collection,
+                    limit,
+                },
+        } => cmds::files::ls(&app, &collection, limit, mode).await,
+        Cmd::Files {
+            cmd:
+                FilesCmd::Get {
+                    app,
+                    collection,
+                    file_id,
+                    out,
+                },
+        } => cmds::files::get(&app, &collection, &file_id, out.as_deref()).await,
+        Cmd::Files {
+            cmd:
+                FilesCmd::Rm {
+                    app,
+                    collection,
+                    file_id,
+                },
+        } => cmds::files::rm(&app, &collection, &file_id).await,
+        Cmd::Queues {
+            cmd: QueuesCmd::Ls { app },
+        } => cmds::queues::ls(&app, mode).await,
+        Cmd::Queues {
+            cmd: QueuesCmd::Show { app, queue_name },
+        } => cmds::queues::show(&app, &queue_name, mode).await,
+        Cmd::Kv {
+            cmd:
+                KvCmd::Ls {
+                    app,
+                    collection,
+                    limit,
+                },
+        } => cmds::kv::ls(&app, &collection, limit, mode).await,
+        Cmd::Kv {
+            cmd:
+                KvCmd::Get {
+                    app,
+                    collection,
+                    key,
+                },
+        } => cmds::kv::get(&app, &collection, &key).await,
     };
 
     match result {

crates/picloud/src/lib.rs

@@ -12,26 +12,26 @@ use picloud_executor_core::{Engine, Limits};
 use picloud_manager_core::{
     admin_router, admins_router, api_keys_router, app_members_router, apps_api, apps_router,
     attach_principal_if_present, auth_router, compile_routes, dead_letters_router,
-    email_inbound_router, files_admin_router, migrations, require_authenticated,
+    dev_emails_router, email_inbound_router, files_admin_router, kv_admin_router, migrations,
-    route_admin_router, secrets_router, topics_router, triggers_router, AbandonedRepo,
+    require_authenticated, route_admin_router, secrets_router, topics_router, triggers_router,
-    AdminPrincipalResolver, AdminSessionRepository, AdminState, AdminUserRepository, AdminsState,
+    AbandonedRepo, AdminPrincipalResolver, AdminSessionRepository, AdminState, AdminUserRepository,
-    ApiKeyRepository, ApiKeysState, AppDomainRepository, AppMembersRepository, AppMembersState,
+    AdminsState, ApiKeyRepository, ApiKeysState, AppDomainRepository, AppMembersRepository,
-    AppRepository, AppsState, AuthState, AuthzRepo, DeadLetterRepo, DeadLettersState, Dispatcher,
+    AppMembersState, AppRepository, AppsState, AuthState, AuthzRepo, DeadLetterRepo,
-    DocsServiceImpl, EmailInboundState, EmailServiceImpl, FilesAdminState, FilesConfig,
+    DeadLettersState, DevEmailState, Dispatcher, DocsServiceImpl, EmailInboundState,
-    FilesServiceImpl, FsFilesRepo, HttpConfig, HttpServiceImpl, InboundNonceDedup, KvServiceImpl,
+    EmailServiceImpl, FilesAdminState, FilesConfig, FilesServiceImpl, FsFilesRepo, HttpConfig,
-    OutboxEventEmitter, OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository,
+    HttpServiceImpl, InboundNonceDedup, KvAdminState, KvServiceImpl, OutboxEventEmitter,
-    PostgresAdminUserRepository, PostgresApiKeyRepository, PostgresAppDomainRepository,
+    OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository, PostgresAdminUserRepository,
-    PostgresAppMembersRepository, PostgresAppRepository, PostgresAppSecretsRepo,
+    PostgresApiKeyRepository, PostgresAppDomainRepository, PostgresAppMembersRepository,
-    PostgresAppUserInvitationRepo, PostgresAppUserPasswordResetRepo, PostgresAppUserRepository,
+    PostgresAppRepository, PostgresAppSecretsRepo, PostgresAppUserInvitationRepo,
-    PostgresAppUserRoleRepo, PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo,
+    PostgresAppUserPasswordResetRepo, PostgresAppUserRepository, PostgresAppUserRoleRepo,
-    PostgresDeadLetterRepo, PostgresDeadLetterService, PostgresDocsRepo,
+    PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo, PostgresDeadLetterRepo,
-    PostgresExecutionLogRepository, PostgresExecutionLogSink, PostgresKvRepo, PostgresOutboxRepo,
+    PostgresDeadLetterService, PostgresDocsRepo, PostgresExecutionLogRepository,
-    PostgresPubsubRepo, PostgresRouteRepository, PostgresScriptRepository, PostgresSecretsRepo,
+    PostgresExecutionLogSink, PostgresKvRepo, PostgresOutboxRepo, PostgresPubsubRepo,
-    PostgresTopicRepo, PostgresTriggerRepo, PrincipalResolver, PubsubServiceImpl,
+    PostgresRouteRepository, PostgresScriptRepository, PostgresSecretsRepo, PostgresTopicRepo,
-    RealtimeAuthorityImpl, RepoResolver, RouteAdminState, RouteRepository, SandboxCeiling,
+    PostgresTriggerRepo, PrincipalResolver, PubsubServiceImpl, RealtimeAuthorityImpl, RepoResolver,
-    ScriptRepository, SecretsConfig, SecretsServiceImpl, SecretsState, SubscriberTokenConfig,
+    RouteAdminState, RouteRepository, SandboxCeiling, ScriptRepository, SecretsConfig,
-    TopicRepo, TopicsState, TriggerConfig, TriggerRepo, TriggersState, UsersServiceConfig,
+    SecretsServiceImpl, SecretsState, SubscriberTokenConfig, TopicRepo, TopicsState, TriggerConfig,
-    UsersServiceImpl,
+    TriggerRepo, TriggersState, UsersServiceConfig, UsersServiceImpl,
 };
 use picloud_orchestrator_core::realtime::DEFAULT_GC_INTERVAL_SECS;
 use picloud_orchestrator_core::routing::{AppDomainTable, RouteTable};
@@ -146,7 +146,7 @@ pub async fn build_app(
         outbox_repo.clone(),
     ));
     let kv: Arc<dyn KvService> = Arc::new(KvServiceImpl::with_max_value_bytes(
-        kv_repo,
+        kv_repo.clone(),
         authz.clone(),
         events.clone(),
         picloud_manager_core::kv_service::kv_max_value_bytes_from_env(),
@@ -239,8 +239,12 @@ pub async fn build_app(
         secrets_config,
     ));
     // v1.1.7 outbound email. Builds a lettre SMTP transport from
-    // PICLOUD_SMTP_* env (disabled mode + warning if unconfigured).
+    // PICLOUD_SMTP_* env (disabled mode + warning if unconfigured). G5: in
-    let email: Arc<dyn EmailService> = Arc::new(EmailServiceImpl::from_env(authz.clone()));
+    // dev mode with no relay, captures mail in memory instead of erroring;
+    // `dev_email_sink` is `Some` then, and we mount the dev inspection
+    // endpoint below.
+    let (email_impl, dev_email_sink) = EmailServiceImpl::from_env_with_dev_capture(authz.clone());
+    let email: Arc<dyn EmailService> = Arc::new(email_impl);
     // v1.1.8 data-plane user management. Wires Argon2id-hashed user
     // rows + SHA-256-hashed sliding-window sessions to the Rhai
     // `users::*` namespace and the admin /apps/{id}/users HTTP surface.
@@ -290,8 +294,10 @@ pub async fn build_app(
     // routes.
     let route_table = Arc::new(RouteTable::new());
     let initial = route_repo.list_all().await?;
-    let compiled = compile_routes(&initial)
+    // Lenient: a single un-compilable stored route (e.g. one whose path
-        .map_err(|e| anyhow::anyhow!("failed to compile stored routes: {e}"))?;
+    // became reserved under stricter validation) is skipped-with-warning
+    // inside compile_routes, never aborting boot. (H1)
+    let compiled = compile_routes(&initial);
     route_table.replace_all(compiled);
 
     // v1.1.9 function composition. InvokeServiceImpl resolves targets
@@ -380,6 +386,7 @@ pub async fn build_app(
         principals,
         executor: executor.clone(),
         gate,
+        log_sink: log_sink.clone(),
         inbox: inbox_resolver,
         queue: queue_repo.clone(),
         config: trigger_config,
@@ -544,7 +551,7 @@ pub async fn build_app(
     // else under /admin gets the require_authenticated layer; capability
     // checks live in each handler (after the resource is loaded so the
     // capability binds to the resource's actual app_id).
-    let guarded_admin = Router::new()
+    let mut guarded_admin = Router::new()
         .merge(admin_router(admin))
         .merge(route_admin_router(route_admin))
         .merge(admins_router(admins_state))
@@ -565,13 +572,24 @@ pub async fn build_app(
             },
         ))
         .merge(files_admin_router(files_admin_state))
+        .merge(kv_admin_router(KvAdminState {
+            kv: kv_repo.clone(),
+            apps: apps_repo.clone(),
+            authz: authz.clone(),
+        }))
         .merge(topics_router(topics_state))
         .merge(secrets_router(secrets_state))
-        .merge(dead_letters_router(dead_letters_state))
+        .merge(dead_letters_router(dead_letters_state));
-        .layer(from_fn_with_state(
+    // G5: dev-only mail inspection — mounted exactly when the email
-            auth_state.clone(),
+    // service is capturing in memory (dev mode + no relay). Same
-            require_authenticated,
+    // `require_authenticated` layer as the rest of /admin applies below.
-        ));
+    if let Some(sink) = dev_email_sink {
+        guarded_admin = guarded_admin.merge(dev_emails_router(DevEmailState { sink }));
+    }
+    let guarded_admin = guarded_admin.layer(from_fn_with_state(
+        auth_state.clone(),
+        require_authenticated,
+    ));
 
     // Silence "unused import" lint on `apps_api` — we re-export via the
     // facade above; the bare module path is retained so it's discoverable.

crates/shared/src/execution_log.rs

@@ -30,9 +30,77 @@ pub struct ExecutionLog {
 
     pub duration_ms: u64,
     pub status: ExecutionStatus,
+
+    /// What dispatched this execution: `http` for direct data-plane
+    /// ingress, or one of the trigger kinds (`kv`, `cron`, `queue`,
+    /// `invoke`, …) for background runs. Materialized so `pic logs`
+    /// can surface — and filter by — the origin of every run, not just
+    /// the HTTP ones. Defaults to `http` for rows written before the
+    /// column existed (migration 0043).
+    #[serde(default)]
+    pub source: ExecutionSource,
+
     pub created_at: DateTime<Utc>,
 }
 
+/// Origin of an execution. Wire strings mirror
+/// `manager-core::OutboxSourceKind` (plus `Queue`, which the queue
+/// consumer dispatches outside the outbox) so a trigger's source kind
+/// maps straight through to its execution-log row. Keep the variants and
+/// the `source` CHECK constraint in migration 0043 in sync.
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
+#[serde(rename_all = "snake_case")]
+pub enum ExecutionSource {
+    #[default]
+    Http,
+    Kv,
+    Docs,
+    DeadLetter,
+    Cron,
+    Files,
+    Pubsub,
+    Email,
+    Invoke,
+    Queue,
+}
+
+impl ExecutionSource {
+    #[must_use]
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::Http => "http",
+            Self::Kv => "kv",
+            Self::Docs => "docs",
+            Self::DeadLetter => "dead_letter",
+            Self::Cron => "cron",
+            Self::Files => "files",
+            Self::Pubsub => "pubsub",
+            Self::Email => "email",
+            Self::Invoke => "invoke",
+            Self::Queue => "queue",
+        }
+    }
+
+    /// Parse a wire string back into a variant. Returns `None` for an
+    /// unknown value (callers treat that as "no filter" or a 422).
+    #[must_use]
+    pub fn from_wire(s: &str) -> Option<Self> {
+        match s {
+            "http" => Some(Self::Http),
+            "kv" => Some(Self::Kv),
+            "docs" => Some(Self::Docs),
+            "dead_letter" => Some(Self::DeadLetter),
+            "cron" => Some(Self::Cron),
+            "files" => Some(Self::Files),
+            "pubsub" => Some(Self::Pubsub),
+            "email" => Some(Self::Email),
+            "invoke" => Some(Self::Invoke),
+            "queue" => Some(Self::Queue),
+            _ => None,
+        }
+    }
+}
+
 /// Matches the CHECK constraint on `execution_logs.status`. Keep the
 /// serde rename in sync with the migration.
 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]

crates/shared/src/lib.rs

@@ -48,7 +48,7 @@ pub use email::{EmailError, EmailService, NoopEmailService, OutboundEmail};
 pub use error::Error;
 pub use events::{EmitError, NoopEventEmitter, ServiceEvent, ServiceEventEmitter};
 pub use exec_summary::ExecResponseSummary;
-pub use execution_log::{ExecutionLog, ExecutionStatus};
+pub use execution_log::{ExecutionLog, ExecutionSource, ExecutionStatus};
 pub use files::{
     sanitize_stored_content_type, validate_collection as validate_files_collection, FileMeta,
     FileUpdate, FilesError, FilesListPage, FilesService, NewFile, NoopFilesService,

docs/sdk-shape.md

@@ -122,6 +122,18 @@ unauthenticated by default — public HTTP scripts run with `None`.
 Services that need an authenticated identity (e.g., `users::*`) check
 `cx.principal.is_some()` and throw if missing.
 
+> **A public route ≠ public *data*.** When a route is unauthenticated,
+> the script runs with `principal: None`, and the capability gate
+> (`authz::script_gate`) returns `Ok` immediately — the script holds
+> **full app authority**. It can read and write *all* of the app's
+> `kv`, `docs`, `files`, `secrets`, `queue`, and `pubsub`, because the
+> platform only gates *authenticated* principals; for anonymous ingress
+> **the script itself is the only access boundary**. If a public route
+> must not expose every secret or every row in the app, the script must
+> enforce that — check a token, scope the collection, gate the
+> operation. "This route is public" does not mean "this code may only
+> touch public data." See blueprint §11.6 (principal model).
+
 ## Sync ↔ async bridge
 
 Rhai is synchronous; service trait methods (KV writes, HTTP calls) are

docs/stdlib-reference.md

@@ -37,16 +37,33 @@ These come with the Rhai engine itself. See the
 `index_of`, `split`, `trim`, `to_lower`, `to_upper`, `replace`, `chars`,
 `pad`, `sub_string`, `crop`, `+` (concatenation).
 
-> **Footgun — `replace` mutates in place and returns `()`.** Rhai's
+> **Footgun — several string methods mutate in place and return `()`.**
-> `String.replace(from, to)` edits the receiver and returns unit, *not*
+> A whole family of Rhai string methods edit the receiver *in place* and
-> a new string. `let t = auth.replace("Bearer ", "")` sets `t` to `()`,
+> return unit, **not** a new string. `let t = auth.replace("Bearer ", "")`
-> so a downstream `users::verify(t)` fails with
+> (or `let t = text.trim()`) sets `t` to `()`, so a downstream
-> `Function not found: users::verify (())`. To strip a known prefix,
+> `text.split(",")` or `users::verify(t)` fails with
-> slice instead:
+> `Function not found: split (())` / `… (())`.
+>
+> | Method | Returns | Use it as |
+> |---|---|---|
+> | `replace`, `trim`, `make_upper`, `make_lower`, `crop`, `truncate`, `pad` | `()` — **mutates in place** | a statement: `text.trim();` then read `text` |
+> | `to_upper`, `to_lower`, `sub_string`, `split`, `chars` | a **new value** (string / array / range) | an expression: `let u = name.to_upper();` |
+>
+> So `to_upper`/`to_lower` are the *copying* variants (safe in `let x = …`),
+> while `make_upper`/`make_lower` are the in-place ones. To trim and keep
+> the result, mutate then read the same binding:
+> ```rhai
+> let token = auth;
+> token.trim();                       // mutate in place
+> if token.starts_with("Bearer ") {   // now use `token`
+>     token.replace("Bearer ", "");   // again: statement, not `let x = …`
+> }
+> ```
+> Or, to strip a known prefix without mutation, slice:
 > ```rhai
 > let token = if auth.starts_with("Bearer ") { auth.sub_string(7) } else { auth };
 > ```
-> Or call `replace` purely for its side effect: `let t = auth; t.replace("Bearer ", "");`.
+> (Verified against the pinned Rhai `=1.24` build.)
 
 **Array:** `push`, `pop`, `shift`, `insert`, `remove`, `len`, `clear`,
 `truncate`, `extend`, `filter`, `map`, `reduce`, `reduce_rev`, `find`,