feat(nodelab): author pinned host-admitted workflows

.github/workflows/quality-gates.yml

@@ -434,6 +434,12 @@ jobs:
           python3 -c 'import pathlib, sys; names = [line for line in pathlib.Path(sys.argv[1]).read_text().splitlines() if "tool_loop" in line and line.endswith(": test")]; print("Tool-loop tests discovered:", len(names)); sys.exit(0 if len(names) == 15 else 1)' "$RUNNER_TEMP/tool-loop-tests.list"
           cargo test --locked -p pantograph-workflow-service --lib tool_loop
 
+      - name: Run exact-head custom-node authoring and admission regression suite
+        run: |
+          cargo test --locked -p pantograph-workflow-service --lib workflow::tests::custom_nodes:: -- --list > "$RUNNER_TEMP/custom-node-tests.list"
+          python3 -c 'import pathlib, sys; names = [line for line in pathlib.Path(sys.argv[1]).read_text().splitlines() if line.startswith("workflow::tests::custom_nodes::") and line.endswith(": test")]; print("Custom-node tests discovered:", len(names)); sys.exit(0 if len(names) == 10 else 1)' "$RUNNER_TEMP/custom-node-tests.list"
+          cargo test --locked -p pantograph-workflow-service --lib workflow::tests::custom_nodes::
+
       - name: Run complete session execution regression suite
         run: |
           cargo test -p pantograph-workflow-service --lib workflow::tests::session_execution:: -- --list > "$RUNNER_TEMP/session-execution-tests.list"

crates/pantograph-workflow-service/examples/nodelab_host.rs

