Fix container sync races that leaked orphan blobs; add a system-prune CLI command

sync_workspace (user-triggered) and the reconciler's sync_bidirectional_sync
could run concurrently for the same project, and store_upload's read-then-
write on a changed path meant two racing imports each wrote their own blob
while only one ever got referenced - the loser leaked forever. Combined with
no build-artifact exclusion, an actively-compiling workspace hit this
constantly and leaked 5.9M orphan blobs (~96GB) in production before it was
caught.

Closes it at the root: api._sync_dir_bidirectional_locked serializes both
call sites per-project (non-blocking - a project already mid-sync is simply
skipped until the next tick), and IMPORT_SKIP_NAMES/IMPORT_SKIP_EXTENSIONS
keep build output (build/, dist/, *.o, *.pyc, ...) out of the walk entirely.

Recovering what already leaked is a separate concern: a new CLI subcommand
(plus matching make targets) sweeps soft-deleted attachment/project-file
blobs and any blob with zero DB reference at all, plus orphaned container
workspace directories. run_maintenance_cleanup.sh wraps the existing
prune/clear commands for routine disk upkeep.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BWJy6PrMMt5hwWxQwia2rd

CLAUDE.md

@@ -58,6 +58,7 @@ devplace token prune                # soft-delete all expired access tokens
 devplace news clear         # delete all news rows
 devplace news sanitize      # strip HTML from news descriptions/content
 devplace attachments prune  # remove orphan attachment records/files
+devplace system prune [--dry-run]  # safe but aggressive: purges soft-deleted attachment/project-file blobs, sweeps blob files with zero DB reference at all (e.g. left by an interrupted/racing sync), and GCs orphaned container workspace dirs
 devplace devii reset-quota <username>   # reset one user's rolling 24h AI quota
 devplace devii reset-quota --guests     # reset every guest quota
 devplace devii reset-quota --all        # reset every quota (users and guests)

Makefile

