Manages the on-disk workspace for a single MCP session.
| 54 | |
| 55 | |
| 56 | class SessionWorkspace: |
| 57 | """Manages the on-disk workspace for a single MCP session.""" |
| 58 | |
| 59 | def __init__(self, repo_path: Path, session_id: str) -> None: |
| 60 | self.root = repo_path / _WORKSPACE_REL / session_id |
| 61 | self.root.mkdir(parents=True, exist_ok=True) |
| 62 | (self.root / "sources").mkdir(exist_ok=True) |
| 63 | logger.debug("Workspace created at %s", self.root) |
| 64 | |
| 65 | # -- writers ---------------------------------------------------------- |
| 66 | |
| 67 | def write_json(self, name: str, data: Any) -> Path: |
| 68 | """Write *data* as pretty-printed JSON and return the file path.""" |
| 69 | p = self.root / name |
| 70 | p.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") |
| 71 | return p |
| 72 | |
| 73 | def write_component_source( |
| 74 | self, |
| 75 | component_id: str, |
| 76 | source: str, |
| 77 | language: str = "", |
| 78 | ) -> Path: |
| 79 | """Write a single component's source code to the ``sources/`` dir.""" |
| 80 | p = self.root / "sources" / _safe_filename(component_id) |
| 81 | header = f"// Component: {component_id}\n// Language: {language}\n" |
| 82 | p.write_text(header + source, encoding="utf-8") |
| 83 | return p |
| 84 | |
| 85 | # -- readers ---------------------------------------------------------- |
| 86 | |
| 87 | def read_json(self, name: str) -> Any: |
| 88 | """Read a JSON file from the workspace. Returns ``None`` if missing.""" |
| 89 | p = self.root / name |
| 90 | if not p.exists(): |
| 91 | return None |
| 92 | return json.loads(p.read_text(encoding="utf-8")) |
| 93 | |
| 94 | # -- cleanup ---------------------------------------------------------- |
| 95 | |
| 96 | def cleanup(self) -> None: |
| 97 | """Remove the session directory and try to prune empty parents.""" |
| 98 | if self.root.exists(): |
| 99 | shutil.rmtree(self.root, ignore_errors=True) |
| 100 | # Walk up and remove empty parent directories |
| 101 | try: |
| 102 | sessions_dir = self.root.parent # .codewiki/sessions |
| 103 | if sessions_dir.exists() and not any(sessions_dir.iterdir()): |
| 104 | sessions_dir.rmdir() |
| 105 | base_dir = sessions_dir.parent # .codewiki |
| 106 | if base_dir.exists() and not any(base_dir.iterdir()): |
| 107 | base_dir.rmdir() |
| 108 | except OSError: |
| 109 | pass |
| 110 | logger.debug("Workspace cleaned up: %s", self.root) |
no outgoing calls
no test coverage detected