WorkFlowy MCP Server

Instructions

Implementation Reference

• src/workflowy_mcp/server.py:1567-1607 (registration)

  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,
      )

• src/workflowy_mcp/nexus_json_tools.py:79-837 (handler)

  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,
      }