gateway/modules/features/graphicalEditor/nodeAdapter.py

# Copyright (c) 2025 Patrick Motsch
# All rights reserved.
"""
Schicht-3 Adapter Layer — projects Schicht-2 Actions into Editor-Node form.

Architecture (see wiki/c-work/1-plan/2026-04-typed-action-architecture.md):
  - Schicht 1: Types Catalog (portTypes.PORT_TYPE_CATALOG)
  - Schicht 2: Methods/Actions (modules/workflows/methods/method*) - source of truth
                for Backend capabilities (parameter types, output types).
  - Schicht 3: Adapters (this module) - Editor-Node + AI-Agent-Tool wrappers around
                Actions. References Action signature, never duplicates types.
  - Schicht 4: Workflow-Bindings + Agent-Tool-Calls (instance-level wiring).

This module defines the in-code Adapter representation (NodeAdapter,
UserParamMapping) and the projection helpers that convert between the
legacy node-dict wire format and the typed Adapter view.

Wire-format compatibility: the legacy dicts in nodeDefinitions/*.py remain
the wire format consumed by the frontend until Phase 4. This module exposes
an Adapter VIEW over those dicts so the validator and AI-tool generator can
operate on a clean, typed structure without breaking consumers.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Dict, List, Mapping, Optional


@dataclass(frozen=True)
class UserParamMapping:
    """Maps an Action argument into a Node's user-facing parameter.

    The Action signature is the source of truth for type/required/description.
    This mapping carries Editor-specific overrides (label, UI hints, conditional
    visibility) but never re-declares the type.
    """

    actionArg: str
    label: Optional[Any] = None
    description: Optional[Any] = None
    uiHint: Optional[str] = None
    frontendOptions: Optional[Any] = None
    visibleWhen: Optional[Dict[str, Any]] = None
    defaultValue: Any = None


@dataclass(frozen=True)
class NodeAdapter:
    """Schicht-3 Editor-Node adapter — binds to a Schicht-2 Action.

    All type information for `userParams` is inherited from the bound Action.
    The adapter only carries Editor-specific concerns (UI labels, port topology,
    icon/color metadata).
    """

    nodeId: str
    bindsAction: str
    category: str
    label: Any
    description: Any
    userParams: List[UserParamMapping] = field(default_factory=list)
    contextParams: Dict[str, str] = field(default_factory=dict)
    inputs: int = 1
    outputs: int = 1
    inputAccepts: List[List[str]] = field(default_factory=list)
    outputLabels: Optional[List[Any]] = None
    meta: Dict[str, Any] = field(default_factory=dict)


def _isMethodBoundNode(node: Mapping[str, Any]) -> bool:
    """True if a legacy node dict is bound to a Schicht-2 Action."""
    return bool(node.get("_method") and node.get("_action"))


def bindsActionFromLegacy(node: Mapping[str, Any]) -> Optional[str]:
    """Build the canonical 'method.action' identifier from a legacy node dict.

    Returns None for framework-primitive nodes (trigger/flow/input/data).
    """
    method = node.get("_method")
    action = node.get("_action")
    if not method or not action:
        return None
    return f"{method}.{action}"


def _userParamFromLegacyParam(legacyParam: Mapping[str, Any]) -> UserParamMapping:
    """Project a legacy parameter dict into a UserParamMapping view.

    The view carries only Editor-overrides; type/required come from the Action.
    """
    return UserParamMapping(
        actionArg=str(legacyParam.get("name", "")),
        label=legacyParam.get("label"),
        description=legacyParam.get("description"),
        uiHint=legacyParam.get("frontendType"),
        frontendOptions=legacyParam.get("frontendOptions"),
        visibleWhen=_extractVisibleWhen(legacyParam.get("frontendOptions")),
        defaultValue=legacyParam.get("default"),
    )


def _extractVisibleWhen(frontendOptions: Any) -> Optional[Dict[str, Any]]:
    """Extract conditional-visibility hint from legacy frontendOptions.showWhen."""
    if not isinstance(frontendOptions, dict):
        return None
    dependsOn = frontendOptions.get("dependsOn")
    showWhen = frontendOptions.get("showWhen")
    if not dependsOn or not showWhen:
        return None
    return {"actionArg": str(dependsOn), "in": list(showWhen) if isinstance(showWhen, (list, tuple)) else [showWhen]}


def _adapterFromLegacyNode(node: Mapping[str, Any]) -> Optional[NodeAdapter]:
    """Build a NodeAdapter view from a legacy node dict.

    Returns None for framework-primitive nodes (no _method/_action binding).
    Pure projection — no validation, no Action-signature lookup.
    """
    if not _isMethodBoundNode(node):
        return None

    bindsAction = bindsActionFromLegacy(node)
    if not bindsAction:
        return None

    inputAccepts = _projectInputAccepts(node)

    return NodeAdapter(
        nodeId=str(node.get("id", "")),
        bindsAction=bindsAction,
        category=str(node.get("category", "")),
        label=node.get("label", ""),
        description=node.get("description", ""),
        userParams=[_userParamFromLegacyParam(p) for p in (node.get("parameters") or [])],
        contextParams={},
        inputs=int(node.get("inputs", 1)),
        outputs=int(node.get("outputs", 1)),
        inputAccepts=inputAccepts,
        outputLabels=node.get("outputLabels"),
        meta=dict(node.get("meta") or {}),
    )


def _projectInputAccepts(node: Mapping[str, Any]) -> List[List[str]]:
    """Convert legacy `inputPorts` dict-of-dicts into a per-port `accepts` list."""
    inputPorts = node.get("inputPorts") or {}
    if not isinstance(inputPorts, dict):
        return []
    inputs = int(node.get("inputs", 0) or 0)
    if inputs <= 0:
        return []
    out: List[List[str]] = []
    for portIdx in range(inputs):
        portCfg = inputPorts.get(portIdx) or inputPorts.get(str(portIdx)) or {}
        accepts = portCfg.get("accepts") if isinstance(portCfg, dict) else None
        out.append(list(accepts) if isinstance(accepts, (list, tuple)) else [])
    return out


def _projectAllAdapters(staticNodes: List[Mapping[str, Any]]) -> Dict[str, NodeAdapter]:
    """Project a list of legacy node dicts into a {nodeId: NodeAdapter} map.

    Framework-primitive nodes (no Action binding) are silently skipped.
    """
    out: Dict[str, NodeAdapter] = {}
    for node in staticNodes:
        adapter = _adapterFromLegacyNode(node)
        if adapter is not None:
            out[adapter.nodeId] = adapter
    return out