@@ -0,0 +1,187 @@
+//! Isolated stdio host for NodeLab authoring qualification. No runtime code loader.
+//! A verifier supplies the loopback HTTP transport; EOF drops the temporary store.
+use async_trait::async_trait;
+use node_engine::{Context, ExecutorExtensions, NodeExecutor};
+use pantograph_node_contracts::{
+    NodeAuthoringMetadata, NodeCategory, NodeExecutionSemantics, NodeTypeContract, PortContract,
+    PortRequirement, PortValueType,
+};
+use pantograph_workflow_service::{
+    custom_nodes::{authoring::CustomNodeDraftRequest, CustomNodeCatalog, TrustedCustomNode},
+    *,
+};
+use serde::{de::DeserializeOwned, Serialize};
+use serde_json::{json, Value};
+use std::{
+    collections::HashMap,
+    io::{BufRead, Write},
+    sync::Arc,
+};
+
+struct Uppercase;
+#[async_trait]
+impl NodeExecutor for Uppercase {
+    async fn execute(
+        &self,
+        _: &str,
+        inputs: HashMap<String, Value>,
+        _: &Context,
+        _: &ExecutorExtensions,
+    ) -> node_engine::Result<HashMap<String, Value>> {
+        Ok(HashMap::from([(
+            "output".into(),
+            json!(inputs["input"]
+                .as_str()
+                .ok_or_else(|| node_engine::NodeEngineError::ExecutionFailed(
+                    "string input required".into()
+                ))?
+                .to_uppercase()),
+        )]))
+    }
+}
+fn catalog() -> Result<Arc<CustomNodeCatalog>, Box<dyn std::error::Error>> {
+    let mut output = PortContract::output("output".parse()?, "Output", PortValueType::String);
+    output.requirement = PortRequirement::Required;
+    let contract = NodeTypeContract {
+        node_type: "host-uppercase".parse()?,
+        category: NodeCategory::Processing,
+        label: "Host uppercase".into(),
+        description: "Pure string transform compiled into this isolated Rust host".into(),
+        inputs: vec![PortContract::input(
+            "input".parse()?,
+            "Input",
+            PortValueType::String,
+            PortRequirement::Required,
+        )],
+        outputs: vec![output],
+        execution_semantics: NodeExecutionSemantics::Batch,
+        capability_requirements: vec![],
+        inference_tasks: vec![],
+        authoring: NodeAuthoringMetadata::default(),
+        contract_version: Some("1.0.0".into()),
+        contract_digest: None,
+    };
+    let digest = |text: &str| format!("blake3:{}", blake3::hash(text.as_bytes()));
+    Ok(Arc::new(CustomNodeCatalog::try_new(vec![
+        TrustedCustomNode::admit(
+            contract,
+            digest("nodelab_host.rs uppercase v1"),
+            digest("isolated Rust authoring host v1"),
+            Arc::new(Uppercase),
+        )?,
+    ])?))
+}
+fn decode<T: DeserializeOwned>(value: Value) -> Result<T, WorkflowServiceError> {
+    serde_json::from_value(value).map_err(|e| WorkflowServiceError::InvalidRequest(e.to_string()))
+}
+fn encode(value: impl Serialize) -> Result<Value, WorkflowServiceError> {
+    serde_json::to_value(value).map_err(|e| WorkflowServiceError::Internal(e.to_string()))
+}
+fn string(args: &Value, name: &str) -> Result<String, WorkflowServiceError> {
+    args[name]
+        .as_str()
+        .map(str::to_owned)
+        .ok_or_else(|| WorkflowServiceError::InvalidRequest(format!("{name} is required")))
+}
+async fn dispatch(
+    service: &WorkflowService,
+    store: &FileSystemWorkflowGraphStore,
+    command: &str,
+    args: Value,
+) -> Result<Value, WorkflowServiceError> {
+    let edit = || -> Result<WorkflowGraphEditSessionGraphRequest, WorkflowServiceError> {
+        Ok(WorkflowGraphEditSessionGraphRequest {
+            session_id: string(&args, "executionId")?,
+        })
+    };
+    match command {
+        "get_custom_node_authoring_catalog" => encode(service.custom_node_authoring_catalog()),
+        "create_custom_node_draft" => encode(
+            service
+                .create_custom_node_draft(decode::<CustomNodeDraftRequest>(
+                    args["request"].clone(),
+                )?)
+                .await?,
+        ),
+        "get_node_lab_draft" => encode(
+            service
+                .workflow_graph_get_edit_session_graph(edit()?)
+                .await?,
+        ),
+        "create_workflow_execution_session" => encode(
+            service
+                .workflow_graph_create_edit_session(WorkflowGraphEditSessionCreateRequest {
+                    graph: decode(args["graph"].clone())?,
+                    workflow_id: args["workflowId"].as_str().map(str::to_owned),
+                })
+                .await?,
+        ),
+        "update_node_data" => encode(
+            service
+                .workflow_graph_update_node_data(WorkflowGraphUpdateNodeDataRequest {
+                    session_id: string(&args, "executionId")?,
+                    node_id: string(&args, "nodeId")?,
+                    data: args["data"].clone(),
+                })
+                .await?,
+        ),
+        "save_workflow" => encode(
+            service
+                .workflow_graph_save(
+                    store,
+                    WorkflowGraphSaveRequest {
+                        name: string(&args, "name")?,
+                        graph: decode(args["graph"].clone())?,
+                    },
+                )?
+                .path,
+        ),
+        "load_workflow" => encode(service.workflow_graph_load(
+            store,
+            WorkflowGraphLoadRequest {
+                path: string(&args, "path")?,
+            },
+        )?),
+        "remove_execution" => {
+            service
+                .workflow_graph_close_edit_session(WorkflowGraphEditSessionCloseRequest {
+                    session_id: edit()?.session_id,
+                })
+                .await?;
+            Ok(Value::Null)
+        }
+        _ => Err(WorkflowServiceError::CapabilityViolation(
+            "unsupported authoring host command".into(),
+        )),
+    }
+}
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let directory = tempfile::tempdir()?;
+    let store = FileSystemWorkflowGraphStore::new(directory.path());
+    let service = if std::env::args().any(|arg| arg == "--empty-catalog") {
+        WorkflowService::new()
+    } else {
+        WorkflowService::with_trusted_custom_nodes(catalog()?)
+    };
+    for line in std::io::stdin().lock().lines() {
+        let result = match serde_json::from_str::<Value>(&line?) {
+            Ok(request) => match request["command"].as_str() {
+                Some(command) => dispatch(&service, &store, command, request["args"].clone()).await,
+                None => Err(WorkflowServiceError::InvalidRequest(
+                    "command is required".into(),
+                )),
+            },
+            Err(error) => Err(WorkflowServiceError::InvalidRequest(error.to_string())),
+        };
+        let response = match result {
+            Ok(value) => json!({"ok": true, "value": value}),
+            Err(error) => {
+                json!({"ok": false, "error": serde_json::from_str::<Value>(&error.to_envelope_json())?})
+            }
+        };
+        println!("{}", serde_json::to_string(&response)?);
+        std::io::stdout().flush()?;
+    }
+    Ok(())
+}

