feat: pipeline parser extension — HttpSink/Source/vil_workflow YAML export

manifest_export.rs extended:
- Parse HttpSinkBuilder::new() → node type=http_sink with port, path
- Parse HttpSourceBuilder::new() → node type=http_source with url, format, json_tap, dialect
- Parse vil_workflow! { routes: [...] } → route entries with from/to/mode
- Resolve const values (WEBHOOK_PORT, SSE_URL, etc.)
- Auto-detect pipeline vs server mode
- extract_constants() + resolve_constants() for const substitution

Golden manifests regenerated for all 9 examples:
- Pipeline: 001 (27 lines), 005 (24), 007 (24), 016 (27)
- Server: 003 (19), 004 (28), 010 (19), 605 (37)
- Minimal: 027 (5 — VilServer, no ServiceProcess)

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

Author: fullstack-aidev

crates/vil_cli_server/src/orm/manifest_export.rs

@@ -10,7 +10,37 @@ use std::path::Path;
 pub struct ParsedApp {
     pub name: String,
     pub port: u16,
+    pub mode: AppMode,
     pub services: Vec<ParsedService>,
+    pub nodes: Vec<ParsedNode>,
+    pub routes: Vec<ParsedRoute>,
+}
+
+#[derive(Debug, PartialEq)]
+pub enum AppMode {
+    Server,
+    Pipeline,
+}
+
+/// Parsed pipeline node (HttpSink/HttpSource).
+#[derive(Debug)]
+pub struct ParsedNode {
+    pub name: String,
+    pub node_type: String, // http_sink, http_source, transform
+    pub port: Option<u16>,
+    pub path: Option<String>,
+    pub url: Option<String>,
+    pub format: Option<String>,
+    pub json_tap: Option<String>,
+    pub dialect: Option<String>,
+}
+
+/// Parsed pipeline route.
+#[derive(Debug)]
+pub struct ParsedRoute {
+    pub from: String,
+    pub to: String,
+    pub mode: String,
 }
 
 /// Parsed ServiceProcess.
