Skip to main content
Glama

nexus_transform_jewel

Transform WorkFlowy outline nodes by applying semantic operations like moving, renaming, creating, deleting, and searching/replacing text within JSON working files.

Instructions

Apply JEWELSTORM semantic operations to a NEXUS working_gem JSON file (PHANTOM GEM working copy). This is the semantic analogue of edit_file for PHANTOM GEM JSON: MOVE_NODE, DELETE_NODE, RENAME_NODE, SET_NOTE, SET_ATTRS, CREATE_NODE, all referencing nodes by jewel_id, plus text-level SEARCH_REPLACE / SEARCH_AND_TAG over name/note fields (substring/whole-word, optional regex, tagging in name and/or note based on matches).

Input Schema

TableJSON Schema
NameRequiredDescriptionDefault
jewel_fileYes
operationsYes
dry_runNo
stop_on_errorNo

Implementation Reference

  • MCP tool registration for 'nexus_transform_jewel' with parameters and description defining the schema and usage.
        name="nexus_transform_jewel",
        description=(
            "Apply JEWELSTORM semantic operations to a NEXUS working_gem JSON file "
            "(PHANTOM GEM working copy). This is the semantic analogue of edit_file "
            "for PHANTOM GEM JSON: MOVE_NODE, DELETE_NODE, RENAME_NODE, SET_NOTE, "
            "SET_ATTRS, CREATE_NODE, all referencing nodes by jewel_id, plus text-level "
            "SEARCH_REPLACE / SEARCH_AND_TAG over name/note fields (substring/whole-word, "
            "optional regex, tagging in name and/or note based on matches)."
        ),
    )
    def nexus_transform_jewel(
        jewel_file: str,
        operations: list[dict[str, Any]],
        dry_run: bool = False,
        stop_on_error: bool = True,
    ) -> dict:
        """JEWELSTORM transform on a NEXUS working_gem JSON file.
    
        This tool wraps nexus_json_tools.transform_jewel via the WorkFlowyClient,
        providing an MCP-friendly interface for JEWELSTORM operations.
    
        Args:
            jewel_file: Absolute path to working_gem JSON (typically a QUILLSTRIKE
                        working stone derived from phantom_gem.json).
            operations: List of operation dicts. Each must include an "op" key and
                        operation-specific fields (e.g., jewel_id, parent_jewel_id,
                        position, etc.).
            dry_run: If True, simulate only (no file write).
            stop_on_error: If True, abort on first error (no write).
    
        Returns:
            Result dict from transform_jewel with success flag, counts, and errors.
        """
        client = get_client()
        return client.nexus_transform_jewel(
            jewel_file=jewel_file,
            operations=operations,
            dry_run=dry_run,
            stop_on_error=stop_on_error,
        )
  • Core handler function implementing all JEWELSTORM operations (MOVE_NODE, DELETE_NODE, RENAME_NODE, SET_NOTE, CREATE_NODE, SEARCH_REPLACE, etc.) on working_gem JSON using jewel_id handles.
    def transform_jewel(
        jewel_file: str,
        operations: List[Dict[str, Any]],
        dry_run: bool = False,
        stop_on_error: bool = True,
    ) -> Dict[str, Any]:
        """Apply JEWELSTORM semantic operations to a NEXUS working_gem JSON file.
    
        This is the semantic analogue of pattern-based edit_file() for PHANTOM GEM
        / working_gem JSON:
        - Operates purely offline (no Workflowy API calls)
        - Works on a local JSON file produced by JEWELSTRIKE (typically phantom_gem.json
          copied to a working stone)
        - Uses jewel_id as the stable identity handle inside JEWELSTORM
        - Never invents real Workflowy IDs: new nodes are written without an "id"
          field so NEXUS/WEAVE can treat them as CREATE operations
    
        Args:
            jewel_file: Path to working_gem JSON file
            operations: List of operation dictionaries, e.g.:
                {
                  "op": "MOVE_NODE",
                  "jewel_id": "J-001",
                  "new_parent_jewel_id": "J-010",
                  "position": "LAST" | "FIRST" | "BEFORE" | "AFTER",
                  "relative_to_jewel_id": "J-999"  # for BEFORE/AFTER
                }
            dry_run: If True, simulate only (no file write)
            stop_on_error: If True, abort on first error (no write)
    
        Returns:
            Dict with success flag, counts, and error details. Example:
            {
              "success": True,
              "applied_count": 3,
              "dry_run": False,
              "nodes_created": 1,
              "nodes_deleted": 0,
              "nodes_moved": 1,
              "nodes_renamed": 1,
              "notes_updated": 0,
              "attrs_updated": 0,
              "errors": []
            }
        """
        import uuid
    
        # ------- Load JSON safely (convert die/SystemExit into structured error) -------
        try:
            data = load_json(jewel_file)
        except SystemExit as e:  # die() inside load_json uses sys.exit
            return {
                "success": False,
                "applied_count": 0,
                "dry_run": dry_run,
                "nodes_created": 0,
                "nodes_deleted": 0,
                "nodes_moved": 0,
                "nodes_renamed": 0,
                "notes_updated": 0,
                "attrs_updated": 0,
                "errors": [
                    {
                        "index": -1,
                        "op": None,
                        "code": "LOAD_ERROR",
                        "message": f"Failed to load JSON from {jewel_file}: {e}",
                    }
                ],
            }
    
        # Determine editable roots list (supports both export-package dict and bare list)
        if isinstance(data, dict) and isinstance(data.get("nodes"), list):
            original_roots = data["nodes"]
        elif isinstance(data, list):
            original_roots = data
        else:
            return {
                "success": False,
                "applied_count": 0,
                "dry_run": dry_run,
                "nodes_created": 0,
                "nodes_deleted": 0,
                "nodes_moved": 0,
                "nodes_renamed": 0,
                "notes_updated": 0,
                "attrs_updated": 0,
                "errors": [
                    {
                        "index": -1,
                        "op": None,
                        "code": "FORMAT_ERROR",
                        "message": "JSON must be a dict with 'nodes' or a bare list of nodes for transform_jewel.",
                    }
                ],
            }
    
        # Work on a deep copy so we never mutate the original unless we succeed
        roots: List[JsonDict] = copy.deepcopy(original_roots)
    
        # ------- Indexing: jewel_id-based identity -------
        by_jewel_id: Dict[str, JsonDict] = {}
        parent_by_jewel_id: Dict[str, Optional[str]] = {}
        existing_ids: Set[str] = set()
    
        def _ensure_children_list(node: JsonDict) -> List[JsonDict]:
            children = node.get("children")
            if not isinstance(children, list):
                children = []
                node["children"] = children
            return children
    
        def _register_node(node: JsonDict, parent_jid: Optional[str]) -> None:
            """Register node and its subtree in jewel_id index.
    
            Identity:
            - Prefer "jewel_id" if present
            - Fallback to "id" for older files (pre-JEWELSTRIKE)
            """
            jewel_id = node.get("jewel_id") or node.get("id")
            if jewel_id:
                if jewel_id in by_jewel_id:
                    raise ValueError(f"Duplicate jewel_id/id {jewel_id!r} in JSON tree")
                by_jewel_id[jewel_id] = node
                parent_by_jewel_id[jewel_id] = parent_jid
                existing_ids.add(jewel_id)
    
            children = node.get("children") or []
            if not isinstance(children, list):
                children = []
                node["children"] = children
            for child in children:
                if isinstance(child, dict):
                    _register_node(child, jewel_id)
    
        for root in roots:
            if isinstance(root, dict):
                _register_node(root, None)
    
        def _new_jewel_id() -> str:
            """Generate a fresh, human-friendly jewel_id that cannot collide."""
            while True:
                candidate = "J-" + uuid.uuid4().hex[:8]
                if candidate not in existing_ids:
                    existing_ids.add(candidate)
                    return candidate
    
        def _resolve_to_jewel_id(identifier: str) -> str:
            """Auto-resolve Workflowy UUID to jewel_id if needed.
            
            If identifier looks like a Workflowy UUID (standard UUID format, not J-NNN),
            search the tree for a node with that id and return its jewel_id.
            
            If identifier is already a jewel_id (J-NNN format), return as-is.
            
            Args:
                identifier: Either a jewel_id (J-NNN) or Workflowy UUID
                
            Returns:
                jewel_id to use for the operation
                
            Raises:
                ValueError: If UUID not found in tree
            """
            import re
            
            # Check if it's a standard UUID format (8-4-4-4-12 hex pattern)
            uuid_pattern = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', re.IGNORECASE)
            
            if uuid_pattern.match(identifier):
                # It's a Workflowy UUID - search for node with this id
                for jewel_id, node in by_jewel_id.items():
                    if node.get("id") == identifier:
                        return jewel_id
                # Not found
                raise ValueError(
                    f"Node with Workflowy UUID '{identifier}' not found in GEM. "
                    "(Auto-resolved from jewel_id parameter - agent can use Workflowy UUIDs directly)"
                )
            else:
                # Already a jewel_id (J-NNN format) or other format - return as-is
                return identifier
    
    
        def _resolve_to_jewel_id(identifier: str) -> str:
            """Auto-resolve Workflowy UUID to jewel_id if needed.
            
            If identifier looks like a Workflowy UUID (standard UUID format, not J-NNN),
            search the tree for a node with that id and return its jewel_id.
            
            If identifier is already a jewel_id (J-NNN format), return as-is.
            
            Args:
                identifier: Either a jewel_id (J-NNN) or Workflowy UUID
                
            Returns:
                jewel_id to use for the operation
                
            Raises:
                ValueError: If UUID not found in tree
            """
            import re
            
            # Check if it's a standard UUID format (8-4-4-4-12 hex pattern)
            uuid_pattern = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', re.IGNORECASE)
            
            if uuid_pattern.match(identifier):
                # It's a Workflowy UUID - search for node with this id
                for jewel_id, node in by_jewel_id.items():
                    if node.get("id") == identifier:
                        return jewel_id
                # Not found
                raise ValueError(
                    f"Node with Workflowy UUID '{identifier}' not found in GEM. "
                    "(Auto-resolved from jewel_id parameter - agent can use Workflowy UUIDs directly)"
                )
            else:
                # Already a jewel_id (J-NNN format) or other format - return as-is
                return identifier
    
    
        def _get_node_and_parent_list(jewel_id: str) -> Tuple[JsonDict, Optional[str], List[JsonDict]]:
            # Auto-resolve Workflowy UUID to jewel_id if needed
            jewel_id = _resolve_to_jewel_id(jewel_id)
            
            node = by_jewel_id.get(jewel_id)
            if node is None:
                raise ValueError(f"Node with jewel_id/id {jewel_id!r} not found")
            parent_jid = parent_by_jewel_id.get(jewel_id)
            if parent_jid is None:
                siblings = roots
            else:
                parent_node = by_jewel_id.get(parent_jid)
                if parent_node is None:
                    raise ValueError(f"Parent with jewel_id/id {parent_jid!r} not found for node {jewel_id!r}")
                siblings = _ensure_children_list(parent_node)
            return node, parent_jid, siblings
    
        def _assert_no_cycle(node_jewel_id: str, new_parent_jewel_id: Optional[str]) -> None:
            cur = new_parent_jewel_id
            while cur is not None:
                if cur == node_jewel_id:
                    raise ValueError(
                        f"Cannot move node {node_jewel_id!r} under its own descendant {new_parent_jewel_id!r}"
                    )
                cur = parent_by_jewel_id.get(cur)
    
        def _remove_subtree_from_index(node: JsonDict) -> None:
            jid = node.get("jewel_id") or node.get("id")
            if jid:
                by_jewel_id.pop(jid, None)
                parent_by_jewel_id.pop(jid, None)
                existing_ids.discard(jid)
            for child in node.get("children") or []:
                if isinstance(child, dict):
                    _remove_subtree_from_index(child)
    
        def _build_subtree_from_spec(spec: Dict[str, Any]) -> JsonDict:
            """Build a new subtree from a CREATE_NODE spec (Pattern A).
    
            Spec keys used:
              - name (required)
              - note (optional)
              - attrs or data (optional, mapped to node['data'])
              - jewel_id (optional; if omitted, auto-generated)
              - children (optional list of nested specs)
              - parent_id (optional, ignored - will be set during insertion)
            """
            if "name" not in spec:
                raise ValueError("CREATE_NODE spec requires 'name' field in node object")
    
            node: JsonDict = {"name": spec["name"]}
    
            if "note" in spec:
                node["note"] = spec["note"]
    
            attrs = spec.get("attrs") or spec.get("data")
            if isinstance(attrs, dict):
                node["data"] = copy.deepcopy(attrs)
    
            # Assign jewel_id (never assign real Workflowy id here)
            jewel_id = spec.get("jewel_id")
            if jewel_id is None:
                jewel_id = _new_jewel_id()
            else:
                if jewel_id in existing_ids:
                    raise ValueError(f"CREATE_NODE requested duplicate jewel_id {jewel_id!r}")
                existing_ids.add(jewel_id)
            node["jewel_id"] = jewel_id
    
            # Recursively build children
            children_specs = spec.get("children") or []
            children_nodes: List[JsonDict] = []
            for child_spec in children_specs:
                if isinstance(child_spec, dict):
                    child_node = _build_subtree_from_spec(child_spec)
                    children_nodes.append(child_node)
            node["children"] = children_nodes
            return node
    
        def _classify_children_for_destructive_ops(node: JsonDict) -> Dict[str, Any]:
            """Classify immediate children for destructive JEWELSTORM ops.
    
            Returns a dict with:
                children: list of child dict nodes
                children_status: str
                truncated: bool
                has_loaded_ether_children: bool  # child has a real Workflowy id
                has_new_children: bool           # child has no id yet (JEWEL-only)
                category: "EMPTY" | "JEWEL_ONLY" | "ETHER_ONLY" | "MIXED"
            """
            children_raw = node.get("children") or []
            children = [c for c in children_raw if isinstance(c, dict)]
            children_status = node.get("children_status") or "complete"
            truncated = children_status in {"truncated_by_depth", "truncated_by_count"}
            has_loaded_ether_children = any(c.get("id") is not None for c in children)
            has_new_children = any(c.get("id") is None for c in children)
    
            if not children:
                category = "EMPTY"
            elif (not has_loaded_ether_children) and has_new_children and not truncated:
                category = "JEWEL_ONLY"
            elif has_loaded_ether_children and not has_new_children:
                category = "ETHER_ONLY"
            else:
                category = "MIXED"
    
            return {
                "children": children,
                "children_status": children_status,
                "truncated": truncated,
                "has_loaded_ether_children": has_loaded_ether_children,
                "has_new_children": has_new_children,
                "category": category,
            }
    
        # ------- Apply operations -------
        applied_count = 0
        nodes_created = 0
        nodes_deleted = 0
        nodes_moved = 0
        nodes_renamed = 0
        notes_updated = 0
        attrs_updated = 0
        errors: List[Dict[str, Any]] = []
    
        for idx, op in enumerate(operations or []):
            op_type_raw = op.get("op") or op.get("operation")
            if not op_type_raw:
                err = {
                    "index": idx,
                    "op": op,
                    "code": "MISSING_OP",
                    "message": "Operation missing 'op' field",
                }
                errors.append(err)
                if stop_on_error:
                    break
                continue
    
            op_type = str(op_type_raw).upper()
    
            try:
                # MOVE_NODE
                if op_type == "MOVE_NODE":
                    src_jid = op.get("jewel_id")
                    if not src_jid:
                        raise ValueError("MOVE_NODE requires 'jewel_id'")
    
                    position = str(op.get("position", "LAST")).upper()
                    # Normalize synonyms
                    if position == "BOTTOM":
                        position = "LAST"
                    if position == "TOP":
                        position = "FIRST"
                    rel_jid = op.get("relative_to_jewel_id")
                    if rel_jid:
                        rel_jid = _resolve_to_jewel_id(rel_jid)
                    new_parent_jid = op.get("new_parent_jewel_id")
    
                    if position in {"BEFORE", "AFTER"}:
                        if not rel_jid:
                            raise ValueError(
                                "MOVE_NODE with position BEFORE/AFTER requires 'relative_to_jewel_id'"
                            )
                        # Target parent/siblings come from relative node
                        _, rel_parent_jid, rel_siblings = _get_node_and_parent_list(rel_jid)
                        target_parent_jid = rel_parent_jid
                        target_list = rel_siblings
                    else:
                        # FIRST/LAST under explicit parent (or root when parent None)
                        target_parent_jid = new_parent_jid
                        if new_parent_jid is None:
                            target_list = roots
                        else:
                            parent_node = by_jewel_id.get(new_parent_jid)
                            if parent_node is None:
                                raise ValueError(f"New parent jewel_id/id {new_parent_jid!r} not found")
                            target_list = _ensure_children_list(parent_node)
    
                    # Cycle check
                    _assert_no_cycle(src_jid, target_parent_jid)
    
                    node, old_parent_jid, old_siblings = _get_node_and_parent_list(src_jid)
    
                    # Remove from old siblings
                    if node in old_siblings:
                        old_siblings.remove(node)
    
                    # Insert into new location
                    if position == "FIRST":
                        target_list.insert(0, node)
                    elif position == "LAST":
                        target_list.append(node)
                    elif position in {"BEFORE", "AFTER"}:
                        rel_node, _, _ = _get_node_and_parent_list(rel_jid)  # type: ignore[arg-type]
                        try:
                            rel_index = target_list.index(rel_node)
                        except ValueError as e:
                            raise ValueError(
                                f"Relative node {rel_jid!r} not found under chosen parent for MOVE_NODE"
                            ) from e
                        insert_index = rel_index if position == "BEFORE" else rel_index + 1
                        target_list.insert(insert_index, node)
                    else:
                        raise ValueError(f"Unsupported MOVE_NODE position {position!r}")
    
                    # Update parent mapping
                    parent_by_jewel_id[src_jid] = target_parent_jid
                    nodes_moved += 1
    
                # DELETE_NODE
                elif op_type == "DELETE_NODE":
                    jid = op.get("jewel_id")
                    if not jid:
                        raise ValueError("DELETE_NODE requires 'jewel_id'")
    
                    delete_from_ether = bool(op.get("delete_from_ether"))
                    mode_raw = op.get("mode")
                    mode = str(mode_raw).upper() if mode_raw is not None else "SMART"
    
                    node, parent_jid, siblings = _get_node_and_parent_list(jid)
                    info = _classify_children_for_destructive_ops(node)
                    children = info["children"]
                    category = info["category"]
    
                    # Legacy strict mode: preserve original FAIL_IF_HAS_CHILDREN semantics
                    if mode == "FAIL_IF_HAS_CHILDREN":
                        if children:
                            raise ValueError(
                                f"DELETE_NODE {jid!r} refused: node has children and mode=FAIL_IF_HAS_CHILDREN"
                            )
                    else:
                        # SMART semantics (default when mode not provided)
                        if category == "MIXED":
                            raise ValueError(
                                f"DELETE_NODE {jid!r} refused: mixed new/ETHER/truncated children not supported; "
                                "delete or move children individually first"
                            )
                        if category == "ETHER_ONLY" and not delete_from_ether:
                            raise ValueError(
                                f"DELETE_NODE {jid!r} refused: node has ETHER-backed children; "
                                "set delete_from_ether=True to delete subtree in Workflowy"
                            )
                        # EMPTY and JEWEL_ONLY are always allowed.
    
                    # Remove from siblings and index
                    if node in siblings:
                        siblings.remove(node)
                    _remove_subtree_from_index(node)
                    nodes_deleted += 1
    
                # DELETE_ALL_CHILDREN
                elif op_type == "DELETE_ALL_CHILDREN":
                    jid = op.get("jewel_id")
                    if not jid:
                        raise ValueError("DELETE_ALL_CHILDREN requires 'jewel_id'")
    
                    delete_from_ether = bool(op.get("delete_from_ether"))
                    mode_raw = op.get("mode")
                    mode = str(mode_raw).upper() if mode_raw is not None else "SMART"
    
                    node = by_jewel_id.get(jid)
                    if node is None:
                        raise ValueError(f"Node with jewel_id/id {jid!r} not found")
    
                    info = _classify_children_for_destructive_ops(node)
                    children = info["children"]
                    category = info["category"]
                    truncated = info["truncated"]
    
                    # Nothing to do
                    if not children and not truncated:
                        pass
                    else:
                        # For DELETE_ALL_CHILDREN we do not support truncated children sets in v1
                        if truncated:
                            raise ValueError(
                                f"DELETE_ALL_CHILDREN {jid!r} refused: children set is truncated; "
                                "re-GLIMPSE with full children or delete the node instead"
                            )
    
                        if mode == "FAIL_IF_HAS_CHILDREN":
                            if children:
                                raise ValueError(
                                    f"DELETE_ALL_CHILDREN {jid!r} refused: node has children and "
                                    "mode=FAIL_IF_HAS_CHILDREN"
                                )
                        else:
                            if category == "MIXED":
                                raise ValueError(
                                    f"DELETE_ALL_CHILDREN {jid!r} refused: mixed new/ETHER children not supported; "
                                    "delete or move children individually first"
                                )
                            if category == "ETHER_ONLY" and not delete_from_ether:
                                raise ValueError(
                                    f"DELETE_ALL_CHILDREN {jid!r} refused: children are ETHER-backed; "
                                    "set delete_from_ether=True to delete them in Workflowy"
                                )
                            # EMPTY and JEWEL_ONLY are always allowed here.
    
                        # Allowed path: remove all current children from index and node
                        for child in list(children):
                            _remove_subtree_from_index(child)
                        node["children"] = []
                        nodes_deleted += len(children)
    
                # RENAME_NODE
                elif op_type == "RENAME_NODE":
                    jid = op.get("jewel_id")
                    if not jid:
                        raise ValueError("RENAME_NODE requires 'jewel_id'")
                    new_name = op.get("new_name")
                    if new_name is None:
                        raise ValueError("RENAME_NODE requires 'new_name'")
    
                    node = by_jewel_id.get(jid)
                    if node is None:
                        raise ValueError(f"Node with jewel_id/id {jid!r} not found")
                    node["name"] = new_name
                    nodes_renamed += 1
    
                # SET_NOTE
                elif op_type == "SET_NOTE":
                    jid = op.get("jewel_id")
                    if not jid:
                        raise ValueError("SET_NOTE requires 'jewel_id'")
                    new_note = op.get("new_note")
                    # Allow empty string / None to clear
    
                    node = by_jewel_id.get(jid)
                    if node is None:
                        raise ValueError(f"Node with jewel_id/id {jid!r} not found")
                    node["note"] = new_note
                    notes_updated += 1
    
                # SET_ATTRS
                elif op_type == "SET_ATTRS":
                    jid = op.get("jewel_id")
                    if not jid:
                        raise ValueError("SET_ATTRS requires 'jewel_id'")
                    attrs = op.get("attrs") or {}
                    if not isinstance(attrs, dict):
                        raise ValueError("SET_ATTRS 'attrs' must be a dict")
    
                    node = by_jewel_id.get(jid)
                    if node is None:
                        raise ValueError(f"Node with jewel_id/id {jid!r} not found")
    
                    data = node.get("data")
                    if not isinstance(data, dict):
                        data = {}
    
                    allowed_keys = {"completed", "layoutMode", "priority", "tags"}
                    for key, value in attrs.items():
                        if key not in allowed_keys:
                            raise ValueError(f"Unsupported attr key {key!r} in SET_ATTRS")
                        if value is None:
                            data.pop(key, None)
                        else:
                            data[key] = value
    
                    if data:
                        node["data"] = data
                    elif "data" in node:
                        del node["data"]
    
                    attrs_updated += 1
    
                # CREATE_NODE (Pattern A with optional jewel_id)
                elif op_type == "CREATE_NODE":
                    parent_jid = op.get("parent_jewel_id")
                    position = str(op.get("position", "LAST")).upper()
                    # Normalize synonyms
                    if position == "BOTTOM":
                        position = "LAST"
                    if position == "TOP":
                        position = "FIRST"
                    rel_jid = op.get("relative_to_jewel_id")
                    if rel_jid:
                        rel_jid = _resolve_to_jewel_id(rel_jid)
    
                    if parent_jid is None:
                        parent_children = roots
                        target_parent_jid = None
                    else:
                        parent_node = by_jewel_id.get(parent_jid)
                        if parent_node is None:
                            raise ValueError(f"CREATE_NODE parent_jewel_id {parent_jid!r} not found")
                        parent_children = _ensure_children_list(parent_node)
                        target_parent_jid = parent_jid
    
                    # Build node spec for subtree
                    # Two formats supported:
                    # 1. Compact: node fields at operation level (name, note, children, etc.)
                    # 2. Wrapped: node fields inside "node" key
                    if "node" in op:
                        spec = op["node"]
                    else:
                        # Remove op-specific keys to get node spec
                        spec = {k: v for k, v in op.items() if k not in {
                            "op",
                            "operation",
                            "parent_jewel_id",
                            "position",
                            "relative_to_jewel_id",
                        }}
    
                    new_node = _build_subtree_from_spec(spec)
    
                    # Register subtree in indexes with correct parent mapping
                    _register_node(new_node, target_parent_jid)
    
                    # Insert relative to siblings
                    if position == "FIRST":
                        parent_children.insert(0, new_node)
                    elif position == "LAST":
                        parent_children.append(new_node)
                    elif position in {"BEFORE", "AFTER"}:
                        if not rel_jid:
                            raise ValueError(
                                "CREATE_NODE with position BEFORE/AFTER requires 'relative_to_jewel_id'"
                            )
                        rel_node, _, rel_siblings = _get_node_and_parent_list(rel_jid)  # type: ignore[arg-type]
                        # Force siblings to be the same list as parent_children
                        if rel_siblings is not parent_children:
                            raise ValueError(
                                "CREATE_NODE BEFORE/AFTER relative_to_jewel_id must share the same parent"
                            )
                        try:
                            rel_index = parent_children.index(rel_node)
                        except ValueError as e:
                            raise ValueError(
                                f"Relative node {rel_jid!r} not found under chosen parent for CREATE_NODE"
                            ) from e
                        insert_index = rel_index if position == "BEFORE" else rel_index + 1
                        parent_children.insert(insert_index, new_node)
                    else:
                        raise ValueError(f"Unsupported CREATE_NODE position {position!r}")
    
                    nodes_created += 1
    
                # SET_ATTRS_BY_PATH (path-based attribute update, used for JEWEL UUID injection)
                elif op_type == "SET_ATTRS_BY_PATH":
                    path = op.get("path")
                    attrs = op.get("attrs") or {}
                    if not isinstance(path, list) or not path:
                        raise ValueError("SET_ATTRS_BY_PATH requires non-empty 'path' list")
                    if not isinstance(attrs, dict):
                        raise ValueError("SET_ATTRS_BY_PATH 'attrs' must be a dict")
    
                    # Navigate by index path from roots
                    current_list: List[JsonDict] = roots
                    target_node: Optional[JsonDict] = None
                    for level, idx in enumerate(path):
                        if not isinstance(idx, int):
                            raise ValueError(
                                f"SET_ATTRS_BY_PATH path index at position {level} must be int, got {type(idx).__name__}"
                            )
                        if idx < 0 or idx >= len(current_list):
                            raise ValueError(
                                f"SET_ATTRS_BY_PATH path index {idx} out of range at position {level}"
                            )
                        target_node = current_list[idx]
                        if level < len(path) - 1:
                            children = target_node.get("children")
                            if not isinstance(children, list):
                                raise ValueError(
                                    f"SET_ATTRS_BY_PATH path descends into non-list children at position {level}"
                                )
                            current_list = children
    
                    if target_node is None:
                        raise ValueError("SET_ATTRS_BY_PATH could not resolve target node from path")
    
                    for key, value in attrs.items():
                        if key == "id":
                            if value is None:
                                target_node.pop("id", None)
                            else:
                                if not isinstance(value, str):
                                    raise ValueError(
                                        "SET_ATTRS_BY_PATH 'id' value must be a string when not None"
                                    )
                                target_node["id"] = value
                        else:
                            raise ValueError(
                                f"Unsupported attr key {key!r} in SET_ATTRS_BY_PATH (only 'id' is currently allowed)"
                            )
    
                    attrs_updated += 1
    
                else:
                    raise ValueError(f"Unknown operation type {op_type!r}")
    
                applied_count += 1
    
            except Exception as e:  # noqa: BLE001
                errors.append(
                    {
                        "index": idx,
                        "op": op,
                        "code": "OP_ERROR",
                        "message": str(e),
                    }
                )
                if stop_on_error:
                    break
    
        # ------- Persist changes (if not dry-run and no stop-on-error failure) -------
        if not dry_run and (not stop_on_error or not errors):
            # Attach modified roots back to original structure
            if isinstance(data, dict) and isinstance(data.get("nodes"), list):
                data["nodes"] = roots
                # Recalculate counts for readability in the JEWEL working_gem,
                # but DO NOT touch children_status here. Truncation semantics belong
                # to the NEXUS/TERRAIN layer (shimmering/enchanted terrain).
                wrapper = {"nodes": roots}
                recalc_all_counts_gem(wrapper)
            elif isinstance(data, list):
                data = roots  # type: ignore[assignment]
                wrapper = {"nodes": roots}
                recalc_all_counts_gem(wrapper)
    
            save_json(jewel_file, data)  # type: ignore[arg-type]
    
        return {
            "success": len(errors) == 0,
            "applied_count": applied_count,
            "dry_run": dry_run,
            "nodes_created": nodes_created,
            "nodes_deleted": nodes_deleted,
            "nodes_moved": nodes_moved,
            "nodes_renamed": nodes_renamed,
            "notes_updated": notes_updated,
            "attrs_updated": attrs_updated,
            "errors": errors,
        }

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/daniel347x/workflowy-mcp-fixed'

If you have feedback or need assistance with the MCP directory API, please join our Discord server