crates/pantograph-workflow-service/src/custom_nodes.rs

@@ -409,3 +409,5 @@ pub(crate) fn validate_payload(
 
 #[cfg(test)]
 pub(crate) mod tests;
+
+pub mod authoring;

crates/pantograph-workflow-service/src/custom_nodes/authoring.rs

@@ -0,0 +1,37 @@
+//! Read-only host catalog and exactly pinned graph authoring. No execution/loading API.
+use super::CustomNodeIdentity;
+use serde::{Deserialize, Serialize};
+
+/// Version 1 is a deliberately narrow projection of admitted string-port contracts.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct CustomNodeAuthoringCatalog {
+    pub contract_version: u16,
+    pub nodes: Vec<CustomNodeAuthoringEntry>,
+}
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct CustomNodeAuthoringEntry {
+    pub identity: CustomNodeIdentity,
+    pub label: String,
+    pub description: String,
+    pub trust: CustomNodeAuthoringTrust,
+    pub inputs: Vec<CustomNodeAuthoringPort>,
+    pub outputs: Vec<CustomNodeAuthoringPort>,
+}
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum CustomNodeAuthoringTrust {
+    TrustedInProcess,
+}
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct CustomNodeAuthoringPort {
+    pub id: String,
+    pub label: String,
+}
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct CustomNodeDraftRequest {
+    pub identity: CustomNodeIdentity,
+}

crates/pantograph-workflow-service/src/graph/connection_insert.rs

@@ -41,6 +41,13 @@ fn resolve_insert_definition<'a>(
     registry: &'a NodeRegistry,
     node_type: &str,
 ) -> Result<&'a NodeDefinition, ConnectionRejection> {
