feat: end-to-end script CRUD + Rhai execution

Brings the MVP feature set online: upload a Rhai script, get an HTTP
endpoint that runs it sandboxed in-process, list/update/delete it, and
have invalid sources rejected at upload time. Verified live through
Caddy with a full lifecycle (`create → list → get → execute → update
→ delete`) plus error paths (syntax error, duplicate name, deleted).

Layout — every concern lands behind the trait seam its layer owns, so
cluster-mode in v1.3+ is a swap of two impls, not a rewrite:

  * shared::ScriptValidator — manager calls into validation without
    a hard dep on executor-core; executor-core impls the trait on
    `Engine`. Pinned in shared so neither crate has to know about
    the other.

  * executor-core::Engine — real Rhai engine: sandbox limits (max
    operations / string size / map size / call depth), disabled
    `print`, blocked `import` (DummyModuleResolver), `log::trace
    /info/warn/error` registered as a static module with shared
    log-capture buffer (no `log::debug` because `debug` is a Rhai
    reserved keyword — `log::trace` covers the same need).

      - `ctx` is pushed as a Scope constant exposing
        execution_id, script_id, script_name, request_id,
        invocation_type, request.{path,headers,body}.

      - Response convention: a Map with `statusCode` is the
        structured shape (`{statusCode, headers?, body}`); any
        other return value is a 200 with the value as the body.

      - Engine::execute is now synchronous (pure compute); the
        async wrapper + wall-clock timeout live in
        LocalExecutorClient, which spawns_blocking and applies a
        300s hard ceiling regardless of per-script config.

      - 10 unit tests cover validate, exec, structured response,
        ctx exposure, log capture, op-budget enforcement, runtime
        errors, blocked imports, JSON round-tripping.

  * manager-core::repo — full sqlx CRUD over the `scripts` table,
    with proper unique-violation handling for duplicate names.
    Embedded migrations via `sqlx::migrate!` (one initial
    `0001_init.sql` for pgcrypto + scripts + execution_logs).

  * manager-core::api — `admin_router` mounts `/scripts` and
    `/scripts/{id}`. Create + Update validate source through the
    injected `ScriptValidator` before persistence. Returns proper
    422/409/404 status codes via `ApiError::IntoResponse`.

  * orchestrator-core::api — `data_plane_router` mounts
    `/execute/{id}`: resolves the script through `ScriptResolver`,
    constructs the `ExecRequest` from headers+body, awaits
    `ExecutorClient::execute(..., timeout)`, translates the
    `ExecResponse` to an axum `Response` with header passthrough.
    Maps `ExecError` variants to 422/504/502/507.

  * picloud all-in-one — opens the pool, runs migrations, builds
    one engine, nests both routers under `/api/admin` and `/api`,
    enables structured JSON tracing and graceful shutdown on
    SIGTERM. Single `PostgresScriptRepository` Arc is shared by
    the admin router (writes) and the resolver (reads).

Other changes:
  * Workspace axum bump 0.7 → 0.8 for the `{id}` path syntax
    matching the route definitions.
  * Workspace clippy: allow `needless_pass_by_value` and
    `boxed_local` to keep API ergonomics over pedantic noise.

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

crates/orchestrator-core/Cargo.toml

@@ -13,6 +13,7 @@ picloud-shared.workspace = true
 picloud-executor-core.workspace = true
 
 async-trait.workspace = true
+axum.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true
@@ -20,3 +21,4 @@ tracing.workspace = true
 uuid.workspace = true
 chrono.workspace = true
 reqwest.workspace = true
+tokio.workspace = true

crates/orchestrator-core/src/api.rs

