Edit an existing documentation file (str_replace, insert, or undo).
(
arguments: Dict[str, Any],
store: SessionStore,
)
| 111 | return json.dumps(result, indent=2, ensure_ascii=False) |
| 112 | |
| 113 | |
| 114 | async def handle_edit_doc_file( |
| 115 | arguments: Dict[str, Any], |
| 116 | store: SessionStore, |
| 117 | ) -> str: |
| 118 | """Edit an existing documentation file (str_replace, insert, or undo).""" |
| 119 | session_id = arguments["session_id"] |
| 120 | session = store.get(session_id) |
| 121 | if session is None: |
| 122 | return json.dumps({"error": f"Session {session_id} not found or expired."}) |
| 123 | |
| 124 | filename = arguments["filename"] |
| 125 | doc_path = _safe_doc_path(session, filename) |
| 126 | if doc_path is None: |
| 127 | return json.dumps({"error": "Filename escapes output directory."}) |
| 128 | |
| 129 | command = arguments["command"] |
| 130 | |
| 131 | if command == "undo": |
| 132 | # Undo via registry history |
| 133 | history = session.registry.get("file_history", {}) |
| 134 | if isinstance(history, str): |
| 135 | history = json.loads(history) |
| 136 | path_history = history.get(str(doc_path), []) |
| 137 | if not path_history: |
| 138 | return json.dumps({"error": f"No edit history found for {filename}."}) |
| 139 | old_content = path_history.pop() |
| 140 | history[str(doc_path)] = path_history |
| 141 | session.registry["file_history"] = history |
| 142 | await asyncio.to_thread(doc_path.write_text, old_content, "utf-8") |
| 143 | |
| 144 | # Validate Mermaid after undo |
| 145 | mermaid_result = await _validate_mermaid(str(doc_path), filename) |
| 146 | return json.dumps( |
| 147 | { |
| 148 | "status": "undone", |
| 149 | "filename": filename, |
| 150 | "mermaid_validation": mermaid_result, |
| 151 | }, |
| 152 | ensure_ascii=False, |
| 153 | ) |
| 154 | |
| 155 | if not await asyncio.to_thread(doc_path.exists): |
| 156 | return json.dumps( |
| 157 | {"error": f"File not found: {filename}. Use write_doc_file to create it."} |
| 158 | ) |
| 159 | |
| 160 | current_content = await asyncio.to_thread(doc_path.read_text, "utf-8") |
| 161 | |
| 162 | if command == "str_replace": |
| 163 | old_str = arguments.get("old_str") |
| 164 | new_str = arguments.get("new_str", "") |
| 165 | if old_str is None: |
| 166 | return json.dumps({"error": "old_str is required for str_replace."}) |
| 167 | |
| 168 | occurrences = current_content.count(old_str) |
| 169 | if occurrences == 0: |
| 170 | return json.dumps({"error": f"old_str not found in {filename}."}) |