ros/autocoder
1
0
Fork 0
mirror of https://github.com/leonvanzyl/autocoder.git synced 2026-01-30 06:12:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: Replace JSON registry with SQLite database

Browse Source 
Replace the JSON-based project registry with SQLite using SQLAlchemy ORM
for improved reliability, concurrency handling, and simpler code.

Changes:
- registry.py: Complete rewrite from JSON to SQLite storage
  - Remove manual file locking (RegistryLock class with msvcrt/fcntl)
  - Remove atomic write logic (temp file + rename pattern)
  - Remove backup/recovery logic for corrupted JSON
  - Add SQLAlchemy Project model with name, path, created_at columns
  - Add singleton database engine pattern (_get_engine)
  - Add session context manager (_get_session) for transactions
  - Simplify from 493 to 367 lines of code

- CLAUDE.md: Update documentation to reflect new storage

Storage location change:
- Before: Platform-specific paths
  - Windows: %APPDATA%\autonomous-coder\projects.json
  - macOS: ~/Library/Application Support/autonomous-coder/projects.json
  - Linux: ~/.config/autonomous-coder/projects.json
- After: Unified path for all platforms
  - ~/.autocoder/registry.db

Benefits:
- SQLite handles concurrency automatically (no manual locking needed)
- Atomic transactions built-in (no temp file + rename dance)
- Crash-safe by default (no backup/recovery code needed)
- Consistent with existing features.db pattern in api/database.py
- Data persists across app updates (stored in user home, not project)

All public API functions maintain their existing signatures, so no
changes are required in the 7 consumer files (start.py,
autonomous_agent_demo.py, and 5 server routers).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Auto
2026-01-02 11:22:51 +02:00
parent e8f3b99a42
commit 81b1b29f24
2 changed files with 140 additions and 268 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
CLAUDE.md
View File

@@ -98,15 +98,13 @@ npm run lint # Run ESLint
### Project Registry ### Project Registry
Projects can be stored in any directory. The registry (`projects.json`) maps project names to paths: Projects can be stored in any directory. The registry maps project names to paths using SQLite:
- **Windows**: `%APPDATA%\autonomous-coder\projects.json` - **All platforms**: `~/.autocoder/registry.db`
- **macOS**: `~/Library/Application Support/autonomous-coder/projects.json`
- **Linux**: `~/.config/autonomous-coder/projects.json`
The registry uses: The registry uses:
- SQLite database with SQLAlchemy ORM
- POSIX path format (forward slashes) for cross-platform compatibility - POSIX path format (forward slashes) for cross-platform compatibility
- File locking for concurrent access safety - SQLite's built-in transaction handling for concurrency safety
- Atomic writes (temp file + rename) to prevent corruption
### Server API (server/) ### Server API (server/)
@@ -146,7 +144,7 @@ MCP tools available to the agent:
### Project Structure for Generated Apps ### Project Structure for Generated Apps
Projects can be stored in any directory (registered in `projects.json`). Each project contains: Projects can be stored in any directory (registered in `~/.autocoder/registry.db`). Each project contains:
- `prompts/app_spec.txt` - Application specification (XML format) - `prompts/app_spec.txt` - Application specification (XML format)
- `prompts/initializer_prompt.md` - First session prompt - `prompts/initializer_prompt.md` - First session prompt
- `prompts/coding_prompt.md` - Continuation session prompt - `prompts/coding_prompt.md` - Continuation session prompt

396
registry.py
View File