@@ -0,0 +1,180 @@
+//! Data-plane HTTP surface. Mounted by the `picloud` all-in-one binary
+//! under `/api` (so the path becomes `/api/execute/:id`) and by the
+//! future split `picloud-orchestrator` binary at its own root.
+
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use axum::{
+    body::Bytes,
+    extract::{Path, State},
+    http::{HeaderMap, HeaderName, HeaderValue, StatusCode},
+    response::{IntoResponse, Response},
+    routing::post,
+    Json, Router,
+};
+use picloud_executor_core::{ExecError, ExecRequest, InvocationType};
+use picloud_shared::{ExecutionId, RequestId, ScriptId};
+use serde_json::Value as Json_;
+
+use crate::client::ExecutorClient;
+use crate::resolver::{ResolverError, ScriptResolver};
+
+/// State shared by data-plane handlers.
+///
+/// Both fields are `Arc` because handlers run concurrently; the
+/// underlying impls are `Send + Sync` (enforced by their traits).
+pub struct DataPlaneState<E, R> {
+    pub executor: Arc<E>,
+    pub resolver: Arc<R>,
+}
+
+impl<E, R> Clone for DataPlaneState<E, R> {
+    fn clone(&self) -> Self {
+        Self {
+            executor: self.executor.clone(),
+            resolver: self.resolver.clone(),
+        }
+    }
+}
+
+/// Build the data-plane router. Handles `POST /execute/:id`.
+pub fn data_plane_router<E, R>(state: DataPlaneState<E, R>) -> Router
+where
+    E: ExecutorClient + 'static,
+    R: ScriptResolver + 'static,
+{
+    Router::new()
+        .route("/execute/{id}", post(execute_by_id::<E, R>))
+        .with_state(state)
+}
+
+// ----------------------------------------------------------------------------
+// Handlers
+// ----------------------------------------------------------------------------
+
+async fn execute_by_id<E, R>(
+    State(state): State<DataPlaneState<E, R>>,
+    Path(id): Path<ScriptId>,
+    headers: HeaderMap,
+    body: Bytes,
+) -> Result<Response, ApiError>
+where
+    E: ExecutorClient + 'static,
+    R: ScriptResolver + 'static,
+{
+    let script = state
+        .resolver
+        .resolve(id)
+        .await?
+        .ok_or(ApiError::NotFound(id))?;
+
+    let req = build_exec_request(id, &script.name, &headers, &body)?;
+
+    let timeout = Duration::from_secs(u64::from(script.timeout_seconds));
+    let resp = state.executor.execute(&script.source, req, timeout).await?;
+
+    Ok(exec_response_to_http(resp))
+}
+
+// ----------------------------------------------------------------------------
+// Marshalling
+// ----------------------------------------------------------------------------
+
+fn build_exec_request(
+    id: ScriptId,
+    name: &str,
+    headers: &HeaderMap,
+    body: &Bytes,
+) -> Result<ExecRequest, ApiError> {
+    let mut hmap = BTreeMap::new();
+    for (k, v) in headers {
+        if let Ok(s) = v.to_str() {
+            hmap.insert(k.as_str().to_string(), s.to_string());
+        }
+    }
+
+    let body_json: Json_ = if body.is_empty() {
+        Json_::Null
+    } else {
+        serde_json::from_slice(body)
+            .map_err(|e| ApiError::BadRequest(format!("invalid JSON body: {e}")))?
+    };
+
+    Ok(ExecRequest {
+        execution_id: ExecutionId::new(),
+        request_id: RequestId::new(),
+        script_id: id,
+        script_name: name.to_string(),
+        invocation_type: InvocationType::Http,
+        path: format!("/api/execute/{id}"),
+        headers: hmap,
+        body: body_json,
+    })
+}
+
+fn exec_response_to_http(resp: picloud_executor_core::ExecResponse) -> Response {
+    let status =
+        StatusCode::from_u16(resp.status_code).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
+
+    let mut http_headers = HeaderMap::new();
+    for (k, v) in resp.headers {
+        if let (Ok(name), Ok(value)) = (k.parse::<HeaderName>(), v.parse::<HeaderValue>()) {
+            http_headers.insert(name, value);
+        }
+    }
+    // Default content type to JSON; the script can override via `headers`.
+    http_headers
+        .entry(axum::http::header::CONTENT_TYPE)
+        .or_insert_with(|| HeaderValue::from_static("application/json"));
+
+    (status, http_headers, Json(resp.body)).into_response()
+}
+
+// ----------------------------------------------------------------------------
+// Errors
+// ----------------------------------------------------------------------------
+
+#[derive(Debug, thiserror::Error)]
+pub enum ApiError {
+    #[error("script not found: {0}")]
+    NotFound(ScriptId),
+
+    #[error("bad request: {0}")]
+    BadRequest(String),
+
+    #[error("resolver error: {0}")]
+    Resolver(#[from] ResolverError),
+
+    #[error("execution error: {0}")]
+    Exec(#[from] ExecError),
+}
+
+impl IntoResponse for ApiError {
+    fn into_response(self) -> Response {
+        use ApiError as E;
+        let (status, message) = match &self {
+            E::NotFound(_) => (StatusCode::NOT_FOUND, self.to_string()),
+            E::BadRequest(_) => (StatusCode::BAD_REQUEST, self.to_string()),
+            E::Resolver(e) => {
+                tracing::error!(error = %e, "resolver failure");
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    "internal error".to_string(),
+                )
+            }
+            E::Exec(e) => match e {
+                ExecError::Parse(_) | ExecError::InvalidResponse(_) => {
+                    (StatusCode::UNPROCESSABLE_ENTITY, e.to_string())
+                }
+                ExecError::Timeout(_) => (StatusCode::GATEWAY_TIMEOUT, e.to_string()),
+                ExecError::OperationBudgetExceeded => {
+                    (StatusCode::INSUFFICIENT_STORAGE, e.to_string())
+                }
+                ExecError::Runtime(_) => (StatusCode::BAD_GATEWAY, e.to_string()),
+            },
+        };
+        (status, Json(serde_json::json!({ "error": message }))).into_response()
+    }
+}

crates/orchestrator-core/src/client.rs

@@ -1,20 +1,35 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use async_trait::async_trait;
 use picloud_executor_core::{Engine, ExecError, ExecRequest, ExecResponse};
 
+/// Maximum wall-clock time we'll wait for a single invocation, regardless
+/// of the per-script `timeout_seconds`. Provides a hard ceiling on
+/// resource usage independent of misconfigured scripts.
+const HARD_TIMEOUT_CAP: Duration = Duration::from_secs(300);
+
 /// The seam between the orchestrator and the executor.
 ///
 /// Single-node mode plugs in `LocalExecutorClient`, which calls
-/// `executor-core` in-process. Cluster mode plugs in `RemoteExecutorClient`,
-/// which forwards over HTTP to an executor node. Everything else in
-/// orchestrator-core depends only on this trait.
+/// `executor-core` in-process via `spawn_blocking`. Cluster mode plugs
+/// in `RemoteExecutorClient`, which forwards over HTTP to an executor
+/// node. Everything else in orchestrator-core depends only on this trait.
 #[async_trait]
 pub trait ExecutorClient: Send + Sync {
-    async fn execute(&self, source: &str, req: ExecRequest) -> Result<ExecResponse, ExecError>;
+    async fn execute(
+        &self,
+        source: &str,
+        req: ExecRequest,
+        timeout: Duration,
+    ) -> Result<ExecResponse, ExecError>;
 }
 
 /// In-process executor — wraps `executor-core::Engine` directly.
+///
+/// `executor-core::Engine::execute` is synchronous; we offload it to a
+/// blocking thread so it doesn't park a Tokio worker, and apply the
+/// wall-clock timeout here.
 pub struct LocalExecutorClient {
     engine: Arc<Engine>,
 }
@@ -28,8 +43,26 @@ impl LocalExecutorClient {
 
 #[async_trait]
 impl ExecutorClient for LocalExecutorClient {
-    async fn execute(&self, source: &str, req: ExecRequest) -> Result<ExecResponse, ExecError> {
-        self.engine.execute(source, req).await
+    async fn execute(
+        &self,
+        source: &str,
+        req: ExecRequest,
+        timeout: Duration,
+    ) -> Result<ExecResponse, ExecError> {
+        let timeout = timeout.min(HARD_TIMEOUT_CAP);
+        let timeout_secs = u32::try_from(timeout.as_secs()).unwrap_or(u32::MAX);
+
+        let engine = self.engine.clone();
+        let source = source.to_string();
+        let join = tokio::task::spawn_blocking(move || engine.execute(&source, req));
+
+        match tokio::time::timeout(timeout, join).await {
+            Err(_) => Err(ExecError::Timeout(timeout_secs)),
+            Ok(Err(join_err)) => Err(ExecError::Runtime(format!(
+                "execution task panicked: {join_err}"
+            ))),
+            Ok(Ok(res)) => res,
+        }
     }
 }
 
@@ -53,7 +86,12 @@ impl RemoteExecutorClient {
 
 #[async_trait]
 impl ExecutorClient for RemoteExecutorClient {
-    async fn execute(&self, _source: &str, _req: ExecRequest) -> Result<ExecResponse, ExecError> {
+    async fn execute(
+        &self,
+        _source: &str,
+        _req: ExecRequest,
+        _timeout: Duration,
+    ) -> Result<ExecResponse, ExecError> {
         Err(ExecError::Runtime(
             "RemoteExecutorClient not implemented (cluster mode is v1.3+)".into(),
         ))

crates/orchestrator-core/src/lib.rs

@@ -8,8 +8,10 @@
 //! trait is the seam that lets the orchestrator call executor logic
 //! in-process (single-node) or over HTTP (cluster).
 
+pub mod api;
 pub mod client;
 pub mod resolver;
 
+pub use api::{data_plane_router, DataPlaneState};
 pub use client::{ExecutorClient, LocalExecutorClient, RemoteExecutorClient};
 pub use resolver::{ResolverError, ScriptResolver};