Source code for continuity.core.schema

"""
Schema management for Continuity project structure.
"""

from pathlib import Path
from typing import Any, Dict, List, Tuple

from .._version import __version__, __version_tuple__


[docs] class ContinuitySchema: """Manages .continuity directory structure and validation.""" CURRENT_VERSION: object = __version_tuple__ CURRENT_VERSION_STR: str = __version__ # .continuity folder structure REQUIRED_DIRECTORIES = [ "agent/mailbox", "agent/mailbox/read", "agent/mailbox/archive", "human/mailbox", "human/mailbox/read", "human/mailbox/archive", "inbox", "inbox/archive", "tmp", ] REQUIRED_FILES = [ "config", "CLI_REFERENCE.md", "AI_AGENT_REFERENCE.md", "ABOUT.md", ] def __init__(self, storage_path: Path): self.storage_path = Path(storage_path).expanduser()
[docs] def get_current_version(self) -> Tuple[int, int, int, str, str]: """Get the current schema version.""" return self.CURRENT_VERSION # type: ignore[return-value]
[docs] def validate_structure(self) -> Dict[str, Any]: """Validate the directory structure.""" issues = [] for dir_path in self.REQUIRED_DIRECTORIES: if not (self.storage_path / dir_path).exists(): issues.append(f"Missing required directory: {dir_path}") for file_path in self.REQUIRED_FILES: if not (self.storage_path / file_path).exists(): issues.append(f"Missing required file: {file_path}") return { "version": f"v{self.CURRENT_VERSION_STR}", "is_valid": len(issues) == 0, "issues": issues, "requires_init": not self.storage_path.exists() or len(issues) > 0, }
[docs] def create_structure(self) -> List[str]: """Create the basic directory structure.""" created_items = [] for dir_path in self.REQUIRED_DIRECTORIES: full_path = self.storage_path / dir_path if not full_path.exists(): full_path.mkdir(parents=True, exist_ok=True) created_items.append(f"Directory: {dir_path}") return created_items
[docs] def get_status(self) -> Dict[str, Any]: """Get schema status information.""" validation = self.validate_structure() return { "current_version": f"v{self.CURRENT_VERSION_STR}", "latest_version": f"v{self.CURRENT_VERSION_STR}", "is_current": True, "is_valid": validation["is_valid"], "issues": validation["issues"], "requires_init": validation["requires_init"], }