@@ -3,21 +3,21 @@ Project Registry Module
======================= =======================
Cross-platform project registry for storing project name to path mappings. Cross-platform project registry for storing project name to path mappings.
Supports Windows, macOS, and Linux with platform-specific config directories. Uses SQLite database stored at ~/.autocoder/registry.db.
""" """
import json
import logging import logging
import os import os
import stat import re
import sys from contextlib import contextmanager
import tempfile
import shutil
import time
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from sqlalchemy import Column, String, DateTime, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# Module logger # Module logger
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -37,7 +37,7 @@ class RegistryNotFound(RegistryError):
class RegistryCorrupted(RegistryError): class RegistryCorrupted(RegistryError):
"""Registry JSON is malformed.""" """Registry database is corrupted."""
pass pass
@@ -47,225 +47,85 @@ class RegistryPermissionDenied(RegistryError):
# ============================================================================= # =============================================================================
# Registry Lock (Cross-Platform) # SQLAlchemy Model
# ============================================================================= # =============================================================================
class RegistryLock: Base = declarative_base()
"""
Context manager for registry file locking.
Uses fcntl on Unix and msvcrt on Windows.
"""
def __init__(self, registry_path: Path):
self.registry_path = registry_path
self.lock_path = registry_path.with_suffix('.lock')
self._file = None
def __enter__(self): class Project(Base):
self.lock_path.parent.mkdir(parents=True, exist_ok=True) """SQLAlchemy model for registered projects."""
self._file = open(self.lock_path, 'w') __tablename__ = "projects"
try: name = Column(String(50), primary_key=True, index=True)
if sys.platform == "win32": path = Column(String, nullable=False) # POSIX format for cross-platform
import msvcrt created_at = Column(DateTime, nullable=False)
# Windows: msvcrt.LK_NBLCK is non-blocking, so we retry with backoff
max_attempts = 10
for attempt in range(max_attempts):
try:
msvcrt.locking(self._file.fileno(), msvcrt.LK_NBLCK, 1)
break # Lock acquired
except OSError:
if attempt == max_attempts - 1:
raise # Give up after max attempts
time.sleep(0.1 * (attempt + 1)) # Exponential backoff
else:
import fcntl
fcntl.flock(self._file.fileno(), fcntl.LOCK_EX)
except Exception as e:
self._file.close()
raise RegistryError(f"Could not acquire registry lock: {e}") from e
return self
def __exit__(self, *args):
if self._file:
try:
if sys.platform != "win32":
import fcntl
fcntl.flock(self._file.fileno(), fcntl.LOCK_UN)
finally:
self._file.close()
try:
self.lock_path.unlink(missing_ok=True)
except Exception:
pass
# ============================================================================= # =============================================================================
# Registry Path Functions # Database Connection
# ============================================================================= # =============================================================================
# Module-level singleton for database engine
_engine = None
_SessionLocal = None
def get_config_dir() -> Path: def get_config_dir() -> Path:
""" """
Get the platform-specific config directory for the application. Get the config directory: ~/.autocoder/
Returns: Returns:
- Windows: %APPDATA%/autonomous-coder/ Path to ~/.autocoder/ (created if it doesn't exist)
- macOS: ~/Library/Application Support/autonomous-coder/
- Linux: ~/.config/autonomous-coder/ (or $XDG_CONFIG_HOME)
""" """
if sys.platform == "win32": config_dir = Path.home() / ".autocoder"
base = Path(os.getenv("APPDATA", Path.home() / "AppData" / "Roaming"))
elif sys.platform == "darwin":
base = Path.home() / "Library" / "Application Support"
else: # Linux and other Unix-like
base = Path(os.getenv("XDG_CONFIG_HOME", Path.home() / ".config"))
config_dir = base / "autonomous-coder"
config_dir.mkdir(parents=True, exist_ok=True) config_dir.mkdir(parents=True, exist_ok=True)
return config_dir return config_dir
def get_registry_path() -> Path: def get_registry_path() -> Path:
"""Get the path to the projects registry file.""" """Get the path to the registry database."""
return get_config_dir() / "projects.json" return get_config_dir() / "registry.db"
# ============================================================================= def _get_engine():
# Registry I/O Functions
# =============================================================================
def _create_empty_registry() -> dict[str, Any]:
"""Create a new empty registry structure."""
return {
"version": 1,
"created_at": datetime.now().isoformat(),
"projects": {}
}
def load_registry(create_if_missing: bool = True) -> dict[str, Any]:
""" """
Load the registry from disk. Get or create the database engine (singleton pattern).
Args:
create_if_missing: If True, create a new registry if none exists.
Returns: Returns:
The registry dictionary. Tuple of (engine, SessionLocal)
Raises:
RegistryNotFound: If registry doesn't exist and create_if_missing is False.
RegistryCorrupted: If registry JSON is malformed.
RegistryPermissionDenied: If can't read the registry file.
""" """
registry_path = get_registry_path() global _engine, _SessionLocal
# Case 1: File doesn't exist if _engine is None:
if not registry_path.exists(): db_path = get_registry_path()
if create_if_missing: db_url = f"sqlite:///{db_path.as_posix()}"
registry = _create_empty_registry() _engine = create_engine(db_url, connect_args={"check_same_thread": False})
save_registry(registry) Base.metadata.create_all(bind=_engine)
return registry _SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=_engine)
else: logger.debug("Initialized registry database at: %s", db_path)
raise RegistryNotFound(f"Registry not found: {registry_path}")
# Case 2: Read the file return _engine, _SessionLocal
try:
content = registry_path.read_text(encoding='utf-8')
except PermissionError as e:
raise RegistryPermissionDenied(f"Cannot read registry: {e}") from e
except OSError as e:
raise RegistryError(f"Error reading registry: {e}") from e
# Case 3: Parse JSON
try:
data = json.loads(content)
except json.JSONDecodeError as e:
# Try to recover from backup
backup_path = registry_path.with_suffix('.json.backup')
logger.warning("Registry corrupted, attempting recovery from backup: %s", backup_path)
if backup_path.exists():
try:
backup_content = backup_path.read_text(encoding='utf-8')
data = json.loads(backup_content)
# Restore from backup
shutil.copy2(backup_path, registry_path)
logger.info("Successfully recovered registry from backup")
return data
except Exception as recovery_error:
logger.error("Failed to recover from backup: %s", recovery_error)
raise RegistryCorrupted(
f"Registry corrupted: {e}\nBackup location: {backup_path}"
) from e
# Ensure required structure
if "projects" not in data:
data["projects"] = {}
if "version" not in data:
data["version"] = 1
return data
def save_registry(registry: dict[str, Any]) -> None: @contextmanager
def _get_session():
""" """
Save the registry to disk atomically. Context manager for database sessions with automatic commit/rollback.
Uses temp file + rename for atomic writes to prevent corruption. Yields:
SQLAlchemy session
Args:
registry: The registry dictionary to save.
Raises:
RegistryPermissionDenied: If can't write to the registry.
RegistryError: If write fails for other reasons.
""" """
registry_path = get_registry_path() _, SessionLocal = _get_engine()
registry_path.parent.mkdir(parents=True, exist_ok=True) session = SessionLocal()
# Create backup before modification (if file exists)
if registry_path.exists():
backup_path = registry_path.with_suffix('.json.backup')
try:
shutil.copy2(registry_path, backup_path)
except Exception as e:
logger.warning("Failed to create registry backup: %s", e)
# Write to temp file in same directory (ensures same filesystem for atomic rename)
# On Windows, we must close the file before renaming it
tmp_path = None
try: try:
# Create temp file yield session
fd, tmp_name = tempfile.mkstemp(suffix='.json', dir=registry_path.parent) session.commit()
tmp_path = Path(tmp_name) except Exception:
session.rollback()
try: raise
# Write content finally:
with os.fdopen(fd, 'w', encoding='utf-8') as tmp_file: session.close()
json.dump(registry, tmp_file, indent=2)
tmp_file.flush()
os.fsync(tmp_file.fileno())
# File is now closed, safe to rename on Windows
# Atomic rename
tmp_path.replace(registry_path)
# Set restrictive permissions (owner read/write only)
# On Windows, this is a best-effort operation
try:
if sys.platform != "win32":
registry_path.chmod(stat.S_IRUSR | stat.S_IWUSR) # 0o600
except Exception:
pass # Best effort - don't fail if permissions can't be set
except Exception:
if tmp_path and tmp_path.exists():
tmp_path.unlink(missing_ok=True)
raise
except PermissionError as e:
raise RegistryPermissionDenied(f"Cannot write registry: {e}") from e
except OSError as e:
raise RegistryError(f"Failed to write registry: {e}") from e
# ============================================================================= # =============================================================================
@@ -285,7 +145,6 @@ def register_project(name: str, path: Path) -> None:
RegistryError: If a project with that name already exists. RegistryError: If a project with that name already exists.
""" """
# Validate name # Validate name
import re
if not re.match(r'^[a-zA-Z0-9_-]{1,50}$', name): if not re.match(r'^[a-zA-Z0-9_-]{1,50}$', name):
raise ValueError( raise ValueError(
"Invalid project name. Use only letters, numbers, hyphens, " "Invalid project name. Use only letters, numbers, hyphens, "
@@ -295,21 +154,20 @@ def register_project(name: str, path: Path) -> None:
# Ensure path is absolute # Ensure path is absolute
path = Path(path).resolve() path = Path(path).resolve()
with RegistryLock(get_registry_path()): with _get_session() as session:
registry = load_registry() existing = session.query(Project).filter(Project.name == name).first()
if existing:
if name in registry["projects"]:
logger.warning("Attempted to register duplicate project: %s", name) logger.warning("Attempted to register duplicate project: %s", name)
raise RegistryError(f"Project '{name}' already exists in registry") raise RegistryError(f"Project '{name}' already exists in registry")
# Store path as POSIX format (forward slashes) for cross-platform consistency project = Project(
registry["projects"][name] = { name=name,
"path": path.as_posix(), path=path.as_posix(),
"created_at": datetime.now().isoformat() created_at=datetime.now()
} )
session.add(project)
save_registry(registry) logger.info("Registered project '%s' at path: %s", name, path)
logger.info("Registered project '%s' at path: %s", name, path)
def unregister_project(name: str) -> bool: def unregister_project(name: str) -> bool:
@@ -322,17 +180,16 @@ def unregister_project(name: str) -> bool:
Returns: Returns:
True if removed, False if project wasn't found. True if removed, False if project wasn't found.
""" """
with RegistryLock(get_registry_path()): with _get_session() as session:
registry = load_registry() project = session.query(Project).filter(Project.name == name).first()
if not project:
if name not in registry["projects"]:
logger.debug("Attempted to unregister non-existent project: %s", name) logger.debug("Attempted to unregister non-existent project: %s", name)
return False return False
del registry["projects"][name] session.delete(project)
save_registry(registry)
logger.info("Unregistered project: %s", name) logger.info("Unregistered project: %s", name)
return True return True
def get_project_path(name: str) -> Path | None: def get_project_path(name: str) -> Path | None:
@@ -345,14 +202,15 @@ def get_project_path(name: str) -> Path | None:
Returns: Returns:
The project Path, or None if not found. The project Path, or None if not found.
""" """
registry = load_registry() _, SessionLocal = _get_engine()
project = registry["projects"].get(name) session = SessionLocal()
try:
if project is None: project = session.query(Project).filter(Project.name == name).first()
return None if project is None:
return None
# Convert POSIX path string back to Path object return Path(project.path)
return Path(project["path"]) finally:
session.close()
def list_registered_projects() -> dict[str, dict[str, Any]]: def list_registered_projects() -> dict[str, dict[str, Any]]:
@@ -362,8 +220,19 @@ def list_registered_projects() -> dict[str, dict[str, Any]]:
Returns: Returns:
Dictionary mapping project names to their info dictionaries. Dictionary mapping project names to their info dictionaries.
""" """
registry = load_registry() _, SessionLocal = _get_engine()
return registry.get("projects", {}) session = SessionLocal()
try:
projects = session.query(Project).all()
return {
p.name: {
"path": p.path,
"created_at": p.created_at.isoformat() if p.created_at else None
}
for p in projects
}
finally:
session.close()
def get_project_info(name: str) -> dict[str, Any] | None: def get_project_info(name: str) -> dict[str, Any] | None:
@@ -376,8 +245,18 @@ def get_project_info(name: str) -> dict[str, Any] | None:
Returns: Returns:
Project info dictionary, or None if not found. Project info dictionary, or None if not found.
""" """
registry = load_registry() _, SessionLocal = _get_engine()
return registry["projects"].get(name) session = SessionLocal()
try:
project = session.query(Project).filter(Project.name == name).first()
if project is None:
return None
return {
"path": project.path,
"created_at": project.created_at.isoformat() if project.created_at else None
}
finally:
session.close()
def update_project_path(name: str, new_path: Path) -> bool: def update_project_path(name: str, new_path: Path) -> bool:
@@ -393,15 +272,14 @@ def update_project_path(name: str, new_path: Path) -> bool:
""" """
new_path = Path(new_path).resolve() new_path = Path(new_path).resolve()
with RegistryLock(get_registry_path()): with _get_session() as session:
registry = load_registry() project = session.query(Project).filter(Project.name == name).first()
if not project:
if name not in registry["projects"]:
return False return False
registry["projects"][name]["path"] = new_path.as_posix() project.path = new_path.as_posix()
save_registry(registry)
return True return True
# ============================================================================= # =============================================================================
@@ -448,22 +326,16 @@ def cleanup_stale_projects() -> list[str]:
""" """
removed = [] removed = []
with RegistryLock(get_registry_path()): with _get_session() as session:
registry = load_registry() projects = session.query(Project).all()
projects = registry.get("projects", {}) for project in projects:
path = Path(project.path)
stale_names = []
for name, info in projects.items():
path = Path(info["path"])
if not path.exists(): if not path.exists():
stale_names.append(name) session.delete(project)
removed.append(project.name)
for name in stale_names: if removed:
del projects[name] logger.info("Cleaned up stale projects: %s", removed)
removed.append(name)
if removed:
save_registry(registry)
return removed return removed
@@ -475,18 +347,20 @@ def list_valid_projects() -> list[dict[str, Any]]:
Returns: Returns:
List of project info dicts with additional 'name' field. List of project info dicts with additional 'name' field.
""" """
registry = load_registry() _, SessionLocal = _get_engine()
projects = registry.get("projects", {}) session = SessionLocal()
try:
valid = [] projects = session.query(Project).all()
for name, info in projects.items(): valid = []
path = Path(info["path"]) for p in projects:
is_valid, _ = validate_project_path(path) path = Path(p.path)
if is_valid: is_valid, _ = validate_project_path(path)
valid.append({ if is_valid:
"name": name, valid.append({
"path": info["path"], "name": p.name,
"created_at": info.get("created_at") "path": p.path,
}) "created_at": p.created_at.isoformat() if p.created_at else None
})
return valid return valid
finally:
session.close()