+    if registry.custom_nodes.get(node_type).is_some() {
+        return Err(ConnectionRejection {
+            reason: ConnectionRejectionReason::UnknownInsertNodeType,
+            message: "custom-node insertion by name is unsupported; add an exactly pinned node and connect its endpoints".into(),
+            contract_diagnostic: None,
+        });
+    }
     registry
         .get_definition(node_type)
         .ok_or_else(|| ConnectionRejection {

crates/pantograph-workflow-service/src/graph/connection_intent.rs

@@ -506,6 +506,7 @@ pub fn connection_candidates_with_surfaces(
     let mut insertable_node_types = registry
         .all_definitions()
         .into_iter()
+        .filter(|definition| registry.custom_nodes.get(&definition.node_type).is_none())
         .filter_map(|definition| {
             let mut matching_input_port_ids = definition
                 .inputs

crates/pantograph-workflow-service/src/graph/session.rs

@@ -75,6 +75,7 @@ fn dirty_tasks_from_seed_nodes_unique(graph: &WorkflowGraph, node_ids: &[String]
 }
 
 pub struct GraphSessionStore {
+    node_registry: Arc<super::registry::NodeRegistry>,
     sessions: RwLock<HashMap<String, GraphSessionHandle>>,
     validation_state: Arc<CurrentInferenceValidationStateStore>,
     validation_lifecycle: Arc<WorkflowGraphValidationLifecycleOwner>,
@@ -136,6 +137,7 @@ impl GraphSessionStore {
         dependency_provider: SharedDependencyEnvironmentProvider,
     ) -> Self {
         Self {
+            node_registry: Arc::new(super::registry::NodeRegistry::new()),
             sessions: RwLock::new(HashMap::new()),
             validation_state: Arc::new(CurrentInferenceValidationStateStore::new()),
             validation_lifecycle: Arc::new(WorkflowGraphValidationLifecycleOwner::new()),
@@ -146,6 +148,14 @@ impl GraphSessionStore {
         }
     }
 
+    pub(crate) fn with_node_registry(
+        mut self,
+        registry: Arc<super::registry::NodeRegistry>,
+    ) -> Self {
+        self.node_registry = registry;
+        self
+    }
+
     pub async fn set_validation_lifecycle_event_sink(
         &self,
         event_sink: Option<Arc<dyn WorkflowGraphValidationLifecycleEventSink>>,
@@ -159,7 +169,11 @@ impl GraphSessionStore {
         workflow_id: Option<String>,
     ) -> WorkflowGraphEditSessionCreateResponse {
         let session_id = Uuid::new_v4().to_string();
-        let session = Arc::new(Mutex::new(GraphEditSession::new(graph, workflow_id)));
+        let session = Arc::new(Mutex::new(GraphEditSession::new(
+            graph,
+            workflow_id,
+            self.node_registry.clone(),
+        )));
         let graph_revision = {
             let state = session.lock().await;
             state.graph.compute_fingerprint()

crates/pantograph-workflow-service/src/graph/session_connection_api.rs

@@ -13,7 +13,6 @@ use super::super::inference_validation_state::{
     WorkflowGraphCurrentConnectionSurfacesStateRequest,
 };
 use super::super::memory_impact::graph_memory_impact_from_graph_change;
-use super::super::registry::NodeRegistry;
 use super::super::session_event::{dirty_tasks_from_seed_nodes, graph_modified_event};
 use super::super::session_graph::sync_embedding_emit_metadata_flags;
 use super::super::session_types::{
@@ -39,10 +38,10 @@ impl GraphSessionStore {
                 request.graph_revision.as_deref(),
             )
             .await?;
-        let registry = NodeRegistry::new();
+        let registry = self.node_registry.as_ref();
         connection_candidates_with_surfaces(
             &graph,
-            &registry,
+            registry,
             InferenceConnectionSurfaceView::new(&connection_surfaces.surfaces),
             request.source_anchor,
             request.graph_revision.as_deref(),
@@ -65,10 +64,10 @@ impl GraphSessionStore {
             let mut state = handle.lock().await;
             state.touch();
             let before_graph = state.graph.clone();
-            let registry = NodeRegistry::new();
+            let registry = self.node_registry.as_ref();
             if let Err(rejection) = commit_connection_with_surfaces(
                 &state.graph,
-                &registry,
+                registry,
                 InferenceConnectionSurfaceView::new(&connection_surfaces.surfaces),
                 &request.graph_revision,
                 &request.source_anchor,
@@ -191,10 +190,10 @@ impl GraphSessionStore {
             let mut state = handle.lock().await;
             state.touch();
             let before_graph = state.graph.clone();
-            let registry = NodeRegistry::new();
+            let registry = self.node_registry.as_ref();
             let (inserted_node, inserted_edge) = match insert_node_and_connect(
                 &state.graph,
-                &registry,
+                registry,
                 &request.graph_revision,
                 &request.source_anchor,
                 &request.node_type,
@@ -252,11 +251,11 @@ impl GraphSessionStore {
         let handle = self.get_session_handle(&request.session_id).await?;
         let mut state = handle.lock().await;
         state.touch();
-        let registry = NodeRegistry::new();
+        let registry = self.node_registry.as_ref();
 
         match preview_node_insert_on_edge(
             &state.graph,
-            &registry,
+            registry,
             &request.graph_revision,
             &request.edge_id,
             &request.node_type,
@@ -283,11 +282,11 @@ impl GraphSessionStore {
             let mut state = handle.lock().await;
             state.touch();
             let before_graph = state.graph.clone();
-            let registry = NodeRegistry::new();
+            let registry = self.node_registry.as_ref();
 
             let (inserted_node, incoming_edge, outgoing_edge, bridge) = match insert_node_on_edge(
                 &state.graph,
-                &registry,
+                registry,
                 &request.graph_revision,
                 &request.edge_id,
                 &request.node_type,