Create a new documentation file in the output directory.
(
arguments: Dict[str, Any],
store: SessionStore,
)
| 71 | |
| 72 | |
| 73 | async def handle_write_doc_file( |
| 74 | arguments: Dict[str, Any], |
| 75 | store: SessionStore, |
| 76 | ) -> str: |
| 77 | """Create a new documentation file in the output directory.""" |
| 78 | session_id = arguments["session_id"] |
| 79 | session = store.get(session_id) |
| 80 | if session is None: |
| 81 | return json.dumps({"error": f"Session {session_id} not found or expired."}) |
| 82 | |
| 83 | filename = arguments["filename"] |
| 84 | doc_path = _safe_doc_path(session, filename) |
| 85 | if doc_path is None: |
| 86 | return json.dumps({"error": "Filename escapes output directory."}) |
| 87 | |
| 88 | content = arguments["content"] |
| 89 | |
| 90 | await asyncio.to_thread(_ensure_parent_dirs, doc_path) |
| 91 | |
| 92 | if await asyncio.to_thread(doc_path.exists): |
| 93 | return json.dumps({ |
| 94 | "error": f"File already exists: {filename}. Use edit_doc_file to modify it." |
| 95 | }) |
| 96 | |
| 97 | await asyncio.to_thread(doc_path.write_text, content, "utf-8") |
| 98 | session.docs_written += 1 |
| 99 | |
| 100 | # Mermaid validation |
| 101 | mermaid_result = await _validate_mermaid(str(doc_path), filename) |
| 102 | |
| 103 | result = { |
| 104 | "status": "created", |
| 105 | "path": str(doc_path), |
| 106 | "filename": filename, |
| 107 | "lines": content.count("\n") + 1, |
| 108 | "mermaid_validation": mermaid_result, |
| 109 | } |
| 110 | return json.dumps(result, indent=2, ensure_ascii=False) |
| 111 | |
| 112 | |
| 113 | async def handle_edit_doc_file( |