feat: manage workspace inventory, deletion, and stale cleanup

README.md

@@ -64,6 +64,7 @@ It is designed to sit **between an issue and a pull request** — with the devel
 - 🌿 **Automatic isolated Git branch per run**
 - 🐙 **GitHub issue import via API**
 - 📦 Isolated per-run workspaces
+- 🧹 Workspace listing, deletion, and stale-workspace cleanup
 - 🔍 Reviewable Git diffs
 - 🏠 Local-first LLM support with Ollama
 - 🔌 OpenAI/OpenRouter-compatible provider support
@@ -184,10 +185,24 @@ Use the planning stage as a fast first pass over an unfamiliar codebase before m
 | `/api/run` | POST | Clone, plan, edit, test, repair, and return a diff |
 | `/api/execute` | POST | Execute a command inside an existing workspace |
 | `/api/workspace` | POST | Inspect workspace files and current diff |
+| `/api/workspaces` | GET | List workspaces with branch, dirty-file count, and update time |
+| `/api/workspaces/{workspace_id}` | DELETE | Delete one workspace safely |
+| `/api/workspaces/cleanup` | POST | Delete stale workspaces older than a bounded age |
 | `/api/github/issue` | GET | Fetch a GitHub issue |
 | `/api/import-issue` | POST | Convert a GitHub issue into an agent task |
 | `/api/openapi-summary` | GET | Return PatchPilot feature metadata |
 
