Write the leaf-first processing order to a workspace file and return its path.
(
arguments: Dict[str, Any],
store: SessionStore,
)
| 107 | |
| 108 | |
| 109 | def handle_get_processing_order( |
| 110 | arguments: Dict[str, Any], |
| 111 | store: SessionStore, |
| 112 | ) -> str: |
| 113 | """Write the leaf-first processing order to a workspace file and return its path.""" |
| 114 | session_id = arguments["session_id"] |
| 115 | session = store.get(session_id) |
| 116 | if session is None: |
| 117 | return json.dumps({"error": f"Session {session_id} not found or expired."}) |
| 118 | |
| 119 | # Try session cache first, then disk |
| 120 | module_tree = session.module_tree |
| 121 | if not module_tree: |
| 122 | tree_path = os.path.join(session.output_dir, MODULE_TREE_FILENAME) |
| 123 | if os.path.exists(tree_path): |
| 124 | with open(tree_path, encoding="utf-8") as f: |
| 125 | module_tree = json.load(f) |
| 126 | session.module_tree = module_tree |
| 127 | else: |
| 128 | return json.dumps({ |
| 129 | "error": "Module tree not found. Call save_module_tree first." |
| 130 | }) |
| 131 | |
| 132 | order = _get_processing_order(module_tree) |
| 133 | |
| 134 | # Write to workspace file |
| 135 | order_file = None |
| 136 | if session.workspace is not None: |
| 137 | order_path = session.workspace.write_json("processing_order.json", order) |
| 138 | order_file = str(order_path) |
| 139 | |
| 140 | result = { |
| 141 | "session_id": session_id, |
| 142 | "module_count": len(module_tree), |
| 143 | "processing_order_file": order_file, |
| 144 | "hint": "Read the processing_order.json file for the full leaf-first order.", |
| 145 | } |
| 146 | return json.dumps(result, indent=2, ensure_ascii=False) |
nothing calls this directly
no test coverage detected