nexus_glimpse
Extract expanded WorkFlowy nodes via WebSocket to generate terrain and gem JSON files without API calls. Control data granularity by expanding nodes locally.
Instructions
GLIMPSE → TERRAIN + PHANTOM GEM (zero API calls). Captures what you've expanded in Workflowy via WebSocket GLIMPSE and creates both coarse_terrain.json and phantom_gem.json from that single local extraction. No Workflowy API calls, instant, you control granularity by expanding nodes.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| nexus_tag | Yes | ||
| workflowy_root_id | Yes | ||
| reset_if_exists | No | ||
| mode | No | full |
Implementation Reference
- src/workflowy_mcp/server.py:1307-1332 (registration)MCP tool registration for 'nexus_glimpse'. Delegates to WorkFlowyClient.nexus_glimpse method.
name="nexus_glimpse", description=( "GLIMPSE → TERRAIN + PHANTOM GEM (zero API calls). " "Captures what you've expanded in Workflowy via WebSocket GLIMPSE and creates both " "coarse_terrain.json and phantom_gem.json from that single local extraction. " "No Workflowy API calls, instant, you control granularity by expanding nodes." ), ) async def nexus_glimpse( nexus_tag: str, workflowy_root_id: str, reset_if_exists: bool = False, mode: str = "full", ) -> dict[str, Any]: """GLIMPSE-based NEXUS initialization.""" client = get_client() ws_conn, ws_queue = get_ws_connection() # Get WebSocket connection return await client.nexus_glimpse( nexus_tag=nexus_tag, workflowy_root_id=workflowy_root_id, reset_if_exists=reset_if_exists, mode=mode, _ws_connection=ws_conn, _ws_queue=ws_queue, ) - Core handler implementation: performs WebSocket GLIMPSE, optional API scry, merges via nexus_helper, writes NEXUS files (coarse_terrain, phantom_gem, shimmering_terrain).
async def nexus_glimpse( self, nexus_tag: str, workflowy_root_id: str, reset_if_exists: bool = False, mode: str = "full", _ws_connection=None, _ws_queue=None, ) -> dict[str, Any]: """GLIMPSE → TERRAIN + PHANTOM GEM (zero API calls).""" import shutil if mode not in ("full", "coarse_terrain_only"): raise NetworkError(f"Invalid mode '{mode}'") base_dir = Path( r"E:\__daniel347x\__Obsidian\__Inking into Mind\--TypingMind\Projects - All\Projects - Individual\TODO\temp\nexus_runs" ) base_dir.mkdir(parents=True, exist_ok=True) # Determine run_dir existing_dir = None if not reset_if_exists: try: existing = self._get_nexus_dir(nexus_tag) existing_dir = Path(existing) except NetworkError: pass if reset_if_exists: suffix = f"__{nexus_tag}" for child in base_dir.iterdir(): if not child.is_dir(): continue if child.name == nexus_tag or child.name.endswith(suffix): shutil.rmtree(child, ignore_errors=True) existing_dir = None if existing_dir: run_dir = existing_dir else: timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") run_dir = base_dir / f"{timestamp}__{nexus_tag}" run_dir.mkdir(parents=True, exist_ok=False) coarse_path = run_dir / "coarse_terrain.json" phantom_gem_path = run_dir / "phantom_gem.json" shimmering_path = run_dir / "shimmering_terrain.json" logger = _ClientLogger() # WebSocket GLIMPSE logger.info(f"🔌 NEXUS GLIMPSE: WebSocket for {workflowy_root_id[:8]}...") ws_glimpse = await self.workflowy_glimpse( workflowy_root_id, _ws_connection=_ws_connection, _ws_queue=_ws_queue ) if not ws_glimpse.get("success"): raise NetworkError(f"GLIMPSE failed: {ws_glimpse.get('error')}") # API GLIMPSE for complete structure logger.info("📡 API fetch...") try: api_glimpse = await self.workflowy_scry( node_id=workflowy_root_id, use_efficient_traversal=False, depth=None, size_limit=50000 ) except Exception as e: logger.warning(f"⚠️ API failed: {e}") children = ws_glimpse.get("children", []) mark_all_nodes_as_potentially_truncated(children) api_glimpse = None # Merge logger.info("🔀 Merging...") children, root_children_status, api_merge = merge_glimpses_for_terrain( workflowy_root_id=workflowy_root_id, ws_glimpse=ws_glimpse, api_glimpse=api_glimpse, ) original_ids = build_original_ids_seen_from_glimpses( workflowy_root_id=workflowy_root_id, ws_glimpse=ws_glimpse, api_glimpse=api_glimpse, ) terrain_data = build_terrain_export_from_glimpse( workflowy_root_id=workflowy_root_id, ws_glimpse=ws_glimpse, children=children, root_children_status=root_children_status, original_ids_seen=original_ids, ) # Attach preview terrain_preview = None try: terrain_preview = self._annotate_preview_ids_and_build_tree( terrain_data.get("nodes") or [], "CT" ) except Exception: pass if terrain_preview: terrain_data = { "export_timestamp": terrain_data.get("export_timestamp"), "export_root_children_status": terrain_data.get("export_root_children_status"), "__preview_tree__": terrain_preview, "export_root_id": terrain_data.get("export_root_id"), "export_root_name": terrain_data.get("export_root_name"), "nodes": terrain_data.get("nodes"), "original_ids_seen": terrain_data.get("original_ids_seen"), } # Write coarse_terrain with open(coarse_path, "w", encoding="utf-8") as f: json.dump(terrain_data, f, ensure_ascii=False, indent=2) result = { "success": True, "nexus_tag": nexus_tag, "coarse_terrain": str(coarse_path), "node_count": ws_glimpse.get("node_count", 0), "depth": ws_glimpse.get("depth", 0), "_source": "glimpse_merged", "_api_merge_performed": api_merge, "mode": mode, } if mode == "full": # Write phantom_gem and shimmering with open(phantom_gem_path, "w", encoding="utf-8") as f: json.dump(terrain_data, f, ensure_ascii=False, indent=2) with open(shimmering_path, "w", encoding="utf-8") as f: json.dump(terrain_data, f, ensure_ascii=False, indent=2) result["phantom_gem"] = str(phantom_gem_path) result["shimmering_terrain"] = str(shimmering_path) _log_glimpse_to_file("glimpse", workflowy_root_id, ws_glimpse) return result - merge_glimpses_for_terrain: Merges WebSocket GLIMPSE (UI-expanded view) with optional API FULL to detect hidden/collapsed children, sets has_hidden_children and children_status per node.
def merge_glimpses_for_terrain( workflowy_root_id: str, ws_glimpse: JsonDict, api_glimpse: Optional[JsonDict] = None, ) -> Tuple[List[JsonDict], str, bool]: """Merge WebSocket + API GLIMPSE results to annotate hidden children. This is the core logic that NEXUS uses to produce safe TERRAIN data from a WebSocket GLIMPSE (what Dan has expanded in the UI) and an optional API GLIMPSE FULL (the complete tree regardless of expansion state). When API data is available and successful: - Each WebSocket node gets ``has_hidden_children`` and ``children_status`` set by comparing its child count to the corresponding API node. - The root's children_status is computed similarly by comparing the top-level child counts. When API data is unavailable or failed: - All nodes are marked via :func:`mark_all_nodes_as_potentially_truncated`. - ``root_children_status`` is set to ``"truncated_by_expansion"``. Args: workflowy_root_id: UUID of the Workflowy root being glimpsed. ws_glimpse: Result dict from the WebSocket GLIMPSE (Dan's UI view). api_glimpse: Optional result dict from ``workflowy_scry`` / GLIMPSE FULL. If ``None`` or unsuccessful, the conservative fallback is used. Returns: (children, root_children_status, api_merge_performed) - children: List of child nodes to be used under ``"nodes"`` in the TERRAIN wrapper (mutated in-place from ``ws_glimpse['children']``). - root_children_status: ``"complete"`` or ``"truncated_by_expansion"`` describing the root's immediate children. - api_merge_performed: ``True`` if we actually compared against API structure; ``False`` if we fell back to WebSocket-only semantics. """ # Default values for the root root_children_status = "complete" api_merge_performed = False ws_children: List[JsonDict] = ws_glimpse.get("children", []) or [] # If we have a successful API GLIMPSE, attempt the full merge. if api_glimpse and api_glimpse.get("success"): api_children = api_glimpse.get("children", []) or [] api_nodes_by_id = _index_nodes_by_id(api_children) api_merge_performed = True nodes_with_hidden = 0 def _merge_flags(nodes: List[JsonDict]) -> None: nonlocal nodes_with_hidden for ws_node in nodes: if not isinstance(ws_node, dict): continue nid = ws_node.get("id") if not nid: # New node (no UUID) – cannot have hidden children in ETHER ws_node["has_hidden_children"] = False ws_node["children_status"] = "complete" children = ws_node.get("children", []) or [] if children: _merge_flags(children) continue api_node = api_nodes_by_id.get(nid) if not api_node: # No API counterpart – treat as complete but logically # this should be rare (GLIMPSE should be subset of SCRY). ws_node["has_hidden_children"] = False ws_node["children_status"] = "complete" children = ws_node.get("children", []) or [] if children: _merge_flags(children) continue ws_count = len(ws_node.get("children", []) or []) api_count = len(api_node.get("children", []) or []) if api_count > ws_count: ws_node["has_hidden_children"] = True ws_node["children_status"] = "truncated_by_expansion" nodes_with_hidden += 1 else: ws_node["has_hidden_children"] = False ws_node["children_status"] = "complete" children = ws_node.get("children", []) or [] if children: _merge_flags(children) _merge_flags(ws_children) # Root-level children_status: compare top-level children counts. api_root_children = api_glimpse.get("children", []) or [] ws_root_children_count = len(ws_children) api_root_children_count = len(api_root_children) if api_root_children_count > ws_root_children_count: root_children_status = "truncated_by_expansion" else: root_children_status = "complete" return ws_children, root_children_status, api_merge_performed # No usable API GLIMPSE – conservative fallback for whole tree. mark_all_nodes_as_potentially_truncated(ws_children) root_children_status = "truncated_by_expansion" api_merge_performed = False return ws_children, root_children_status, api_merge_performed - mark_all_nodes_as_potentially_truncated: Fallback when API unavailable - marks entire tree as potentially truncated to prevent accidental deletions in WEAVE.
def mark_all_nodes_as_potentially_truncated(nodes: List[JsonDict]) -> None: """Mark ALL nodes as potentially having hidden children (UNSAFE fallback). This is the conservative fallback used when an API SCRY/GLIMPSE FULL is not available to validate true child counts. It ensures that downstream WEAVE logic will treat every node as if it might have collapsed/hidden children, preventing accidental deletions. Semantics applied to each node in-place: - ``has_hidden_children = True`` - ``children_status = "truncated_by_expansion"`` Args: nodes: Node tree to mark (modified in-place). """ for node in nodes: node["has_hidden_children"] = True node["children_status"] = "truncated_by_expansion" grandchildren = node.get("children", []) or [] if grandchildren: mark_all_nodes_as_potentially_truncated(grandchildren) - build_terrain_export_from_glimpse: Wraps merged children into canonical TERRAIN JSON format for NEXUS pipeline.
def build_terrain_export_from_glimpse( workflowy_root_id: str, ws_glimpse: JsonDict, children: List[JsonDict], root_children_status: str, original_ids_seen: Set[str], ) -> JsonDict: """Construct a TERRAIN-style export wrapper from GLIMPSE data. This mirrors the structure produced by ``bulk_export_to_file`` and used throughout the NEXUS pipeline (SCRY / GLIMPSE / IGNITE / ANCHOR). Args: workflowy_root_id: UUID of the Workflowy root being exported. ws_glimpse: WebSocket GLIMPSE dict (used for root name). children: List of child nodes (typically merged via :func:`merge_glimpses_for_terrain`). root_children_status: Children status for the export root, e.g. ``"complete"`` or ``"truncated_by_expansion"``. original_ids_seen: Set of all node IDs known at export time. Returns: A dict suitable for JSON dumping as a TERRAIN export. """ root = ws_glimpse.get("root") or {} export_root_name = root.get("name") or "Root" terrain_data: JsonDict = { "export_root_id": workflowy_root_id, "export_root_name": export_root_name, "export_timestamp": None, # GLIMPSE path does not track timestamps "export_root_children_status": root_children_status, "original_ids_seen": sorted(original_ids_seen), "nodes": children, } return terrain_data