@@ -33,33 +63,84 @@ pub fn parse_rust_source(path: &Path) -> Result<ParsedApp, String> {
     let source = std::fs::read_to_string(path)
         .map_err(|e| format!("Failed to read {}: {}", path.display(), e))?;
 
-    let name = extract_app_name(&source).unwrap_or_else(|| "app".to_string());
-    let port = extract_port(&source).unwrap_or(8080);
-    let services = extract_services(&source);
+    // Resolve constants: const NAME: type = "value";
+    let constants = extract_constants(&source);
+    let resolved = resolve_constants(&source, &constants);
+
+    let services = extract_services(&resolved);
+    let nodes = extract_pipeline_nodes(&resolved);
+    let routes = extract_pipeline_routes(&resolved);
 
-    Ok(ParsedApp { name, port, services })
+    let is_pipeline = !nodes.is_empty() || source.contains("vil_workflow!");
+    let mode = if is_pipeline { AppMode::Pipeline } else { AppMode::Server };
+
+    let name = if is_pipeline {
+        extract_workflow_name(&resolved)
+            .or_else(|| extract_app_name(&resolved))
+            .unwrap_or_else(|| "app".to_string())
+    } else {
+        extract_app_name(&resolved).unwrap_or_else(|| "app".to_string())
+    };
+
+    let port = extract_port(&resolved)
+        .or_else(|| nodes.iter().find(|n| n.port.is_some()).and_then(|n| n.port))
+        .unwrap_or(8080);
+
+    Ok(ParsedApp { name, port, mode, services, nodes, routes })
 }
 
 /// Generate YAML manifest from parsed app.
 pub fn to_manifest_yaml(app: &ParsedApp) -> String {
     let mut lines = Vec::new();
-    lines.push(format!("vil_version: \"6.0.0\""));
+    lines.push("vil_version: \"6.0.0\"".to_string());
     lines.push(format!("name: {}", app.name));
     lines.push(format!("port: {}", app.port));
-    lines.push("mode: server".to_string());
-
-    if !app.services.is_empty() {
-        lines.push(String::new());
-        lines.push("services:".to_string());
-        for svc in &app.services {
-            lines.push(format!("  - name: {}", svc.name));
-            lines.push(format!("    prefix: /api/{}", svc.name));
-            if !svc.endpoints.is_empty() {
-                lines.push("    endpoints:".to_string());
-                for ep in &svc.endpoints {
-                    lines.push(format!("      - method: {}", ep.method));
-                    lines.push(format!("        path: {}", ep.path));
-                    lines.push(format!("        handler: {}", ep.handler));
+    lines.push(format!("token: shm"));
+
+    match app.mode {
+        AppMode::Pipeline => {
+            // Nodes
+            if !app.nodes.is_empty() {
+                lines.push(String::new());
+                lines.push("nodes:".to_string());
+                for node in &app.nodes {
+                    lines.push(format!("  {}:", node.name));
+                    lines.push(format!("    type: {}", node.node_type));
+                    if let Some(p) = node.port { lines.push(format!("    port: {}", p)); }
+                    if let Some(ref p) = node.path { lines.push(format!("    path: \"{}\"", p)); }
+                    if let Some(ref u) = node.url { lines.push(format!("    url: \"{}\"", u)); }
+                    if let Some(ref f) = node.format { lines.push(format!("    format: {}", f)); }
+                    if let Some(ref j) = node.json_tap { lines.push(format!("    json_tap: \"{}\"", j)); }
+                    if let Some(ref d) = node.dialect { lines.push(format!("    dialect: {}", d)); }
+                }
+            }
+            // Routes
+            if !app.routes.is_empty() {
+                lines.push(String::new());
+                lines.push("routes:".to_string());
+                for r in &app.routes {
+                    lines.push(format!("  - from: {}", r.from));
+                    lines.push(format!("    to: {}", r.to));
+                    lines.push(format!("    mode: {}", r.mode));
+                }
+            }
+        }
+        AppMode::Server => {
+            lines.push("mode: server".to_string());
+            if !app.services.is_empty() {
+                lines.push(String::new());
+                lines.push("services:".to_string());
+                for svc in &app.services {
+                    lines.push(format!("  - name: {}", svc.name));
+                    lines.push(format!("    prefix: /api/{}", svc.name));
+                    if !svc.endpoints.is_empty() {
+                        lines.push("    endpoints:".to_string());
+                        for ep in &svc.endpoints {
+                            lines.push(format!("      - method: {}", ep.method));
+                            lines.push(format!("        path: {}", ep.path));
+                            lines.push(format!("        handler: {}", ep.handler));
+                        }
+                    }
                 }
             }
         }
@@ -203,6 +284,181 @@ fn extract_quoted(s: &str) -> Option<String> {
     Some(s[start..end].to_string())
 }
 
+// ── Pipeline Parsing ──
+
+/// Extract constants: `const NAME: type = value;` → HashMap
+fn extract_constants(source: &str) -> std::collections::HashMap<String, String> {
+    let mut map = std::collections::HashMap::new();
+    for line in source.lines() {
+        let trimmed = line.trim();
+        if trimmed.starts_with("const ") {
+            // const NAME: type = "value";
+            let parts: Vec<&str> = trimmed.splitn(2, '=').collect();
+            if parts.len() == 2 {
+                let name_part = parts[0].trim();
+                let name = name_part.split(':').next().unwrap_or("").trim()
+                    .strip_prefix("const ").unwrap_or("").trim();
+                let val = parts[1].trim().trim_end_matches(';').trim();
+                // Store quoted and numeric values
+                if let Some(q) = extract_quoted(val) {
+                    map.insert(name.to_string(), q);
+                } else if let Ok(n) = val.parse::<u64>() {
+                    map.insert(name.to_string(), n.to_string());
+                }
+            }
+        }
+    }
+    map
+}
+
+/// Resolve constant references in source text.
+fn resolve_constants(source: &str, constants: &std::collections::HashMap<String, String>) -> String {
+    let mut result = source.to_string();
+    for (name, value) in constants {
+        // Replace uses like .port(WEBHOOK_PORT) → .port(3080)
+        result = result.replace(name, &format!("\"{}\"", value));
+    }
+    result
+}
+
+/// Extract workflow name from `vil_workflow! { name: "..." }`
+fn extract_workflow_name(source: &str) -> Option<String> {
+    for line in source.lines() {
+        let trimmed = line.trim();
+        if trimmed.starts_with("name:") && !trimmed.contains("vil_version") {
+            return extract_quoted(trimmed);
+        }
+    }
+    None
+}
+
+/// Extract pipeline nodes from HttpSinkBuilder/HttpSourceBuilder patterns.
+fn extract_pipeline_nodes(source: &str) -> Vec<ParsedNode> {
+    let mut nodes = Vec::new();
+    let mut current_node: Option<ParsedNode> = None;
+
+    for line in source.lines() {
+        let trimmed = line.trim();
+
+        // HttpSinkBuilder::new("Name")
+        if trimmed.contains("HttpSinkBuilder::new(") {
+            if let Some(node) = current_node.take() { nodes.push(node); }
+            let name = extract_quoted(trimmed).unwrap_or_else(|| "http_sink".to_string());
+            current_node = Some(ParsedNode {
+                name: to_snake(&name),
+                node_type: "http_sink".to_string(),
+                port: None, path: None, url: None, format: None, json_tap: None, dialect: None,
+            });
+        }
+
+        // HttpSourceBuilder::new("Name")
+        if trimmed.contains("HttpSourceBuilder::new(") {
+            if let Some(node) = current_node.take() { nodes.push(node); }
+            let name = extract_quoted(trimmed).unwrap_or_else(|| "http_source".to_string());
+            current_node = Some(ParsedNode {
+                name: to_snake(&name),
+                node_type: "http_source".to_string(),
+                port: None, path: None, url: None, format: None, json_tap: None, dialect: None,
+            });
+        }
+
+        // Chained builder methods
+        if let Some(ref mut node) = current_node {
+            if trimmed.starts_with(".port(") {
+                if let Some(q) = extract_quoted(trimmed) {
+                    node.port = q.parse().ok();
+                }
+            }
+            if trimmed.starts_with(".path(") {
+                node.path = extract_quoted(trimmed);
+            }
+            if trimmed.starts_with(".url(") {
+                node.url = extract_quoted(trimmed);
+            }
+            if trimmed.starts_with(".format(") {
+                if trimmed.contains("SSE") { node.format = Some("sse".to_string()); }
+                else if trimmed.contains("JSON") { node.format = Some("json".to_string()); }
+                else if trimmed.contains("NDJSON") { node.format = Some("ndjson".to_string()); }
+            }
+            if trimmed.starts_with(".json_tap(") {
+                node.json_tap = extract_quoted(trimmed);
+            }
+            if trimmed.starts_with(".dialect(") {
+                if trimmed.contains("OpenAi") { node.dialect = Some("openai".to_string()); }
+                else if trimmed.contains("Anthropic") { node.dialect = Some("anthropic".to_string()); }
+                else if trimmed.contains("Ollama") { node.dialect = Some("ollama".to_string()); }
+            }
+        }
+    }
+
+    if let Some(node) = current_node { nodes.push(node); }
+    nodes
+}
+
+/// Extract pipeline routes from `vil_workflow! { routes: [...] }`.
+fn extract_pipeline_routes(source: &str) -> Vec<ParsedRoute> {
+    let mut routes = Vec::new();
+    let mut in_routes = false;
+
+    for line in source.lines() {
+        let trimmed = line.trim();
+
+        if trimmed.starts_with("routes:") && trimmed.contains('[') {
+            in_routes = true;
+            continue;
+        }
+
+        if in_routes {
+            if trimmed.contains(']') { in_routes = false; continue; }
+
+            // Pattern: sink_builder.trigger_out -> source_builder.trigger_in (LoanWrite),
+            if trimmed.contains("->") && trimmed.contains('(') {
+                let parts: Vec<&str> = trimmed.split("->").collect();
+                if parts.len() == 2 {
+                    let from = parts[0].trim().replace("_builder", "");
+                    let to_mode = parts[1].trim().trim_end_matches(',');
+
+                    // Split "source.port (Mode)"
+                    let to_parts: Vec<&str> = to_mode.split('(').collect();
+                    let to = to_parts[0].trim().replace("_builder", "");
+                    let mode = if to_parts.len() > 1 {
+                        to_parts[1].trim().trim_end_matches(')').trim().to_string()
+                    } else {
+                        "LoanWrite".to_string()
+                    };
+
+                    // Convert snake_case builder names to node names
+                    let from_name = from.split('.').next().unwrap_or(&from);
+                    let from_port = from.split('.').nth(1).unwrap_or("data_out");
+                    let to_name = to.split('.').next().unwrap_or(&to);
+                    let to_port = to.split('.').nth(1).unwrap_or("data_in");
+
+                    routes.push(ParsedRoute {
+                        from: format!("{}.{}", to_snake(from_name), from_port),
+                        to: format!("{}.{}", to_snake(to_name), to_port),
+                        mode,
+                    });
+                }
+            }
+        }
+    }
+
+    routes
+}
+
+/// Convert PascalCase/camelCase to snake_case.
+fn to_snake(s: &str) -> String {
+    let trimmed = s.trim();
+    let mut result = String::new();
+    for (i, ch) in trimmed.chars().enumerate() {
+        if ch.is_uppercase() && i > 0 {
+            result.push('_');
+        }
+        result.push(ch.to_lowercase().next().unwrap_or(ch));
+    }
+    result
+}
+
 // ── Tests ──
 
 #[cfg(test)]
@@ -270,6 +526,9 @@ VilApp::new("my-server")
         let app = ParsedApp {
             name: "test-app".to_string(),
             port: 8080,
+            mode: AppMode::Server,
+            nodes: vec![],
+            routes: vec![],
             services: vec![ParsedService {
                 name: "tasks".to_string(),
                 endpoints: vec![

examples/001-basic-ai-gw-demo/manifest.yaml

@@ -1,4 +1,27 @@
 vil_version: "6.0.0"
-name: app
-port: 8080
-mode: server
+name: DecomposedPipeline
+port: 3080
+token: shm
+
+nodes:
+  webhook_trigger:
+    type: http_sink
+    port: 3080
+    path: "/trigger"
+  sse_inference:
+    type: http_source
+    url: "http://127.0.0.1:4545/v1/chat/completions"
+    format: sse
+    json_tap: "choices[0].delta.content"
+    dialect: openai
+
+routes:
+  - from: sink.trigger_out
+    to: source.trigger_in
+    mode: LoanWrite
+  - from: source.response_data_out
+    to: sink.response_data_in
+    mode: LoanWrite
+  - from: source.response_ctrl_out
+    to: sink.response_ctrl_in
+    mode: Copy

examples/003-basic-hello-server/manifest.yaml

@@ -1,6 +1,7 @@
 vil_version: "6.0.0"
 name: vil-basic-hello-server
 port: 8080
+token: shm
 mode: server
 
 services:

examples/004-basic-rest-crud/manifest.yaml

@@ -1,6 +1,7 @@
 vil_version: "6.0.0"
 name: crud-vilorm
 port: 8080
+token: shm
 mode: server
 
 services: