ROADMAP.md

python/package_lint.py

@@ -11,6 +11,7 @@ import argparse
 import json
 import logging
 import sys
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Callable, Iterable, Optional, Sequence
 
@@ -22,8 +23,6 @@ FIELD_TO_FOLDER = {
     "shaders": "shaders",
     "workflows": "workflows",
 }
-WORKFLOW_TOP_LEVEL_KEYS = {"template", "nodes", "steps", "connections"}
-WORKFLOW_NODE_KEYS = {"id", "name", "plugin", "type", "position", "inputs", "outputs", "parameters"}
 PACKAGE_ALLOWED_KEYS = {
     "name",
     "version",
@@ -38,60 +37,29 @@ PACKAGE_ALLOWED_KEYS = {
     "notes",
 }
 
-
-class WorkflowReferenceProfile:
-    def __init__(self,
-                 required_top_keys: set[str],
-                 allowed_top_keys: set[str],
-                 require_nodes: bool,
-                 require_template: bool,
-                 require_connections: bool,
-                 require_id: bool,
-                 require_plugin: bool,
-                 require_position: bool):
-        self.required_top_keys = required_top_keys
-        self.allowed_top_keys = allowed_top_keys
-        self.require_nodes = require_nodes
-        self.require_template = require_template
-        self.require_connections = require_connections
-        self.require_id = require_id
-        self.require_plugin = require_plugin
-        self.require_position = require_position
-
-
-def build_workflow_profile(reference: dict) -> WorkflowReferenceProfile:
-    required_top_keys = set(reference.keys())
-    allowed_top_keys = set(reference.keys())
-    require_nodes = "nodes" in reference
-    require_template = "template" in reference
-    require_connections = "connections" in reference
-    require_id = True
-    require_plugin = True
-    require_position = False
-    if require_nodes:
-        nodes = reference.get("nodes")
-        if isinstance(nodes, list) and nodes:
-            require_position = all(
-                isinstance(node, dict) and "position" in node
-                for node in nodes
-            )
-    return WorkflowReferenceProfile(
-        required_top_keys,
-        allowed_top_keys,
-        require_nodes,
-        require_template,
-        require_connections,
-        require_id,
-        require_plugin,
-        require_position,
-    )
-
 logger = logging.getLogger("package_lint")
 
 try:
-    from jsonschema import Draft7Validator
+    from jsonschema import Draft202012Validator
 except ImportError:
-    Draft7Validator = None
+    Draft202012Validator = None
+
+
+@dataclass(frozen=True)
+class WorkflowSchemaDefinition:
+    raw_schema: dict
+    top_level_keys: set[str]
+    required_top_keys: set[str]
+    node_keys: set[str]
+    node_required: set[str]
+    tag_keys: set[str]
+    tag_required: set[str]
+    settings_keys: set[str]
+    credential_ref_keys: set[str]
+    credential_ref_required: set[str]
+    credential_binding_keys: set[str]
+    credential_binding_required: set[str]
+    connection_types: set[str]
 
 
 def load_json(path: Path) -> dict:
@@ -100,6 +68,58 @@ def load_json(path: Path) -> dict:
         return json.load(handle)
 
 
+def load_schema_from_roadmap(roadmap_path: Path) -> dict:
+    if not roadmap_path.exists():
+        raise FileNotFoundError(f"ROADMAP not found at {roadmap_path}")
+    lines = roadmap_path.read_text(encoding="utf-8").splitlines()
+    in_schema_section = False
+    in_block = False
+    schema_lines: list[str] = []
+    for line in lines:
+        if not in_schema_section:
+            if line.strip().lower() == "n8n style schema:":
+                in_schema_section = True
+            continue
+        if not in_block:
+            if line.strip().startswith("```json"):
+                in_block = True
+            continue
+        if line.strip().startswith("```"):
+            break
+        schema_lines.append(line)
+    if not schema_lines:
+        raise ValueError("Failed to locate n8n schema block in ROADMAP.md")
+    return json.loads("\n".join(schema_lines))
+
+
+def build_schema_definition(schema: dict) -> WorkflowSchemaDefinition:
+    properties = schema.get("properties") or {}
+    required = schema.get("required") or []
+    defs = schema.get("$defs") or {}
+    node_def = defs.get("node") or {}
+    tag_def = defs.get("tag") or {}
+    settings_def = defs.get("workflowSettings") or {}
+    credential_ref_def = defs.get("credentialRef") or {}
+    credential_binding_def = defs.get("credentialBinding") or {}
+    connections_def = defs.get("nodeConnectionsByType") or {}
+    connection_types = set((connections_def.get("properties") or {}).keys())
+    return WorkflowSchemaDefinition(
+        raw_schema=schema,
+        top_level_keys=set(properties.keys()),
+        required_top_keys=set(required),
+        node_keys=set((node_def.get("properties") or {}).keys()),
+        node_required=set(node_def.get("required") or []),
+        tag_keys=set((tag_def.get("properties") or {}).keys()),
+        tag_required=set(tag_def.get("required") or []),
+        settings_keys=set((settings_def.get("properties") or {}).keys()),
+        credential_ref_keys=set((credential_ref_def.get("properties") or {}).keys()),
+        credential_ref_required=set(credential_ref_def.get("required") or []),
+        credential_binding_keys=set((credential_binding_def.get("properties") or {}).keys()),
+        credential_binding_required=set(credential_binding_def.get("required") or []),
+        connection_types=connection_types or {"main", "error"},
+    )
+
+
 def check_paths(
     root: Path,
     entries: Iterable[str],
@@ -122,27 +142,18 @@ def check_paths(
     return missing
 
 
-def validate_workflow_schema(workflow_path: Path, validator) -> list[str]:
-    """Validate a workflow JSON file against the provided schema validator."""
-    try:
-        content = load_json(workflow_path)
-    except json.JSONDecodeError as exc:
-        return [f"invalid JSON: {exc}"]
-
-    issues: list[str] = []
-    for err in sorted(
-        validator.iter_errors(content),
-        key=lambda x: tuple(x.absolute_path),
-    ):
-        pointer = "/".join(str(part) for part in err.absolute_path) or "<root>"
-        issues.append(f"schema violation at {pointer}: {err.message}")
-    return issues
-
-
 def _is_non_empty_string(value: object) -> bool:
     return isinstance(value, str) and value.strip() != ""
 
 
+def _is_number(value: object) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool)
+
+
+def _is_int(value: object) -> bool:
+    return isinstance(value, int) and not isinstance(value, bool)
+
+
 def _validate_string_map(value: object, context: str) -> list[str]:
     if not isinstance(value, dict):
         return [f"{context} must be an object"]
@@ -185,44 +196,195 @@ def _validate_parameters(value: object) -> list[str]:
         if not _is_non_empty_string(key):
             issues.append("parameters keys must be non-empty strings")
             continue
+        if key in {"inputs", "outputs"}:
+            issues.extend(_validate_string_map(item, f"parameters.{key}"))
+            continue
         issues.extend(_validate_parameter_value(item, f"parameters.{key}"))
     return issues
 
 
-def _validate_node_entry(node: dict,
-                         index: int,
-                         reference_profile: Optional[WorkflowReferenceProfile]) -> tuple[str, list[str]]:
+def _validate_tags(tags: object, schema_def: WorkflowSchemaDefinition) -> list[str]:
+    if not isinstance(tags, list):
+        return ["tags must be an array"]
     issues: list[str] = []
-    if not isinstance(node, dict):
-        return "", [f"nodes[{index}] must be an object"]
-    extra_keys = set(node.keys()) - WORKFLOW_NODE_KEYS
+    for index, tag in enumerate(tags):
+        if not isinstance(tag, dict):
+            issues.append(f"tags[{index}] must be an object")
+            continue
+        extra_keys = set(tag.keys()) - schema_def.tag_keys
+        if extra_keys:
+            issues.append(f"tags[{index}] has unsupported keys: {sorted(extra_keys)}")
+        missing_keys = schema_def.tag_required - set(tag.keys())
+        if missing_keys:
+            issues.append(f"tags[{index}] missing required keys: {sorted(missing_keys)}")
+        name = tag.get("name")
+        if not _is_non_empty_string(name):
+            issues.append(f"tags[{index}].name must be a non-empty string")
+        if "id" in tag and not isinstance(tag["id"], (str, int)):
+            issues.append(f"tags[{index}].id must be a string or integer")
+    return issues
+
+
+def _validate_settings(settings: object, schema_def: WorkflowSchemaDefinition) -> list[str]:
+    if not isinstance(settings, dict):
+        return ["settings must be an object"]
+    issues: list[str] = []
+    extra_keys = set(settings.keys()) - schema_def.settings_keys
     if extra_keys:
-        issues.append(
-            f"nodes[{index}] has unsupported keys: {sorted(extra_keys)}"
-        )
-    node_id = node.get("id") if reference_profile and reference_profile.require_id else node.get("id") or node.get("name")
-    if not _is_non_empty_string(node_id):
-        issues.append(f"nodes[{index}] requires non-empty id")
-    plugin = node.get("plugin") if reference_profile and reference_profile.require_plugin else node.get("plugin") or node.get("type")
-    if not _is_non_empty_string(plugin):
-        issues.append(f"nodes[{index}] requires non-empty plugin")
-    if "inputs" in node:
-        issues.extend(_validate_string_map(node["inputs"], f"nodes[{index}].inputs"))
-    if "outputs" in node:
-        issues.extend(_validate_string_map(node["outputs"], f"nodes[{index}].outputs"))
-    if "parameters" in node:
-        issues.extend(_validate_parameters(node["parameters"]))
-    if reference_profile and reference_profile.require_position and "position" not in node:
-        issues.append(f"nodes[{index}] requires position")
-    if "position" in node:
-        position = node["position"]
-        if (not isinstance(position, list) or len(position) != 2 or
-                not all(isinstance(item, (int, float)) for item in position)):
-            issues.append(f"nodes[{index}].position must be [x, y] numbers")
-    return (node_id if isinstance(node_id, str) else ""), issues
+        issues.append(f"settings has unsupported keys: {sorted(extra_keys)}")
+    if "timezone" in settings and not _is_non_empty_string(settings["timezone"]):
+        issues.append("settings.timezone must be a non-empty string")
+    if "executionTimeout" in settings:
+        value = settings["executionTimeout"]
+        if not _is_int(value) or value < 0:
+            issues.append("settings.executionTimeout must be an integer >= 0")
+    for key in ("saveExecutionProgress", "saveManualExecutions"):
+        if key in settings and not isinstance(settings[key], bool):
+            issues.append(f"settings.{key} must be a boolean")
+    for key in ("saveDataErrorExecution", "saveDataSuccessExecution", "saveDataManualExecution"):
+        if key in settings:
+            value = settings[key]
+            if not _is_non_empty_string(value):
+                issues.append(f"settings.{key} must be a non-empty string")
+            elif value not in {"all", "none"}:
+                issues.append(f"settings.{key} must be 'all' or 'none'")
+    if "errorWorkflowId" in settings and not isinstance(settings["errorWorkflowId"], (str, int)):
+        issues.append("settings.errorWorkflowId must be a string or integer")
+    if "callerPolicy" in settings and not _is_non_empty_string(settings["callerPolicy"]):
+        issues.append("settings.callerPolicy must be a non-empty string")
+    return issues
 
 
-def _validate_connections(connections: object, node_ids: set[str]) -> list[str]:
+def _validate_credential_ref(value: object, context: str, schema_def: WorkflowSchemaDefinition) -> list[str]:
+    if not isinstance(value, dict):
+        return [f"{context} must be an object"]
+    issues: list[str] = []
+    extra_keys = set(value.keys()) - schema_def.credential_ref_keys
+    if extra_keys:
+        issues.append(f"{context} has unsupported keys: {sorted(extra_keys)}")
+    missing = schema_def.credential_ref_required - set(value.keys())
+    if missing:
+        issues.append(f"{context} missing required keys: {sorted(missing)}")
+    if "id" in value and not isinstance(value["id"], (str, int)):
+        issues.append(f"{context}.id must be a string or integer")
+    if "name" in value and not _is_non_empty_string(value["name"]):
+        issues.append(f"{context}.name must be a non-empty string")
+    return issues
+
+
+def _validate_credential_binding(value: object, index: int, schema_def: WorkflowSchemaDefinition) -> list[str]:
+    context = f"credentials[{index}]"
+    if not isinstance(value, dict):
+        return [f"{context} must be an object"]
+    issues: list[str] = []
+    extra_keys = set(value.keys()) - schema_def.credential_binding_keys
+    if extra_keys:
+        issues.append(f"{context} has unsupported keys: {sorted(extra_keys)}")
+    missing = schema_def.credential_binding_required - set(value.keys())
+    if missing:
+        issues.append(f"{context} missing required keys: {sorted(missing)}")
+    if "nodeId" in value and not _is_non_empty_string(value["nodeId"]):
+        issues.append(f"{context}.nodeId must be a non-empty string")
+    if "credentialType" in value and not _is_non_empty_string(value["credentialType"]):
+        issues.append(f"{context}.credentialType must be a non-empty string")
+    if "credentialId" in value and not isinstance(value["credentialId"], (str, int)):
+        issues.append(f"{context}.credentialId must be a string or integer")
+    return issues
+
+
+def _validate_nodes(nodes: object, schema_def: WorkflowSchemaDefinition) -> tuple[list[str], list[str], list[str]]:
+    if not isinstance(nodes, list):
+        return ["nodes must be an array"], [], []
+    if not nodes:
+        return ["nodes must contain at least one node"], [], []
+    issues: list[str] = []
+    node_names: list[str] = []
+    node_ids: list[str] = []
+    seen_names: set[str] = set()
+    seen_ids: set[str] = set()
+    for index, node in enumerate(nodes):
+        if not isinstance(node, dict):
+            issues.append(f"nodes[{index}] must be an object")
+            continue
+        extra_keys = set(node.keys()) - schema_def.node_keys
+        if extra_keys:
+            issues.append(f"nodes[{index}] has unsupported keys: {sorted(extra_keys)}")
+        missing_keys = schema_def.node_required - set(node.keys())
+        if missing_keys:
+            issues.append(f"nodes[{index}] missing required keys: {sorted(missing_keys)}")
+        node_id = node.get("id")
+        if not _is_non_empty_string(node_id):
+            issues.append(f"nodes[{index}].id must be a non-empty string")
+        else:
+            if node_id in seen_ids:
+                issues.append(f"duplicate node id '{node_id}'")
+            seen_ids.add(node_id)
+            node_ids.append(node_id)
+        node_name = node.get("name")
+        if not _is_non_empty_string(node_name):
+            issues.append(f"nodes[{index}].name must be a non-empty string")
+        else:
+            if node_name in seen_names:
+                issues.append(f"duplicate node name '{node_name}'")
+            seen_names.add(node_name)
+            node_names.append(node_name)
+        node_type = node.get("type")
+        if not _is_non_empty_string(node_type):
+            issues.append(f"nodes[{index}].type must be a non-empty string")
+        version = node.get("typeVersion")
+        if version is not None:
+            if not _is_number(version) or version < 1:
+                issues.append(f"nodes[{index}].typeVersion must be a number >= 1")
+        position = node.get("position")
+        if position is not None:
+            if (not isinstance(position, list) or len(position) != 2 or
+                    not all(_is_number(item) for item in position)):
+                issues.append(f"nodes[{index}].position must be [x, y] numbers")
+        for key in ("disabled", "notesInFlow", "retryOnFail", "continueOnFail",
+                    "alwaysOutputData", "executeOnce"):
+            if key in node and not isinstance(node[key], bool):
+                issues.append(f"nodes[{index}].{key} must be a boolean")
+        if "notes" in node and not isinstance(node["notes"], str):
+            issues.append(f"nodes[{index}].notes must be a string")
+        if "maxTries" in node:
+            value = node["maxTries"]
+            if not _is_int(value) or value < 1:
+                issues.append(f"nodes[{index}].maxTries must be an integer >= 1")
+        if "waitBetweenTries" in node:
+            value = node["waitBetweenTries"]
+            if not _is_int(value) or value < 0:
+                issues.append(f"nodes[{index}].waitBetweenTries must be an integer >= 0")
+        if "parameters" in node:
+            issues.extend(_validate_parameters(node["parameters"]))
+        if "credentials" in node:
+            credentials = node["credentials"]
+            if not isinstance(credentials, dict):
+                issues.append(f"nodes[{index}].credentials must be an object")
+            else:
+                for cred_key, cred_value in credentials.items():
+                    if not _is_non_empty_string(cred_key):
+                        issues.append(f"nodes[{index}].credentials keys must be non-empty strings")
+                        continue
+                    issues.extend(
+                        _validate_credential_ref(
+                            cred_value,
+                            f"nodes[{index}].credentials.{cred_key}",
+                            schema_def,
+                        )
+                    )
+        if "webhookId" in node and not _is_non_empty_string(node["webhookId"]):
+            issues.append(f"nodes[{index}].webhookId must be a non-empty string")
+        if "onError" in node:
+            value = node["onError"]
+            allowed = {"stopWorkflow", "continueRegularOutput", "continueErrorOutput"}
+            if not _is_non_empty_string(value) or value not in allowed:
+                issues.append(f"nodes[{index}].onError must be one of {sorted(allowed)}")
+    return issues, node_names, node_ids
+
+
+def _validate_connections(connections: object,
+                          node_names: set[str],
+                          schema_def: WorkflowSchemaDefinition) -> list[str]:
     if not isinstance(connections, dict):
         return ["connections must be an object"]
     issues: list[str] = []
@@ -230,124 +392,129 @@ def _validate_connections(connections: object, node_ids: set[str]) -> list[str]:
         if not _is_non_empty_string(from_node):
             issues.append("connections keys must be non-empty strings")
             continue
-        if from_node not in node_ids:
+        if from_node not in node_names:
             issues.append(f"connections references unknown node '{from_node}'")
         if not isinstance(link, dict):
             issues.append(f"connections.{from_node} must be an object")
             continue
-        extra_keys = set(link.keys()) - {"main"}
+        extra_keys = set(link.keys()) - schema_def.connection_types
         if extra_keys:
             issues.append(f"connections.{from_node} has unsupported keys: {sorted(extra_keys)}")
-        if "main" not in link:
-            continue
-        main_value = link["main"]
-        if not isinstance(main_value, list):
-            issues.append(f"connections.{from_node}.main must be an array")
-            continue
-        for branch_index, branch in enumerate(main_value):
-            if not isinstance(branch, list):
-                issues.append(f"connections.{from_node}.main[{branch_index}] must be an array")
+        if not any(key in link for key in schema_def.connection_types):
+            issues.append(f"connections.{from_node} must define at least one connection type")
+        for conn_type in schema_def.connection_types:
+            if conn_type not in link:
                 continue
-            for entry_index, entry in enumerate(branch):
-                if not isinstance(entry, dict):
+            index_map = link[conn_type]
+            if not isinstance(index_map, dict):
+                issues.append(f"connections.{from_node}.{conn_type} must be an object")
+                continue
+            for index_key, targets in index_map.items():
+                if not _is_non_empty_string(index_key) or not index_key.isdigit():
                     issues.append(
-                        f"connections.{from_node}.main[{branch_index}][{entry_index}] must be an object"
+                        f"connections.{from_node}.{conn_type} index keys must be numeric strings"
                     )
                     continue
-                node_name = entry.get("node")
-                if not _is_non_empty_string(node_name):
+                if not isinstance(targets, list):
                     issues.append(
-                        f"connections.{from_node}.main[{branch_index}][{entry_index}] missing node"
+                        f"connections.{from_node}.{conn_type}.{index_key} must be an array"
                     )
                     continue
-                if node_name not in node_ids:
-                    issues.append(
-                        f"connections.{from_node}.main[{branch_index}][{entry_index}] "
-                        f"references unknown node '{node_name}'"
-                    )
-                if "type" in entry and not _is_non_empty_string(entry["type"]):
-                    issues.append(
-                        f"connections.{from_node}.main[{branch_index}][{entry_index}].type "
-                        "must be a non-empty string"
-                    )
-                if "index" in entry and not isinstance(entry["index"], int):
-                    issues.append(
-                        f"connections.{from_node}.main[{branch_index}][{entry_index}].index "
-                        "must be an integer"
-                    )
+                for target_index, target in enumerate(targets):
+                    context = f"connections.{from_node}.{conn_type}.{index_key}[{target_index}]"
+                    if not isinstance(target, dict):
+                        issues.append(f"{context} must be an object")
+                        continue
+                    extra_keys = set(target.keys()) - {"node", "type", "index"}
+                    if extra_keys:
+                        issues.append(f"{context} has unsupported keys: {sorted(extra_keys)}")
+                    node_name = target.get("node")
+                    if not _is_non_empty_string(node_name):
+                        issues.append(f"{context}.node must be a non-empty string")
+                    elif node_name not in node_names:
+                        issues.append(f"{context} references unknown node '{node_name}'")
+                    if "type" in target and not _is_non_empty_string(target["type"]):
+                        issues.append(f"{context}.type must be a non-empty string")
+                    if "index" in target:
+                        index_value = target["index"]
+                        if not _is_int(index_value) or index_value < 0:
+                            issues.append(f"{context}.index must be an integer >= 0")
     return issues
 
 
 def validate_workflow_structure(workflow_path: Path,
                                 content: dict,
-                                reference_profile: Optional[WorkflowReferenceProfile]) -> list[str]:
+                                schema_def: WorkflowSchemaDefinition) -> list[str]:
     issues: list[str] = []
     logger.debug("Validating workflow structure: %s", workflow_path)
-    allowed_top_keys = WORKFLOW_TOP_LEVEL_KEYS
-    required_top_keys = set()
-    if reference_profile:
-        allowed_top_keys = reference_profile.allowed_top_keys
-        required_top_keys = reference_profile.required_top_keys
-    extra_keys = set(content.keys()) - allowed_top_keys
+    extra_keys = set(content.keys()) - schema_def.top_level_keys
     if extra_keys:
         issues.append(f"unsupported workflow keys: {sorted(extra_keys)}")
-    missing_keys = required_top_keys - set(content.keys())
+    missing_keys = schema_def.required_top_keys - set(content.keys())
     if missing_keys:
         issues.append(f"workflow missing required keys: {sorted(missing_keys)}")
-    has_nodes = "nodes" in content
-    has_steps = "steps" in content
-    if has_nodes and has_steps:
-        issues.append("workflow cannot define both 'nodes' and 'steps'")
-    if reference_profile and reference_profile.require_nodes and has_steps:
-        issues.append("workflow must not define 'steps' when using reference schema")
-    if not has_nodes and not has_steps:
-        issues.append("workflow must define 'nodes' or 'steps'")
-        return issues
-    if reference_profile and reference_profile.require_template and "template" not in content:
-        issues.append("workflow missing required template")
-    if "template" in content and not _is_non_empty_string(content["template"]):
-        issues.append("workflow template must be a non-empty string")
-    if reference_profile and reference_profile.require_connections and "connections" not in content:
-        issues.append("workflow missing required connections")
+    if "name" in content and not _is_non_empty_string(content["name"]):
+        issues.append("workflow name must be a non-empty string")
+    if "id" in content and not isinstance(content["id"], (str, int)):
+        issues.append("workflow id must be a string or integer")
+    if "active" in content and not isinstance(content["active"], bool):
+        issues.append("workflow active must be a boolean")
+    for key in ("versionId", "createdAt", "updatedAt"):
+        if key in content and not isinstance(content[key], str):
+            issues.append(f"workflow {key} must be a string")
+    if "tags" in content:
+        issues.extend(_validate_tags(content["tags"], schema_def))
+    if "meta" in content and not isinstance(content["meta"], dict):
+        issues.append("workflow meta must be an object")
+    if "settings" in content:
+        issues.extend(_validate_settings(content["settings"], schema_def))
+    if "pinData" in content:
+        pin_data = content["pinData"]
+        if not isinstance(pin_data, dict):
+            issues.append("workflow pinData must be an object")
+        else:
+            for pin_key, pin_value in pin_data.items():
+                if not _is_non_empty_string(pin_key):
+                    issues.append("workflow pinData keys must be non-empty strings")
+                    continue
+                if not isinstance(pin_value, list):
+                    issues.append(f"workflow pinData.{pin_key} must be an array")
+                    continue
+                for entry_index, entry in enumerate(pin_value):
+                    if not isinstance(entry, dict):
+                        issues.append(f"workflow pinData.{pin_key}[{entry_index}] must be an object")
+    if "staticData" in content and not isinstance(content["staticData"], dict):
+        issues.append("workflow staticData must be an object")
+    node_issues: list[str] = []
+    node_names: list[str] = []
     node_ids: list[str] = []
-    if has_nodes:
-        nodes = content.get("nodes")
-        if not isinstance(nodes, list) or not nodes:
-            issues.append("workflow nodes must be a non-empty array")
-        else:
-            seen = set()
-            for index, node in enumerate(nodes):
-                node_id, node_issues = _validate_node_entry(node, index, reference_profile)
-                issues.extend(node_issues)
-                if node_id:
-                    if node_id in seen:
-                        issues.append(f"duplicate node id '{node_id}'")
-                    else:
-                        seen.add(node_id)
-                        node_ids.append(node_id)
-    if has_steps:
-        steps = content.get("steps")
-        if not isinstance(steps, list) or not steps:
-            issues.append("workflow steps must be a non-empty array")
-        else:
-            seen = set()
-            for index, step in enumerate(steps):
-                node_id, node_issues = _validate_node_entry(step, index, reference_profile)
-                issues.extend(node_issues)
-                if node_id:
-                    if node_id in seen:
-                        issues.append(f"duplicate step id '{node_id}'")
-                    else:
-                        seen.add(node_id)
-                        node_ids.append(node_id)
+    if "nodes" in content:
+        node_issues, node_names, node_ids = _validate_nodes(content["nodes"], schema_def)
+        issues.extend(node_issues)
     if "connections" in content:
-        issues.extend(_validate_connections(content["connections"], set(node_ids)))
+        issues.extend(_validate_connections(content["connections"], set(node_names), schema_def))
+    if "credentials" in content:
+        credentials = content["credentials"]
+        if not isinstance(credentials, list):
+            issues.append("workflow credentials must be an array")
+        else:
+            for index, entry in enumerate(credentials):
+                issues.extend(_validate_credential_binding(entry, index, schema_def))
+    if node_ids and "credentials" in content and isinstance(content.get("credentials"), list):
+        known_ids = set(node_ids)
+        for index, entry in enumerate(content.get("credentials", [])):
+            if isinstance(entry, dict) and "nodeId" in entry:
+                node_id = entry["nodeId"]
+                if isinstance(node_id, str) and node_id not in known_ids:
+                    issues.append(
+                        f"credentials[{index}].nodeId references unknown node id '{node_id}'"
+                    )
     return issues
 
 
 def validate_workflow(workflow_path: Path,
-                      validator: Optional["Draft7Validator"],
-                      reference_profile: Optional[WorkflowReferenceProfile]) -> list[str]:
+                      validator: Optional["Draft202012Validator"],
+                      schema_def: WorkflowSchemaDefinition) -> list[str]:
     try:
         content = load_json(workflow_path)
     except json.JSONDecodeError as exc:
@@ -360,7 +527,7 @@ def validate_workflow(workflow_path: Path,
         ):
             pointer = "/".join(str(part) for part in err.absolute_path) or "<root>"
             issues.append(f"schema violation at {pointer}: {err.message}")
-    issues.extend(validate_workflow_structure(workflow_path, content, reference_profile))
+    issues.extend(validate_workflow_structure(workflow_path, content, schema_def))
     return issues
 
 
@@ -369,8 +536,8 @@ def validate_package(
     pkg_data: dict,
     registry_names: Sequence[str],
     available_dirs: Sequence[str],
-    workflow_schema_validator: Optional["Draft7Validator"] = None,
-    workflow_reference_profile: Optional[WorkflowReferenceProfile] = None,
+    workflow_schema_validator: Optional["Draft202012Validator"],
+    workflow_schema_def: WorkflowSchemaDefinition,
 ) -> tuple[list[str], list[str]]:
     errors: list[str] = []
     warnings: list[str] = []
@@ -426,7 +593,7 @@ def validate_package(
             def on_exist(candidate: Path, rel: str) -> None:
                 schema_issues = validate_workflow(candidate,
                                                  workflow_schema_validator,
-                                                 workflow_reference_profile)
+                                                 workflow_schema_def)
                 for issue in schema_issues:
                     errors.append(f"workflow `{rel}`: {issue}")
         def validate_entry(entry: str) -> None:
@@ -488,15 +655,15 @@ def main() -> int:
         help="Root folder containing package directories",
     )
     parser.add_argument(
-        "--workflow-schema",
+        "--roadmap",
         type=Path,
-        help="Optional workflow JSON schema (default: config/schema/workflow_v1.schema.json when available)",
+        default=Path("ROADMAP.md"),
+        help="Path to ROADMAP containing the n8n workflow schema",
     )
     parser.add_argument(
-        "--workflow-reference",
+        "--workflow-schema",
         type=Path,
-        help="Reference n8n-style workflow JSON used to validate workflow structure "
-             "(default: packages/seed/workflows/demo_gameplay.json when available)",
+        help="Optional workflow JSON schema override",
     )
     parser.add_argument(
         "--verbose",
@@ -515,46 +682,39 @@ def main() -> int:
         return 2
 
     schema_candidate = args.workflow_schema
-    default_schema = Path("config/schema/workflow_v1.schema.json")
-    if schema_candidate is None and default_schema.exists():
-        schema_candidate = default_schema
+    if schema_candidate is None:
+        schema_candidate = args.roadmap
 
-    workflow_validator: Optional["Draft7Validator"] = None
+    workflow_schema: Optional[dict] = None
     if schema_candidate:
         if not schema_candidate.exists():
-            logger.error("specified workflow schema %s not found", schema_candidate)
+            logger.error("specified workflow schema source %s not found", schema_candidate)
             return 5
         try:
-            workflow_schema = load_json(schema_candidate)
-        except json.JSONDecodeError as exc:
-            logger.error("invalid JSON schema %s: %s", schema_candidate, exc)
+            workflow_schema = (
+                load_json(schema_candidate)
+                if schema_candidate.suffix == ".json"
+                else load_schema_from_roadmap(schema_candidate)
+            )
+        except (json.JSONDecodeError, ValueError, FileNotFoundError) as exc:
+            logger.error("invalid workflow schema source %s: %s", schema_candidate, exc)
             return 6
-        if Draft7Validator is None:
-            logger.warning("jsonschema dependency not installed; skipping workflow schema validation")
-        else:
-            try:
-                workflow_validator = Draft7Validator(workflow_schema)
-            except Exception as exc:
-                logger.error("failed to compile workflow schema %s: %s", schema_candidate, exc)
-                return 7
 
-    reference_path = args.workflow_reference
-    default_reference = Path("packages/seed/workflows/demo_gameplay.json")
-    if reference_path is None and default_reference.exists():
-        reference_path = default_reference
+    if not workflow_schema:
+        logger.error("workflow schema could not be loaded")
+        return 7
 
-    workflow_reference_profile: Optional[WorkflowReferenceProfile] = None
-    if reference_path:
-        if not reference_path.exists():
-            logger.error("specified workflow reference %s not found", reference_path)
-            return 8
+    workflow_schema_def = build_schema_definition(workflow_schema)
+    workflow_validator: Optional["Draft202012Validator"] = None
+    if Draft202012Validator is None:
+        logger.warning("jsonschema dependency not installed; skipping JSON Schema validation")
+    else:
         try:
-            reference_workflow = load_json(reference_path)
-        except json.JSONDecodeError as exc:
-            logger.error("invalid workflow reference %s: %s", reference_path, exc)
-            return 9
-        workflow_reference_profile = build_workflow_profile(reference_workflow)
-        logger.info("workflow reference loaded: %s", reference_path)
+            workflow_validator = Draft202012Validator(workflow_schema)
+        except Exception as exc:
+            logger.error("failed to compile workflow schema: %s", exc)
+            return 8
+    logger.info("workflow schema loaded from %s", schema_candidate)
 
     package_dirs = [
         child
@@ -594,7 +754,7 @@ def main() -> int:
             registry_names,
             available_dirs,
             workflow_validator,
-            workflow_reference_profile,
+            workflow_schema_def,
         )
         for err in errors:
             logger.error("%s: %s", pkg_json_file, err)