@@ -18,7 +18,7 @@ BOOTSTRAP_PYTHON := $(shell command -v python3 2>/dev/null || command -v python
 PYTHONDONTWRITEBYTECODE := 1
 export PYTHONDONTWRITEBYTECODE
 
-.PHONY: venv install dev prod clean tree tree-loc zip test test-headed test-unit test-api test-e2e test-fast test-failed test-first-failure test-slowest test-cache-clean coverage coverage-headed coverage-html locust locust-headless
+.PHONY: venv install dev prod clean tree tree-loc zip test test-headed test-unit test-api test-e2e test-fast test-failed test-first-failure test-slowest test-cache-clean coverage coverage-headed coverage-html locust locust-headless prune prune-dry-run
 
 $(PYTHON):
 	@test -n "$(BOOTSTRAP_PYTHON)" || { echo "python3 is required to create $(VENV)"; exit 1; }
@@ -148,6 +148,15 @@ clean:
 test-cache-clean:
 	rm -rf .pytest_cache
 
+# Safe but aggressive disk-space cleanup, acting on the REAL data/devplace.db
+# and data/ tree (never the test database) - see CLAUDE.md "devplace system
+# prune". prune-dry-run reports what would be removed without deleting.
+prune: $(VENV_STAMP)
+	$(PYTHON) -m devplacepy.cli system prune
+
+prune-dry-run: $(VENV_STAMP)
+	$(PYTHON) -m devplacepy.cli system prune --dry-run
+
 # Container Manager works out of the box: the overlay installs the docker CLI in
 # the image and mounts the host socket. DOCKER_GID is read straight from the
 # socket so the UID-1000 app can use it; the data dir is the project's own data/

devplacepy/attachments.py

@@ -3,6 +3,7 @@
 import asyncio
 import ipaddress
 import logging
+import os
 import socket
 from datetime import datetime, timezone
 from pathlib import Path
@@ -637,6 +638,73 @@ def restore_attachment(uid):
     return True
 
 
+def purge_soft_deleted_attachments(*, dry_run: bool = False) -> tuple[int, int]:
+    if "attachments" not in db.tables:
+        return 0, 0
+    table = get_table("attachments")
+    rows = list(table.find(table.table.columns.deleted_at.isnot(None)))
+    removed = 0
+    freed = 0
+    for row in rows:
+        directory = row.get("directory")
+        stored_name = row.get("stored_name")
+        if directory and stored_name:
+            path = ATTACHMENTS_DIR / directory / stored_name
+            try:
+                freed += path.stat().st_size
+            except OSError:
+                pass
+            for thumb in (ATTACHMENTS_DIR / directory).glob(
+                f"{Path(stored_name).stem}_thumb.*"
+            ):
+                try:
+                    freed += thumb.stat().st_size
+                except OSError:
+                    pass
+            if not dry_run:
+                _unlink_attachment_files(row)
+        if not dry_run:
+            table.delete(id=row["id"])
+        removed += 1
+    return removed, freed
+
+
+def sweep_orphan_attachment_blobs(*, dry_run: bool = False) -> tuple[int, int]:
+    if not ATTACHMENTS_DIR.exists():
+        return 0, 0
+    referenced_stems = set()
+    if "attachments" in db.tables:
+        for row in get_table("attachments").find():
+            directory = row.get("directory")
+            stored_name = row.get("stored_name")
+            if directory and stored_name:
+                referenced_stems.add((directory, Path(stored_name).stem))
+    removed = 0
+    freed = 0
+    base = str(ATTACHMENTS_DIR)
+    for root, _dirs, files in os.walk(base):
+        directory = os.path.relpath(root, base)
+        for name in files:
+            stem = Path(name).stem
+            if stem.endswith("_thumb"):
+                stem = stem[: -len("_thumb")]
+            if (directory, stem) in referenced_stems:
+                continue
+            file_path = os.path.join(root, name)
+            try:
+                size = os.path.getsize(file_path)
+            except OSError:
+                continue
+            if not dry_run:
+                try:
+                    os.unlink(file_path)
+                except OSError:
+                    continue
+            removed += 1
+            freed += size
+    return removed, freed
+
+
 def soft_delete_target_attachments(target_type, target_uid, deleted_by):
     stamp = datetime.now(timezone.utc).isoformat()
     for row in get_table("attachments").find(

devplacepy/cli/__init__.py

@@ -45,6 +45,7 @@ from devplacepy.cli.containers import (
 )
 from devplacepy.cli.quiz import cmd_quiz_prune
 from devplacepy.cli.migrate import cmd_emoji_sync, cmd_migrate_data
+from devplacepy.cli.system import cmd_system_prune
 
 __all__ = [
     "main",
@@ -91,4 +92,5 @@ __all__ = [
     "cmd_quiz_prune",
     "cmd_emoji_sync",
     "cmd_migrate_data",
+    "cmd_system_prune",
 ]

devplacepy/cli/containers.py

@@ -67,19 +67,9 @@ def cmd_containers_prune_builds(args):
 
 
 def cmd_containers_gc_workspaces(args):
-    import shutil
-    from pathlib import Path
-    from devplacepy import config
     from devplacepy.services.containers import store
 
-    active = {inst["project_uid"] for inst in store.all_instances()}
-    base = Path(config.CONTAINER_WORKSPACES_DIR)
-    removed = 0
-    if base.is_dir():
-        for child in base.iterdir():
-            if child.is_dir() and child.name not in active:
-                shutil.rmtree(child, ignore_errors=True)
-                removed += 1
+    removed = store.gc_workspaces()
     _audit_cli("cli.containers.gc_workspaces", f"CLI removed {removed} unused workspace dirs", metadata={"count": removed})
     print(
         f"Removed {removed} unused workspace director{'y' if removed == 1 else 'ies'}"

devplacepy/cli/main.py

@@ -17,6 +17,7 @@ from devplacepy.cli.game import register_game
 from devplacepy.cli.quiz import register_quiz
 from devplacepy.cli.gateway import register_gateway
 from devplacepy.cli.messaging import register_messaging
+from devplacepy.cli.system import register_system
 
 
 def build_parser():
@@ -38,6 +39,7 @@ def build_parser():
     register_gateway(sub)
     register_messaging(sub)
     register_accounts(sub)
+    register_system(sub)
 
     return parser
 

devplacepy/cli/system.py

@@ -0,0 +1,92 @@
+# retoor <retoor@molodetz.nl>
+
+from devplacepy.cli._shared import _audit_cli
+
+
+def _human(n):
+    n = float(n)
+    for unit in ("B", "KB", "MB", "GB", "TB"):
+        if n < 1024:
+            return f"{n:.1f}{unit}"
+        n /= 1024
+    return f"{n:.1f}PB"
+
+
+def cmd_system_prune(args):
+    from devplacepy.attachments import (
+        purge_soft_deleted_attachments,
+        sweep_orphan_attachment_blobs,
+    )
+    from devplacepy.project_files import (
+        purge_soft_deleted_project_files,
+        sweep_orphan_project_file_blobs,
+    )
+    from devplacepy.services.containers import store as container_store
+
+    dry_run = bool(args.dry_run)
+    verb = "Would remove" if dry_run else "Removed"
+
+    totals = {"items": 0, "freed": 0}
+
+    def report(name, count, freed):
+        totals["items"] += count
+        totals["freed"] += freed
+        print(f"  {name}: {verb.lower()} {count} item(s), {_human(freed)}")
+
+    print("System prune - safe but aggressive disk-space cleanup")
+    print(
+        "Removes only content with zero live reference: soft-deleted attachment/"
+        "\nproject-file blobs, and blob files with no matching database row at all"
+        "\n(orphans, e.g. left behind by an interrupted or racing sync). Never"
+        "\ntouches live content."
+    )
+    if dry_run:
+        print("DRY RUN: nothing will be deleted.")
+    print()
+
+    count, freed = purge_soft_deleted_attachments(dry_run=dry_run)
+    report("soft-deleted attachments", count, freed)
+
+    count, freed = purge_soft_deleted_project_files(dry_run=dry_run)
+    report("soft-deleted project files", count, freed)
+
+    count, freed = sweep_orphan_attachment_blobs(dry_run=dry_run)
+    report("orphan attachment blobs", count, freed)
+
+    count, freed = sweep_orphan_project_file_blobs(dry_run=dry_run)
+    report("orphan project-file blobs", count, freed)
+
+    removed = container_store.gc_workspaces(dry_run=dry_run)
+    totals["items"] += removed
+    plural = "y" if removed == 1 else "ies"
+    print(f"  orphaned container workspaces: {verb.lower()} {removed} director{plural}")
+
+    print()
+    suffix = " (estimate)" if dry_run else ""
+    print(f"{verb} {totals['items']} item(s) total, {_human(totals['freed'])} of disk{suffix}")
+
+    if not dry_run:
+        _audit_cli(
+            "cli.system.prune",
+            f"CLI system prune removed {totals['items']} item(s), freed {_human(totals['freed'])}",
+            metadata={"items": totals["items"], "bytes_freed": totals["freed"]},
+        )
+
+
+def register_system(subparsers):
+    system = subparsers.add_parser("system", help="Cross-cutting system maintenance")
+    system_sub = system.add_subparsers(title="action", dest="action")
+    prune = system_sub.add_parser(
+        "prune",
+        help=(
+            "Safe but aggressive disk-space cleanup: purges soft-deleted "
+            "attachment/project-file blobs and any blob with zero database "
+            "reference at all, plus orphaned container workspace directories"
+        ),
+    )
+    prune.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Report what would be removed without deleting anything",
+    )
+    prune.set_defaults(func=cmd_system_prune)

devplacepy/database/pagination.py

@@ -18,6 +18,7 @@ def paginate(
     order=None,
     cursor_field="created_at",
     viewer_uid=None,
+    limit=PAGE_SIZE,
     **filters,
 ):
     order = order or ["-" + cursor_field]
@@ -30,9 +31,9 @@ def paginate(
             clauses.append(table.table.columns.user_uid.notin_(blocked))
     if before:
         clauses.append(table.table.columns[cursor_field] < before)
-    rows = list(table.find(*clauses, **filters, order_by=order, _limit=PAGE_SIZE + 1))
-    has_more = len(rows) > PAGE_SIZE
-    rows = rows[:PAGE_SIZE]
+    rows = list(table.find(*clauses, **filters, order_by=order, _limit=limit + 1))
+    has_more = len(rows) > limit
+    rows = rows[:limit]
     next_cursor = rows[-1][cursor_field] if has_more and rows else None
     return rows, next_cursor
 
@@ -64,6 +65,7 @@ def paginate_diverse(
     cursor_field="created_at",
     uid_key="user_uid",
     viewer_uid=None,
+    limit=PAGE_SIZE,
     **filters,
 ):
     rows, next_cursor = paginate(
@@ -73,6 +75,7 @@ def paginate_diverse(
         order=order,
         cursor_field=cursor_field,
         viewer_uid=viewer_uid,
+        limit=limit,
         **filters,
     )
     return interleave_by_author(rows, uid_key=uid_key), next_cursor

devplacepy/project_files.py

@@ -1,6 +1,7 @@
 # retoor <retoor@molodetz.nl>
 
 import logging
+import os
 import shutil
 from datetime import datetime, timezone
 from pathlib import Path
@@ -554,7 +555,7 @@ def _walk_workspace_files(src, skip):
     for root, dirs, files in src.walk():
         dirs[:] = [d for d in sorted(dirs) if d not in skip]
         for name in sorted(files):
-            if name in skip:
+            if name in skip or Path(name).suffix in IMPORT_SKIP_EXTENSIONS:
                 continue
             full = Path(root) / name
             if full.is_symlink() or not full.is_file():
@@ -780,6 +781,36 @@ IMPORT_SKIP_NAMES = {
     ".DS_Store",
     ".idea",
     ".cache",
+    "build",
+    "dist",
+    "target",
+    "out",
+    "bin",
+    "obj",
+    ".next",
+    ".nuxt",
+    ".gradle",
+    ".tox",
+    "cmake-build-debug",
+    "cmake-build-release",
+}
+# Compiled/build-artifact extensions, regenerated wholesale on every build -
+# never worth importing/syncing regardless of which directory they land in
+# (unlike IMPORT_SKIP_NAMES, matched by suffix rather than exact name; see
+# _walk_workspace_files). A missing exclusion here is what let an unlocked
+# concurrent sync (see api._sync_dir_bidirectional_locked) leak millions of
+# orphaned blobs from an actively-compiling container workspace.
+IMPORT_SKIP_EXTENSIONS = {
+    ".o",
+    ".obj",
+    ".pyc",
+    ".pyo",
+    ".class",
+    ".so",
+    ".dylib",
+    ".dll",
+    ".a",
+    ".exe",
 }
 SYNC_SKIP_NAMES = IMPORT_SKIP_NAMES | {
     ".devplace_boot.py",
@@ -998,3 +1029,61 @@ def delete_all_project_files(project_uid: str) -> None:
         if row.get("is_binary"):
             _unlink_blob(row)
     _table().delete(project_uid=project_uid)
+
+
+def purge_soft_deleted_project_files(*, dry_run: bool = False) -> tuple[int, int]:
+    if "project_files" not in db.tables:
+        return 0, 0
+    table = _table()
+    rows = list(table.find(table.table.columns.deleted_at.isnot(None), is_binary=1))
+    removed = 0
+    freed = 0
+    for row in rows:
+        directory = row.get("directory")
+        stored_name = row.get("stored_name")
+        if directory and stored_name:
+            path = PROJECT_FILES_DIR / directory / stored_name
+            try:
+                freed += path.stat().st_size
+            except OSError:
+                pass
+            else:
+                if not dry_run:
+                    _unlink_blob(row)
+        if not dry_run:
+            table.delete(id=row["id"])
+        removed += 1
+    return removed, freed
+
+
+def sweep_orphan_project_file_blobs(*, dry_run: bool = False) -> tuple[int, int]:
+    if not PROJECT_FILES_DIR.exists():
+        return 0, 0
+    referenced = set()
+    if "project_files" in db.tables:
+        for row in _table().find(is_binary=1):
+            directory = row.get("directory")
+            stored_name = row.get("stored_name")
+            if directory and stored_name:
+                referenced.add((directory, stored_name))
+    removed = 0
+    freed = 0
+    base = str(PROJECT_FILES_DIR)
+    for root, _dirs, files in os.walk(base):
+        directory = os.path.relpath(root, base)
+        for name in files:
+            if (directory, name) in referenced:
+                continue
+            file_path = os.path.join(root, name)
+            try:
+                size = os.path.getsize(file_path)
+            except OSError:
+                continue
+            if not dry_run:
+                try:
+                    os.unlink(file_path)
+                except OSError:
+                    continue
+            removed += 1
+            freed += size
+    return removed, freed

devplacepy/services/containers/CLAUDE.md

@@ -408,6 +408,46 @@ gated on `time.monotonic()` independent of the 5s reconcile tick). The per-insta
 boot-helper files (`.devplace_boot.py`/`.devplace_boot.sh`) are in `SYNC_SKIP_NAMES` so they
 never round-trip into the project, and therefore never enter the manifest either.
 
+**Both call sites are unsynchronized, and that used to leak millions of orphan blobs
+(load-bearing, incident-derived).** `sync_workspace` (user-triggered) and
+`sync_bidirectional_sync` (the reconciler tick) can run for the SAME project concurrently
+whenever a sync outlasts the reconcile interval - trivial for any workspace that is
+actively compiling. `project_files.store_upload` (the import path for a changed file)
+writes a NEW blob under a fresh uuid, then updates the existing row to point at it; two
+racing imports of the same changed path each write their own blob and both update the
+same row (last writer wins), so the loser's freshly-written blob is referenced by nothing
+- an orphan on every race, not just on a crash. Combined with no build-artifact exclusion,
+an actively-compiling workspace regenerates thousands of files with fresh mtimes on every
+build, which is exactly the high-churn pattern that maximizes how often the race fires;
+sustained over time this leaked **5.9 million orphan blobs (~96GB)** in production, discovered
+only when the host disk hit 100% full. Both are now closed at their root:
+
+- **`api._sync_dir_bidirectional_locked`** wraps `project_files.sync_dir_bidirectional`
+  behind a per-`project_uid` `threading.Lock` (`api._sync_lock_for`, a lazily-created
+  registry - both `sync_workspace` and `sync_bidirectional_sync` call it instead of the
+  raw function). The acquire is **non-blocking**: a second sync for a project already mid-sync
+  is skipped outright (returns the zero-counts dict), never queued or awaited. This runs on
+  an `asyncio.to_thread` worker in both call sites, so blocking would tie up a thread pool
+  slot for no reason - a skipped project is simply picked up on the next tick or the user's
+  next explicit sync.
+- **`project_files.IMPORT_SKIP_NAMES`** gained common build-output directory names (`build`,
+  `dist`, `target`, `out`, `bin`, `obj`, `.next`, `.nuxt`, `.gradle`, `.tox`,
+  `cmake-build-debug`, `cmake-build-release`) and a NEW parallel set,
+  **`IMPORT_SKIP_EXTENSIONS`** (`.o`, `.obj`, `.pyc`, `.pyo`, `.class`, `.so`, `.dylib`,
+  `.dll`, `.a`, `.exe`), matched by suffix in `_walk_workspace_files` rather than exact
+  name (`IMPORT_SKIP_NAMES`/`SYNC_SKIP_NAMES` are exact-name-only sets, which cannot express
+  "any file ending in `.o`"). Both apply to every `_walk_workspace_files` caller (`import_from_dir`
+  AND the sync path via `SYNC_SKIP_NAMES = IMPORT_SKIP_NAMES | {...}`), so compiled/build
+  artifacts are never imported into a project regardless of entry point, not just never synced.
+
+Neither fix requires (or should ever regress into) a redesign of the manifest/reconciliation
+model above - the lock only prevents two runs from touching the same project's `project_files`
+rows at once, and the skip sets only shrink what `_walk_workspace_files` yields. Recovering
+already-leaked blobs is a separate, data-layer concern: see `devplace system prune`
+(`cli/system.py`) and `attachments.sweep_orphan_attachment_blobs`/
+`project_files.sweep_orphan_project_file_blobs`, which sweep any blob with zero database
+reference at all, regardless of what leaked it.
+
 ## Run-as user = identity + API key ONLY (load-bearing constraint)
 
 An instance's `run_as_uid` column selects WHICH DevPlace user's identity and `api_key` are injected (`DEVPLACE_API_KEY`, `DEVPLACE_USER_UID`), resolved in `api.pravda_env` ahead of the `created_by`/`owner_uid` fallback chain. It does **NOT** change the container OS user, which is ALWAYS `pravda` (uid 1000) - required for the bind-mounted `/app` (DooD uid maps 1:1 to host). Validate it against an existing user via `api.validate_run_as`.

devplacepy/services/containers/api.py

@@ -5,6 +5,7 @@ import json
 import re
 import secrets
 import socket
+import threading
 from pathlib import Path
 
 from devplacepy import config, project_files, stealth
@@ -611,12 +612,49 @@ def run_spec_for(instance: dict, image_tag: str) -> RunSpec:
     )
 
 
+_SYNC_LOCKS: dict[str, threading.Lock] = {}
+_SYNC_LOCKS_GUARD = threading.Lock()
+_SYNC_EMPTY_COUNTS = {
+    "exported": 0,
+    "imported": 0,
+    "deleted_in_project": 0,
+    "deleted_in_workspace": 0,
+}
+
+
+def _sync_lock_for(project_uid: str) -> threading.Lock:
+    with _SYNC_LOCKS_GUARD:
+        lock = _SYNC_LOCKS.get(project_uid)
+        if lock is None:
+            lock = threading.Lock()
+            _SYNC_LOCKS[project_uid] = lock
+        return lock
+
+
+def _sync_dir_bidirectional_locked(project_uid: str, workspace, user: dict) -> dict:
+    # The user-triggered sync and the reconciler's periodic sync both call
+    # this, unsynchronized. Without this lock, two overlapping runs for the
+    # SAME project race sync_dir_bidirectional -> store_upload: each writes
+    # its own fresh blob for a changed file, both update the same DB row
+    # (last writer wins), and the loser's blob is now referenced by
+    # nothing - an orphan on every race, forever. A non-blocking skip (never
+    # a blocking wait) is deliberate: this runs on an asyncio.to_thread
+    # worker, and a project mid-sync simply gets picked up on the next tick.
+    lock = _sync_lock_for(project_uid)
+    if not lock.acquire(blocking=False):
+        return dict(_SYNC_EMPTY_COUNTS)
+    try:
+        return project_files.sync_dir_bidirectional(project_uid, workspace, user)
+    finally:
+        lock.release()
+
+
 async def sync_workspace(instance: dict, user: dict) -> dict:
     workspace = instance.get("workspace_dir")
     if not workspace:
         raise ContainerError("instance has no workspace")
     counts = await asyncio.to_thread(
-        project_files.sync_dir_bidirectional, instance["project_uid"], workspace, user
+        _sync_dir_bidirectional_locked, instance["project_uid"], workspace, user
     )
     store.record_event(
         instance,
@@ -632,7 +670,7 @@ def sync_bidirectional_sync(instance: dict, user: dict) -> dict:
     workspace = instance.get("workspace_dir")
     if not workspace:
         return {"exported": 0, "imported": 0}
-    counts = project_files.sync_dir_bidirectional(
+    counts = _sync_dir_bidirectional_locked(
         instance["project_uid"], workspace, user
     )
     if counts["exported"] or counts["imported"]:

devplacepy/services/containers/store.py

@@ -92,6 +92,25 @@ def all_instances() -> list:
     return list(get_table("instances").find(deleted_at=None))
 
 
+def gc_workspaces(*, dry_run: bool = False) -> int:
+    import shutil
+    from pathlib import Path
+
+    from devplacepy import config
+
+    active = {inst["project_uid"] for inst in all_instances()}
+    base = Path(config.CONTAINER_WORKSPACES_DIR)
+    if not base.is_dir():
+        return 0
+    removed = 0
+    for child in base.iterdir():
+        if child.is_dir() and child.name not in active:
+            if not dry_run:
+                shutil.rmtree(child, ignore_errors=True)
+            removed += 1
+    return removed
+
+
 def find_instance_by_ingress(slug: str):
     if not slug or not _exists("instances"):
         return None

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "devplacepy"
-version = "1.0.3"
+version = "1.0.4"
 description = "DevPlace - The Developer Social Network"
 requires-python = ">=3.12"
 dependencies = [

run_maintenance_cleanup.sh

@@ -0,0 +1,210 @@
+#!/usr/bin/env bash
+# retoor <retoor@molodetz.nl>
+
+set -uo pipefail
+cd "$(dirname "${BASH_SOURCE[0]}")" 2>/dev/null || true
+
+REPO_DIR="${DEVPLACE_REPO_DIR:-/home/retoor/projects/devplacepy}"
+cd "$REPO_DIR" || { echo "Cannot cd into $REPO_DIR - set DEVPLACE_REPO_DIR"; exit 1; }
+
+if [ -x "$REPO_DIR/.venv/bin/devplace" ]; then
+    DEVPLACE="$REPO_DIR/.venv/bin/devplace"
+elif command -v devplace >/dev/null 2>&1; then
+    DEVPLACE="devplace"
+else
+    echo "Cannot find the devplace binary (looked in $REPO_DIR/.venv/bin and PATH)."
+    echo "Run 'make install' once to create the .venv, or activate it yourself."
+    exit 1
+fi
+
+if [ -x "$REPO_DIR/.venv/bin/python" ]; then
+    PYTHON="$REPO_DIR/.venv/bin/python"
+elif command -v python3 >/dev/null 2>&1; then
+    PYTHON="python3"
+else
+    echo "Cannot find a python interpreter (looked in $REPO_DIR/.venv/bin and PATH)."
+    exit 1
+fi
+
+DIRS=(
+    data/uploads
+    data/uploads/attachments
+    data/uploads/project_files
+    data/zips
+    data/zip_staging
+    data/fork_staging
+    data/seo_reports
+    data/deepsearch
+    data/isslop
+    data/container_workspaces
+    data/workspace_state
+)
+
+show_sizes() {
+    echo "--- $1 ---"
+    for d in "${DIRS[@]}"; do
+        if [ -d "$d" ]; then
+            printf "  %-10s  %s\n" "$(du -sh "$d" 2>/dev/null | cut -f1)" "$d"
+        else
+            printf "  %-10s  %s\n" "-" "$d (absent)"
+        fi
+    done
+    echo
+}
+
+FAILED=()
+
+run() {
+    echo "=== devplace $* ==="
+    "$DEVPLACE" "$@"
+    local status=$?
+    if [ "$status" -ne 0 ]; then
+        echo "!!! FAILED (exit $status): devplace $*"
+        FAILED+=("devplace $*")
+    fi
+    echo
+}
+
+echo "########################################"
+echo "# DevPlace maintenance cleanup"
+echo "# backups and news are deliberately excluded"
+echo "########################################"
+echo
+
+show_sizes "BEFORE"
+
+run attachments prune
+run zips prune
+run zips clear
+run forks prune
+run forks clear
+run seo prune
+run seo clear
+run seo-meta prune
+run seo-meta clear
+run deepsearch prune
+run deepsearch clear
+run isslop prune
+run isslop clear
+run quiz prune
+run game market prune
+run game steals prune
+run messaging prune-tickets
+run devii tasks prune
+run accounts prune
+run containers prune
+run containers prune-builds
+run containers gc-workspaces
+
+show_sizes "AFTER"
+
+echo "########################################"
+echo "# Soft-deleted attachment / project-file blobs still on disk"
+echo "# (READ-ONLY report - nothing below this line deletes anything;"
+echo "#  none of the commands above purge these, only the admin Trash"
+echo "#  'Purge' button per-event, or accounts prune, do that today)"
+echo "########################################"
+"$PYTHON" - "$REPO_DIR" <<'PY'
+import sqlite3
+import sys
+from pathlib import Path
+
+repo_dir = Path(sys.argv[1])
+data_dir = Path(__import__("os").environ.get("DEVPLACE_DATA_DIR", str(repo_dir / "data")))
+db_path = data_dir / "devplace.db"
+
+
+def human(n):
+    n = float(n)
+    for unit in ("B", "KB", "MB", "GB", "TB"):
+        if n < 1024:
+            return f"{n:.1f}{unit}"
+        n /= 1024
+    return f"{n:.1f}PB"
+
+
+if not db_path.exists():
+    print(f"  {db_path} does not exist, skipping")
+    raise SystemExit(0)
+
+con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
+try:
+    cur = con.cursor()
+
+    def columns(table):
+        return {row[1] for row in cur.execute(f"PRAGMA table_info('{table}')")}
+
+    # attachments: data/uploads/attachments/{directory}/{stored_name}, plus any
+    # {stem}_thumb.* sibling thumbnail (see attachments._unlink_attachment_files)
+    cols = columns("attachments")
+    if {"directory", "stored_name", "deleted_at"} <= cols:
+        rows = cur.execute(
+            "SELECT directory, stored_name FROM attachments WHERE deleted_at IS NOT NULL"
+        ).fetchall()
+        base = data_dir / "uploads" / "attachments"
+        present, total = 0, 0
+        for directory, stored_name in rows:
+            if not directory or not stored_name:
+                continue
+            fp = base / directory / stored_name
+            if fp.exists():
+                present += 1
+                try:
+                    total += fp.stat().st_size
+                except OSError:
+                    pass
+                stem = Path(stored_name).stem
+                for thumb in (base / directory).glob(f"{stem}_thumb.*"):
+                    try:
+                        total += thumb.stat().st_size
+                    except OSError:
+                        pass
+        print(
+            f"  attachments:   {len(rows)} soft-deleted row(s), "
+            f"{present} still on disk, {human(total)} reclaimable"
+        )
+    else:
+        print(f"  attachments: unexpected schema, columns={sorted(cols)}")
+
+    # project_files: data/uploads/project_files/{directory}/{stored_name},
+    # only rows with is_binary=1 have a blob at all (text lives in the DB row)
+    cols = columns("project_files")
+    if {"directory", "stored_name", "deleted_at", "is_binary"} <= cols:
+        rows = cur.execute(
+            "SELECT directory, stored_name FROM project_files "
+            "WHERE deleted_at IS NOT NULL AND is_binary = 1"
+        ).fetchall()
+        base = data_dir / "uploads" / "project_files"
+        present, total = 0, 0
+        for directory, stored_name in rows:
+            if not directory or not stored_name:
+                continue
+            fp = base / directory / stored_name
+            if fp.exists():
+                present += 1
+                try:
+                    total += fp.stat().st_size
+                except OSError:
+                    pass
+        print(
+            f"  project_files: {len(rows)} soft-deleted binary row(s), "
+            f"{present} still on disk, {human(total)} reclaimable"
+        )
+    else:
+        print(f"  project_files: unexpected schema, columns={sorted(cols)}")
+finally:
+    con.close()
+PY
+echo
+
+if [ "${#FAILED[@]}" -gt 0 ]; then
+    echo "########################################"
+    echo "# ${#FAILED[@]} command(s) failed:"
+    for cmd in "${FAILED[@]}"; do
+        echo "#   $cmd"
+    done
+    echo "########################################"
+    exit 1
+fi
+
+echo "All commands completed successfully."

tests/unit/attachments.py

@@ -0,0 +1,149 @@
+# retoor <retoor@molodetz.nl>
+
+from devplacepy import attachments as att
+from devplacepy.database import get_table
+from devplacepy.utils import generate_uid
+
+
+def _make_attachment(directory, stored_name, deleted_at=None):
+    uid = generate_uid()
+    get_table("attachments").insert(
+        {
+            "uid": uid,
+            "deleted_at": deleted_at,
+            "deleted_by": None,
+            "directory": directory,
+            "stored_name": stored_name,
+            "target_type": "post",
+            "target_uid": generate_uid(),
+        }
+    )
+    return uid
+
+
+def _write_blob(base, directory, stored_name, content=b"data"):
+    file_dir = base / directory
+    file_dir.mkdir(parents=True, exist_ok=True)
+    (file_dir / stored_name).write_bytes(content)
+    return file_dir / stored_name
+
+
+def test_purge_soft_deleted_attachments_removes_row_and_blob(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
+    directory = "ab/cd"
+    stored_name = f"{generate_uid()}.png"
+    _write_blob(tmp_path, directory, stored_name, b"x" * 100)
+    uid = _make_attachment(directory, stored_name, deleted_at="2020-01-01T00:00:00+00:00")
+
+    removed, freed = att.purge_soft_deleted_attachments()
+
+    assert removed >= 1
+    assert freed == 100
+    assert get_table("attachments").find_one(uid=uid) is None
+    assert not (tmp_path / directory / stored_name).exists()
+
+
+def test_purge_soft_deleted_attachments_dry_run_changes_nothing(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
+    directory = "ab/cd"
+    stored_name = f"{generate_uid()}.png"
+    _write_blob(tmp_path, directory, stored_name, b"x" * 50)
+    uid = _make_attachment(directory, stored_name, deleted_at="2020-01-01T00:00:00+00:00")
+
+    removed, freed = att.purge_soft_deleted_attachments(dry_run=True)
+
+    assert removed >= 1
+    assert freed == 50
+    assert get_table("attachments").find_one(uid=uid) is not None
+    assert (tmp_path / directory / stored_name).exists()
+
+
+def test_purge_soft_deleted_attachments_ignores_live_rows(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
+    directory = "ab/cd"
+    stored_name = f"{generate_uid()}.png"
+    _write_blob(tmp_path, directory, stored_name)
+    uid = _make_attachment(directory, stored_name, deleted_at=None)
+
+    att.purge_soft_deleted_attachments()
+
+    assert get_table("attachments").find_one(uid=uid) is not None
+    assert (tmp_path / directory / stored_name).exists()
+
+
+def test_purge_soft_deleted_attachments_also_removes_thumbnail(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
+    directory = "ab/cd"
+    stem = generate_uid()
+    stored_name = f"{stem}.jpg"
+    _write_blob(tmp_path, directory, stored_name, b"x" * 20)
+    _write_blob(tmp_path, directory, f"{stem}_thumb.jpg", b"y" * 5)
+    _make_attachment(directory, stored_name, deleted_at="2020-01-01T00:00:00+00:00")
+
+    removed, freed = att.purge_soft_deleted_attachments()
+
+    assert removed >= 1
+    assert freed == 25
+    assert not (tmp_path / directory / f"{stem}_thumb.jpg").exists()
+
+
+def test_sweep_orphan_attachment_blobs_removes_unreferenced_file(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
+    directory = "ab/cd"
+    orphan_name = f"{generate_uid()}.o"
+    _write_blob(tmp_path, directory, orphan_name, b"y" * 30)
+
+    removed, freed = att.sweep_orphan_attachment_blobs()
+
+    assert removed == 1
+    assert freed == 30
+    assert not (tmp_path / directory / orphan_name).exists()
+
+
+def test_sweep_orphan_attachment_blobs_keeps_referenced_file_and_its_thumbnail(
+    local_db, tmp_path, monkeypatch
+):
+    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
+    directory = "ab/cd"
+    stem = generate_uid()
+    stored_name = f"{stem}.jpg"
+    thumb_name = f"{stem}_thumb.jpg"
+    _write_blob(tmp_path, directory, stored_name)
+    _write_blob(tmp_path, directory, thumb_name)
+    _make_attachment(directory, stored_name, deleted_at=None)
+
+    removed, freed = att.sweep_orphan_attachment_blobs()
+
+    assert removed == 0
+    assert freed == 0
+    assert (tmp_path / directory / stored_name).exists()
+    assert (tmp_path / directory / thumb_name).exists()
+
+
+def test_sweep_orphan_attachment_blobs_keeps_blob_referenced_only_by_soft_deleted_row(
+    local_db, tmp_path, monkeypatch
+):
+    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
+    directory = "ab/cd"
+    stored_name = f"{generate_uid()}.png"
+    _write_blob(tmp_path, directory, stored_name)
+    _make_attachment(directory, stored_name, deleted_at="2020-01-01T00:00:00+00:00")
+
+    removed, freed = att.sweep_orphan_attachment_blobs()
+
+    assert removed == 0
+    assert freed == 0
+    assert (tmp_path / directory / stored_name).exists()
+
+
+def test_sweep_orphan_attachment_blobs_dry_run_changes_nothing(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path)
+    directory = "ab/cd"
+    orphan_name = f"{generate_uid()}.o"
+    _write_blob(tmp_path, directory, orphan_name, b"z" * 10)
+
+    removed, freed = att.sweep_orphan_attachment_blobs(dry_run=True)
+
+    assert removed == 1
+    assert freed == 10
+    assert (tmp_path / directory / orphan_name).exists()

tests/unit/cli.py

@@ -161,3 +161,44 @@ def test_main_dispatches_subcommand(local_db, monkeypatch, capsys):
     monkeypatch.setattr("sys.argv", ["devplace", "role", "get", username])
     cli.main()
     assert capsys.readouterr().out.strip() == "admin"
+
+
+def test_system_prune_reclaims_orphans_across_subsystems(local_db, tmp_path, capsys, monkeypatch):
+    from devplacepy import attachments as att
+    from devplacepy import project_files as pf
+
+    monkeypatch.setattr(att, "ATTACHMENTS_DIR", tmp_path / "attachments")
+    monkeypatch.setattr(pf, "PROJECT_FILES_DIR", tmp_path / "project_files")
+    monkeypatch.setattr("devplacepy.config.CONTAINER_WORKSPACES_DIR", tmp_path / "workspaces")
+
+    attachments_dir = tmp_path / "attachments" / "ab" / "cd"
+    attachments_dir.mkdir(parents=True)
+    (attachments_dir / "orphan.png").write_bytes(b"a" * 10)
+
+    project_files_dir = tmp_path / "project_files" / "ab" / "cd"
+    project_files_dir.mkdir(parents=True)
+    (project_files_dir / "orphan.o").write_bytes(b"b" * 20)
+
+    (tmp_path / "workspaces").mkdir()
+    (tmp_path / "workspaces" / "orphan-project").mkdir()
+
+    cli.cmd_system_prune(argparse.Namespace(dry_run=True))
+    dry_output = capsys.readouterr().out
+    assert "DRY RUN" in dry_output
+    assert (attachments_dir / "orphan.png").exists()
+    assert (project_files_dir / "orphan.o").exists()
+    assert (tmp_path / "workspaces" / "orphan-project").exists()
+
+    cli.cmd_system_prune(argparse.Namespace(dry_run=False))
+    real_output = capsys.readouterr().out
+    assert "DRY RUN" not in real_output
+    assert not (attachments_dir / "orphan.png").exists()
+    assert not (project_files_dir / "orphan.o").exists()
+    assert not (tmp_path / "workspaces" / "orphan-project").exists()
+
+
+def test_system_prune_registered_in_parser():
+    parser = cli.build_parser()
+    args = parser.parse_args(["system", "prune", "--dry-run"])
+    assert args.func is cli.cmd_system_prune
+    assert args.dry_run is True

tests/unit/project_files.py

@@ -388,3 +388,144 @@ def test_docs_group_present():
     assert group is not None
     ids = {e["id"] for e in group["endpoints"]}
     assert "project-files-write" in ids and "project-files-list" in ids
+
+
+def _insert_binary_node(project_uid, directory, stored_name, deleted_at=None):
+    from devplacepy.utils import generate_uid
+
+    uid = generate_uid()
+    pf._table().insert(
+        {
+            "uid": uid,
+            "project_uid": project_uid,
+            "deleted_at": deleted_at,
+            "deleted_by": None,
+            "user_uid": "system-prune-test-user",
+            "path": f"/{stored_name}",
+            "name": stored_name,
+            "parent_path": "/",
+            "type": "file",
+            "content": None,
+            "is_binary": 1,
+            "stored_name": stored_name,
+            "directory": directory,
+            "mime_type": "application/octet-stream",
+            "size": 0,
+            "created_at": pf._now(),
+            "updated_at": pf._now(),
+        }
+    )
+    return uid
+
+
+def _write_project_blob(base, directory, stored_name, content=b"data"):
+    file_dir = base / directory
+    file_dir.mkdir(parents=True, exist_ok=True)
+    (file_dir / stored_name).write_bytes(content)
+
+
+def test_purge_soft_deleted_project_files_removes_row_and_blob(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr(pf, "PROJECT_FILES_DIR", tmp_path)
+    directory = "ab/cd"
+    stored_name = "obj.o"
+    _write_project_blob(tmp_path, directory, stored_name, b"x" * 40)
+    uid = _insert_binary_node(
+        "sys-prune-proj-1", directory, stored_name, deleted_at="2020-01-01T00:00:00+00:00"
+    )
+
+    removed, freed = pf.purge_soft_deleted_project_files()
+
+    assert removed >= 1
+    assert freed == 40
+    assert pf._table().find_one(uid=uid) is None
+    assert not (tmp_path / directory / stored_name).exists()
+
+
+def test_purge_soft_deleted_project_files_dry_run_changes_nothing(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr(pf, "PROJECT_FILES_DIR", tmp_path)
+    directory = "ab/cd"
+    stored_name = "obj2.o"
+    _write_project_blob(tmp_path, directory, stored_name, b"x" * 15)
+    uid = _insert_binary_node(
+        "sys-prune-proj-2", directory, stored_name, deleted_at="2020-01-01T00:00:00+00:00"
+    )
+
+    removed, freed = pf.purge_soft_deleted_project_files(dry_run=True)
+
+    assert removed >= 1
+    assert freed == 15
+    assert pf._table().find_one(uid=uid) is not None
+    assert (tmp_path / directory / stored_name).exists()
+
+
+def test_purge_soft_deleted_project_files_ignores_live_and_text_rows(
+    local_db, tmp_path, monkeypatch
+):
+    monkeypatch.setattr(pf, "PROJECT_FILES_DIR", tmp_path)
+    directory = "ab/cd"
+    stored_name = "live.o"
+    _write_project_blob(tmp_path, directory, stored_name)
+    live_uid = _insert_binary_node("sys-prune-proj-3", directory, stored_name, deleted_at=None)
+
+    pf.purge_soft_deleted_project_files()
+
+    assert pf._table().find_one(uid=live_uid) is not None
+    assert (tmp_path / directory / stored_name).exists()
+
+
+def test_sweep_orphan_project_file_blobs_removes_unreferenced_file(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr(pf, "PROJECT_FILES_DIR", tmp_path)
+    directory = "ab/cd"
+    orphan_name = "leftover.right"
+    _write_project_blob(tmp_path, directory, orphan_name, b"q" * 60)
+
+    removed, freed = pf.sweep_orphan_project_file_blobs()
+
+    assert removed == 1
+    assert freed == 60
+    assert not (tmp_path / directory / orphan_name).exists()
+
+
+def test_sweep_orphan_project_file_blobs_keeps_referenced_file(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr(pf, "PROJECT_FILES_DIR", tmp_path)
+    directory = "ab/cd"
+    stored_name = "kept.bin"
+    _write_project_blob(tmp_path, directory, stored_name)
+    _insert_binary_node("sys-prune-proj-4", directory, stored_name, deleted_at=None)
+
+    removed, freed = pf.sweep_orphan_project_file_blobs()
+
+    assert removed == 0
+    assert freed == 0
+    assert (tmp_path / directory / stored_name).exists()
+
+
+def test_sweep_orphan_project_file_blobs_keeps_blob_referenced_only_by_soft_deleted_row(
+    local_db, tmp_path, monkeypatch
+):
+    monkeypatch.setattr(pf, "PROJECT_FILES_DIR", tmp_path)
+    directory = "ab/cd"
+    stored_name = "still-soft-deleted.bin"
+    _write_project_blob(tmp_path, directory, stored_name)
+    _insert_binary_node(
+        "sys-prune-proj-5", directory, stored_name, deleted_at="2020-01-01T00:00:00+00:00"
+    )
+
+    removed, freed = pf.sweep_orphan_project_file_blobs()
+
+    assert removed == 0
+    assert freed == 0
+    assert (tmp_path / directory / stored_name).exists()
+
+
+def test_sweep_orphan_project_file_blobs_dry_run_changes_nothing(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr(pf, "PROJECT_FILES_DIR", tmp_path)
+    directory = "ab/cd"
+    orphan_name = "dry.right"
+    _write_project_blob(tmp_path, directory, orphan_name, b"w" * 12)
+
+    removed, freed = pf.sweep_orphan_project_file_blobs(dry_run=True)
+
+    assert removed == 1
+    assert freed == 12
+    assert (tmp_path / directory / orphan_name).exists()

tests/unit/services/containers/api.py

@@ -570,6 +570,85 @@ def test_bidirectional_sync_manifest_is_pruned_after_both_sides_agree(env, tmp_p
     assert remaining == []
 
 
+def test_bidirectional_sync_skips_build_artifact_directories(env, tmp_path):
+    workspace = tmp_path / "sync-build-ws"
+    workspace.mkdir()
+    pid = env["project"]["uid"]
+    user = env["user"]
+    (workspace / "build").mkdir()
+    (workspace / "build" / "output.bin").write_text("junk")
+    (workspace / "dist").mkdir()
+    (workspace / "dist" / "bundle.js").write_text("junk")
+    (workspace / "real.txt").write_text("keep me\n")
+
+    counts = project_files.sync_dir_bidirectional(pid, str(workspace), user)
+
+    assert counts["imported"] == 1
+    assert project_files.get_node(pid, "real.txt") is not None
+    assert project_files.get_node(pid, "build/output.bin") is None
+    assert project_files.get_node(pid, "dist/bundle.js") is None
+
+
+def test_bidirectional_sync_skips_compiled_artifact_extensions(env, tmp_path):
+    workspace = tmp_path / "sync-ext-ws"
+    workspace.mkdir()
+    pid = env["project"]["uid"]
+    user = env["user"]
+    (workspace / "main.o").write_bytes(b"junk")
+    (workspace / "helper.pyc").write_bytes(b"junk")
+    (workspace / "main.c").write_text("int main() { return 0; }\n")
+
+    counts = project_files.sync_dir_bidirectional(pid, str(workspace), user)
+
+    assert counts["imported"] == 1
+    assert project_files.get_node(pid, "main.c") is not None
+    assert project_files.get_node(pid, "main.o") is None
+    assert project_files.get_node(pid, "helper.pyc") is None
+
+
+def test_sync_dir_bidirectional_locked_skips_concurrent_run_for_same_project(
+    env, tmp_path
+):
+    workspace = tmp_path / "sync-lock-ws"
+    workspace.mkdir()
+    pid = env["project"]["uid"]
+    user = env["user"]
+
+    lock = api._sync_lock_for(pid)
+    lock.acquire()
+    try:
+        counts = api._sync_dir_bidirectional_locked(pid, str(workspace), user)
+    finally:
+        lock.release()
+
+    assert counts == {
+        "exported": 0,
+        "imported": 0,
+        "deleted_in_project": 0,
+        "deleted_in_workspace": 0,
+    }
+    assert project_files.get_node(pid, "real.txt") is None
+
+
+def test_sync_dir_bidirectional_locked_runs_when_lock_is_free(env, tmp_path):
+    workspace = tmp_path / "sync-lock-free-ws"
+    workspace.mkdir()
+    pid = env["project"]["uid"]
+    user = env["user"]
+    (workspace / "real.txt").write_text("keep me\n")
+
+    counts = api._sync_dir_bidirectional_locked(pid, str(workspace), user)
+
+    assert counts["imported"] == 1
+    assert project_files.get_node(pid, "real.txt") is not None
+
+
+def test_sync_dir_bidirectional_locked_is_per_project(env, tmp_path):
+    other_pid = "sync-lock-other-project"
+    lock = api._sync_lock_for(other_pid)
+    assert lock is not api._sync_lock_for(env["project"]["uid"])
+
+
 
 def _proxy_scope(kind: str, headers: dict, scheme: str, query: str = "") -> dict:
     return {

tests/unit/services/containers/store.py

@@ -0,0 +1,76 @@
+# retoor <retoor@molodetz.nl>
+
+import pytest
+
+from devplacepy.database import get_table, init_db
+from devplacepy.services.containers import store
+from devplacepy.utils import generate_uid
+
+
+@pytest.fixture(autouse=True)
+def _init_db_containers_store():
+    init_db()
+    yield
+
+
+def _make_instance(project_uid, deleted_at=None):
+    uid = generate_uid()
+    get_table("instances").insert(
+        {
+            "uid": uid,
+            "project_uid": project_uid,
+            "deleted_at": deleted_at,
+            "deleted_by": None,
+            "name": f"inst-{uid[:8]}",
+            "status": "running",
+            "desired_state": "running",
+        }
+    )
+    return uid
+
+
+def test_gc_workspaces_removes_directories_with_no_live_instance(
+    local_db, tmp_path, monkeypatch
+):
+    monkeypatch.setattr("devplacepy.config.CONTAINER_WORKSPACES_DIR", tmp_path)
+    active_project = "gcws-active-project"
+    orphan_project = "gcws-orphan-project"
+    (tmp_path / active_project).mkdir()
+    (tmp_path / orphan_project).mkdir()
+    _make_instance(active_project)
+
+    removed = store.gc_workspaces()
+
+    assert removed == 1
+    assert (tmp_path / active_project).exists()
+    assert not (tmp_path / orphan_project).exists()
+
+
+def test_gc_workspaces_ignores_soft_deleted_instances(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr("devplacepy.config.CONTAINER_WORKSPACES_DIR", tmp_path)
+    project = "gcws-soft-deleted-project"
+    (tmp_path / project).mkdir()
+    _make_instance(project, deleted_at="2020-01-01T00:00:00+00:00")
+
+    removed = store.gc_workspaces()
+
+    assert removed == 1
+    assert not (tmp_path / project).exists()
+
+
+def test_gc_workspaces_dry_run_changes_nothing(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr("devplacepy.config.CONTAINER_WORKSPACES_DIR", tmp_path)
+    orphan_project = "gcws-dry-run-project"
+    (tmp_path / orphan_project).mkdir()
+
+    removed = store.gc_workspaces(dry_run=True)
+
+    assert removed == 1
+    assert (tmp_path / orphan_project).exists()
+
+
+def test_gc_workspaces_missing_base_dir_is_a_noop(local_db, tmp_path, monkeypatch):
+    monkeypatch.setattr(
+        "devplacepy.config.CONTAINER_WORKSPACES_DIR", tmp_path / "does-not-exist"
+    )
+    assert store.gc_workspaces() == 0