+Workspace cleanup is explicit and bounded. For example, to remove workspaces older than 24 hours:
+
+```http
+POST /api/workspaces/cleanup
+Content-Type: application/json
+
+{"max_age_hours": 24}
+```
+
+The cleanup request accepts 1 hour through 30 days. Workspace deletion uses the same containment boundary as workspace access so an id cannot escape `.workspaces`.
+
 The FastAPI application also exposes its generated API documentation when the backend is running:
 
 ```text

backend/app/main.py

@@ -6,6 +6,7 @@
 import shutil
 import subprocess
 import tempfile
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
@@ -49,6 +50,10 @@ class CommandRequest(BaseModel):
     command: str = Field(..., min_length=1, max_length=1000)
 
 
+class WorkspaceCleanupRequest(BaseModel):
+    max_age_hours: int = Field(default=24, ge=1, le=24 * 30)
+
+
 class IssueRequest(BaseModel):
     repo: str = Field(..., pattern=r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
     issue_number: int = Field(..., ge=1)
@@ -263,16 +268,91 @@ def model_error_preview(raw: str) -> str:
     return (raw or "").strip()[-3000:]
 
 
-def workspace_repo(workspace_id: str) -> Path:
+def workspace_dir(workspace_id: str) -> Path:
     root = WORKSPACES.resolve()
-    repo = (root / workspace_id / "repo").resolve()
+    workdir = (root / workspace_id).resolve()
     try:
-        repo.relative_to(root)
+        workdir.relative_to(root)
     except ValueError as exc:
         raise HTTPException(400, "Invalid workspace id") from exc
-    if not repo.is_dir():
+    if not workdir.is_dir():
         raise HTTPException(404, "Workspace not found")
-    return repo
+    return workdir
+
+
+def workspace_repo(workspace_id: str) -> Path:
+    repo = workspace_dir(workspace_id) / "repo"
+    if not repo.is_dir():
+        raise HTTPException(404, "Workspace repository not found")
+    return repo.resolve()
+
+
+def workspace_summary(workspace_id: str) -> dict[str, Any]:
+    workdir = workspace_dir(workspace_id)
+    repo = workspace_repo(workspace_id)
+    branch_code, branch = run(["git", "branch", "--show-current"], repo, 30)
+    status_code, status = run(["git", "status", "--short"], repo, 30)
+    return {
+        "workspace_id": workspace_id,
+        "branch": branch.strip() if branch_code == 0 else "",
+        "changed_files": (
+            len([line for line in status.splitlines() if line.strip()])
+            if status_code == 0
+            else None
+        ),
+        "updated_at": datetime.fromtimestamp(
+            workdir.stat().st_mtime,
+            tz=timezone.utc,
+        ).isoformat(),
+    }
+
+
+def list_workspace_summaries(limit: int = 100) -> list[dict[str, Any]]:
+    root = WORKSPACES.resolve()
+    rows: list[dict[str, Any]] = []
+    for entry in root.iterdir():
+        if not entry.is_dir() or not (entry / "repo").is_dir():
+            continue
+        try:
+            rows.append(workspace_summary(entry.name))
+        except HTTPException:
+            continue
+    rows.sort(key=lambda item: item["updated_at"], reverse=True)
+    return rows[:limit]
+
+
+def delete_workspace(workspace_id: str) -> None:
+    workdir = workspace_dir(workspace_id)
+    shutil.rmtree(workdir)
+
+
+def cleanup_stale_workspaces(
+    max_age_hours: int,
+    *,
+    now: datetime | None = None,
+) -> list[str]:
+    current = now or datetime.now(timezone.utc)
+    cutoff = current.timestamp() - max_age_hours * 60 * 60
+    deleted: list[str] = []
+    root = WORKSPACES.resolve()
+    for entry in root.iterdir():
+        if not entry.is_dir():
+            continue
+        try:
+            resolved = entry.resolve()
+            resolved.relative_to(root)
+        except (OSError, ValueError):
+            continue
+        try:
+            modified = resolved.stat().st_mtime
+        except OSError:
+            continue
+        if modified >= cutoff:
+            continue
+        shutil.rmtree(resolved, ignore_errors=True)
+        if not resolved.exists():
+            deleted.append(entry.name)
+    return sorted(deleted)
 
 
 def github_headers() -> dict[str, str]:
@@ -413,6 +493,28 @@ async def workspace(req: WorkspaceRequest) -> dict[str, Any]:
     return {"workspace_id": req.workspace_id, "files": list_files(repo), "diff": build_patch(repo)}
 
 
+@app.get("/api/workspaces")
+async def workspaces() -> dict[str, Any]:
+    rows = list_workspace_summaries()
+    return {"workspaces": rows, "count": len(rows)}
+
+
+@app.delete("/api/workspaces/{workspace_id}")
+async def remove_workspace(workspace_id: str) -> dict[str, Any]:
+    delete_workspace(workspace_id)
+    return {"workspace_id": workspace_id, "deleted": True}
+
+
+@app.post("/api/workspaces/cleanup")
+async def cleanup_workspaces(req: WorkspaceCleanupRequest) -> dict[str, Any]:
+    deleted = cleanup_stale_workspaces(req.max_age_hours)
+    return {
+        "deleted": deleted,
+        "count": len(deleted),
+        "max_age_hours": req.max_age_hours,
+    }
+
+
 @app.get("/api/openapi-summary")
 async def openapi_summary() -> dict[str, Any]:
     return {
@@ -425,6 +527,7 @@ async def openapi_summary() -> dict[str, Any]:
             "test execution",
             "repair loop",
             "isolated branches",
+            "workspace lifecycle management",
             "GitHub issue import",
         ],
     }

backend/tests/test_main.py

@@ -2,6 +2,7 @@
 import os
 import tempfile
 import unittest
+from datetime import datetime, timezone
 from pathlib import Path
 from unittest.mock import patch
 
@@ -10,7 +11,10 @@
 
 from app.main import (
     RunRequest,
+    WorkspaceCleanupRequest,
     apply_edits,
+    cleanup_stale_workspaces,
+    delete_workspace,
     github_headers,
     health,
     parse_json_object,
@@ -123,6 +127,58 @@ def test_workspace_repo_returns_existing_repo_inside_root(self) -> None:
             with patch("app.main.WORKSPACES", root):
                 self.assertEqual(workspace_repo("safe-id"), repo.resolve())
 
+    def test_cleanup_request_rejects_excessive_retention_window(self) -> None:
+        with self.assertRaises(ValidationError):
+            WorkspaceCleanupRequest(max_age_hours=24 * 31)
+
+    def test_cleanup_stale_workspaces_deletes_only_expired_directories(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir) / "workspaces"
+            old = root / "old-one"
+            fresh = root / "fresh-one"
+            old.mkdir(parents=True)
+            fresh.mkdir()
+            now = datetime(2026, 9, 23, 12, 0, tzinfo=timezone.utc)
+            old_ts = now.timestamp() - 48 * 60 * 60
+            fresh_ts = now.timestamp() - 2 * 60 * 60
+            os.utime(old, (old_ts, old_ts))
+            os.utime(fresh, (fresh_ts, fresh_ts))
+
+            with patch("app.main.WORKSPACES", root):
+                deleted = cleanup_stale_workspaces(24, now=now)
+
+            self.assertEqual(deleted, ["old-one"])
+            self.assertFalse(old.exists())
+            self.assertTrue(fresh.exists())
+
+    def test_delete_workspace_removes_only_selected_workspace(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir) / "workspaces"
+            first = root / "first" / "repo"
+            second = root / "second" / "repo"
+            first.mkdir(parents=True)
+            second.mkdir(parents=True)
+
+            with patch("app.main.WORKSPACES", root):
+                delete_workspace("first")
+
+            self.assertFalse(first.parent.exists())
+            self.assertTrue(second.parent.exists())
+
+    def test_delete_workspace_rejects_path_traversal(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir) / "workspaces"
+            outside = Path(temp_dir) / "outside"
+            root.mkdir()
+            outside.mkdir()
+
+            with patch("app.main.WORKSPACES", root):
+                with self.assertRaises(HTTPException) as context:
+                    delete_workspace("../outside")
+
+            self.assertEqual(context.exception.status_code, 400)
+            self.assertTrue(outside.exists())
+
     def test_github_headers_requires_token(self) -> None:
         with patch.dict(os.environ, {}, clear=True):
             with self.assertRaises(HTTPException) as context: