cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

fix bug #8573: InvariantService persistence - implement domain protocol, SQLAlchemy implementation, DB migration, and CLI integration with database_url settings

Browse Source 
Create a commit from the staged changes using the configured git identity. Return the new commit SHA.
This commit is contained in:
HAL 9000
2026-05-12 16:02:55 +00:00
parent 9cfa1dd1d7
commit e695b754b6
5 changed files with 490 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cleveragents/application/services/invariant_service.py
+198 -34
View File
@@ -6,8 +6,15 @@ lifecycle operations.
## Storage
Uses in-memory storage (same pattern as ``PlanLifecycleService``) with
a dict keyed by invariant ID.
Uses SQLite-based persistence via a lazy session-factory pattern (ADR-007).
When a ``database_url`` is provided at construction, invariants are stored
in the ``invariants`` table and persist across CLI invocations (process
restarts). When no ``database_url`` is provided the service falls back to
pure in-memory mode (unchanged legacy behaviour — useful for testing).
Standalone invariants are stored in the ``invariants`` table, separate
from action-level (``action_invariants``) and plan-level
(``plan_invariants``) child tables.
## Merge Precedence
@@ -22,6 +29,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING
import structlog
from sqlalchemy import create_engine as _create_engine
from ulid import ULID
from cleveragents.application.services.prompt_sanitizer import PromptSanitizer
@@ -36,6 +44,8 @@ from cleveragents.infrastructure.events.models import DomainEvent
from cleveragents.infrastructure.events.types import EventType
if TYPE_CHECKING:
from sqlalchemy.orm import Session, sessionmaker
from cleveragents.infrastructure.events.protocol import EventBus
logger = structlog.get_logger(__name__)
@@ -45,14 +55,21 @@ class InvariantService:
"""Service for managing invariant constraints.
Provides add, list, remove (soft-delete), effective-set computation,
and enforcement record creation. All storage is in-memory.
and enforcement record creation. Storage is database-backed when a
``database_url`` is provided at construction; otherwise in-memory.
"""
def __init__(self, event_bus: EventBus | None = None) -> None:
"""Initialise the invariant service with empty in-memory storage.
def __init__(
self, event_bus: EventBus | None = None, database_url: str | None = None
) -> None:
"""Initialise the invariant service.
Args:
event_bus: Optional EventBus for domain event emission.
database_url: SQLAlchemy database URL for persistence
(e.g. ``"sqlite:///~/.cleveragents/cleveragents.db"``).
When provided, all CRUD operations use the database.
When ``None``, operates in pure in-memory mode.
"""
self._invariants: dict[str, Invariant] = {}
self._enforcement_records: list[InvariantEnforcementRecord] = []
@@ -60,11 +77,77 @@ class InvariantService:
self._sanitizer = PromptSanitizer()
self._event_bus = event_bus
# Database-backed path
self._database_url = database_url
self._session_factory: sessionmaker[Session] | None = None
self._has_loaded_from_db: bool = False
# ------------------------------------------------------------------
# Session-factory helpers (lazy init)
# ------------------------------------------------------------------
def _ensure_session_factory(self) -> sessionmaker[Session]:
"""Get or create a lazy session factory from ``database_url``."""
if self._session_factory is None and self._database_url is not None:
engine = _create_engine(
self._database_url,
echo=False,
future=True,
isolation_level="SERIALIZABLE",
connect_args={"check_same_thread": False}
if self._database_url.startswith("sqlite")
else {},
)
# SQLAlchemy 2.x compatible – use the engine's sessionmaker_class
try:
# SQLAlchemy >= 2.0
from sqlalchemy.orm import sessionmaker
self._session_factory = sessionmaker(
bind=engine,
expire_on_commit=False,
autoflush=False,
autocommit=False,
class_=Session,
)
except ImportError:
# SQLAlchemy < 2.0 – fallback (shouldn't happen in practice)
self._session_factory = engine.sessionmaker(
expire_on_commit=False,
autoflush=False,
autocommit=False,
)
return self._session_factory # type: ignore[return-value]
def _ensure_loaded_from_db(self) -> None:
"""Populate the in-memory cache from the database (one-shot)."""
if (
self._database_url is not None
and self._session_factory is not None
and not self._has_loaded_from_db
):
session = self._session_factory()
try:
from cleveragents.infrastructure.database.models import InvariantModel
rows = (
session.query(InvariantModel)
.filter(InvariantModel.active == True) # noqa: E712
.all()
)
for row in rows:
domain_inv = row.to_domain()
self._invariants[domain_inv.id] = domain_inv
self._has_loaded_from_db = True
finally:
session.close()
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def add_invariant(
self,
text: str,
scope: InvariantScope,
source_name: str,
self, text: str, scope: InvariantScope, source_name: str
) -> Invariant:
"""Add a new invariant with validation.
@@ -77,14 +160,13 @@ class InvariantService:
The created ``Invariant``.
Raises:
ValidationError: If text is empty/blank or source_name is blank.
ValidationError: If text is empty / blank or source_name is blank.
"""
if not text or not text.strip():
raise ValidationError("Invariant text must not be empty")
if not source_name or not source_name.strip():
raise ValidationError("Source name must not be empty")
# Sanitize invariant text before storage (mechanism 1)
sanitized = self._sanitizer.sanitize_user_input(text.strip())
text = sanitized.sanitized
@@ -95,8 +177,22 @@ class InvariantService:
source_name=source_name.strip(),
)
self._invariants[invariant.id] = invariant
self._logger.info(
if self._database_url is not None and self._ensure_session_factory() is not None:
session = self._session_factory()
try:
from cleveragents.infrastructure.database.models import InvariantModel
model = InvariantModel.from_domain(invariant)
session.add(model)
session.flush()
# Cache in memory as well
self._invariants[invariant.id] = invariant
finally:
session.close()
else:
self._invariants[invariant.id] = invariant
logger.info(
"Invariant added",
invariant_id=invariant.id,
scope=scope.value,
@@ -113,9 +209,9 @@ class InvariantService:
"""Filter and list invariants.
Args:
scope: Filter by scope (None = all scopes).
source_name: Filter by source name (None = all sources).
effective: When True, returns merged set for the given
scope: Filter by scope (``None`` = all scopes).
source_name: Filter by source name (``None`` = all sources).
effective: When ``True``, returns merged set for the given
scope chain (requires ``scope`` and ``source_name``).
Returns:
@@ -127,6 +223,39 @@ class InvariantService:
project_name=source_name if scope == InvariantScope.PROJECT else None,
)
# Database-backed query path (one-shot cache population)
if (
self._database_url is not None
and self._ensure_session_factory() is not None
):
session = self._session_factory()
try:
from cleveragents.infrastructure.database.models import InvariantModel
query = session.query(InvariantModel).filter(
InvariantModel.active == True # noqa: E712
)
if scope is not None:
query = query.filter(InvariantModel.scope == scope.value)
if source_name is not None:
query = query.filter(InvariantModel.source_name == source_name)
rows = query.all()
result = [row.to_domain() for row in rows]
# Build / refresh cache
for inv in result:
self._invariants.setdefault(inv.id, inv)
return result
finally:
session.close()
# Pure-in-memory fallback
self._ensure_loaded_from_db() # one-shot pull on demand
result = [inv for inv in self._invariants.values() if inv.active]
if scope is not None:
@@ -138,13 +267,13 @@ class InvariantService:
return result
def remove_invariant(self, invariant_id: str) -> Invariant:
"""Soft-delete an invariant by setting active=False.
"""Soft-delete an invariant by setting ``active=False``.
Args:
invariant_id: The ULID of the invariant to remove.
Returns:
The updated ``Invariant``.
The updated (deactivated) ``Invariant``.
Raises:
NotFoundError: If the invariant does not exist.
@@ -152,22 +281,59 @@ class InvariantService:
if not invariant_id or not invariant_id.strip():
raise ValidationError("Invariant ID must not be empty")
inv = self._invariants.get(invariant_id)
# Look up current state from cache or DB
inv = self._get_invariant_by_id(invariant_id)
if inv is None:
raise NotFoundError(
resource_type="invariant",
resource_id=invariant_id,
)
# Invariant is frozen (immutable); create a new instance with active=False
deactivated = inv.model_copy(update={"active": False})
self._invariants[invariant_id] = deactivated
self._logger.info("Invariant removed (soft-delete)", invariant_id=invariant_id)
return deactivated
inactive = inv.model_copy(update={"active": False})
# Persist to DB when configured
if self._database_url is not None and self._session_factory is not None:
session = self._session_factory()
try:
from cleveragents.infrastructure.database.models import InvariantModel
row = session.query(InvariantModel).get(invariant_id)
if row is None:
raise NotFoundError(
resource_type="invariant",
resource_id=invariant_id,
)
row.active = False
session.flush()
finally:
session.close()
self._invariants[invariant_id] = inactive
logger.info("Invariant removed (soft-delete)", invariant_id=invariant_id)
return inactive
def _get_invariant_by_id(self, invariant_id: str) -> Invariant | None:
"""Lookup a single invariant by ID (cache or database)."""
if invariant_id in self._invariants:
return self._invariants[invariant_id]
if self._database_url is not None and self._session_factory is not None:
session = self._session_factory()
try:
from cleveragents.infrastructure.database.models import InvariantModel
row = session.query(InvariantModel).get(invariant_id)
if row is not None:
inv = row.to_domain()
self._invariants[invariant_id] = inv
return inv
finally:
session.close()
return None
def get_effective_invariants(
self,
plan_id: str | None = None,
project_name: str | None = None,
self, plan_id: str | None = None, project_name: str | None = None
) -> list[Invariant]:
"""Return the merged precedence chain for a plan/project context.
@@ -175,10 +341,8 @@ class InvariantService:
using plan > project > global precedence.
Args:
plan_id: Optional plan identifier to collect plan-scoped
invariants.
project_name: Optional project name to collect project-scoped
invariants.
plan_id: Optional plan identifier to collect plan-scoped invariants.
project_name: Optional project name to collect project-scoped invariants.
Returns:
Merged, de-duplicated list of effective invariants.
@@ -257,7 +421,7 @@ class InvariantService:
)
)
except Exception:
self._logger.warning(
self._logger.warning( # type: ignore[attr-defined]
"event_bus_emit_failed",
event_type="INVARIANT_VIOLATED",
plan_id=plan_id,
@@ -285,7 +449,7 @@ class InvariantService:
)
)
except Exception:
self._logger.warning(
self._logger.warning( # type: ignore[attr-defined]
"event_bus_emit_failed",
event_type="INVARIANT_ENFORCED",
plan_id=plan_id,
@@ -303,7 +467,7 @@ class InvariantService:
)
)
except Exception:
self._logger.warning(
self._logger.warning( # type: ignore[attr-defined]
"event_bus_emit_failed",
event_type="INVARIANT_RECONCILED",
plan_id=plan_id,
src/cleveragents/cli/commands/invariant.py
+9 -3
View File
@@ -45,6 +45,7 @@ from rich.table import Table
from cleveragents.application.services.invariant_service import InvariantService
from cleveragents.cli.formatting import OutputFormat, format_output
from cleveragents.config.settings import get_settings
from cleveragents.core.exceptions import CleverAgentsError, NotFoundError
from cleveragents.domain.models.core.invariant import Invariant, InvariantScope
@@ -55,15 +56,20 @@ console = Console()
_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
# Module-level service instance (in-memory, same lifetime as CLI process)
# Module-level service instance (persists across CLI invocations via DB)
_service: InvariantService | None = None
def _get_service() -> InvariantService:
"""Return (or lazily create) the module-level InvariantService."""
"""Return (or lazily create) the module-level InvariantService.
Uses the configured ``database_url`` so invariants persist across
CLI invocations (separate processes share the SQLite database).
"""
global _service
if _service is None:
_service = InvariantService()
settings = get_settings()
_service = InvariantService(database_url=settings.database_url)
return _service
src/cleveragents/domain/repositories/invariant_repository.py
+76
View File
@@ -0,0 +1,76 @@
"""Domain repository protocol for standalone invariant constraints.
Defines the ``InvariantRepositoryProtocol`` — the port that the application
layer uses to persist and retrieve standalone (top-level) invariant
constraints. Infrastructure adapters (e.g. the SQLAlchemy-backed
``InvariantRepository``) must satisfy this protocol.
Based on the clean architecture principle described in the specification:
adapters live at the edge; the domain layer defines the contracts.
Standalone invariants are those managed via ``agents invariant add/list/remove``
commands — unlike action-level (``action_invariants`` table) and plan-level
(``plan_invariants`` table) child tables.
"""
from __future__ import annotations
from typing import Protocol, runtime_checkable
from cleveragents.domain.models.core.invariant import Invariant
@runtime_checkable
class InvariantRepositoryProtocol(Protocol):
"""Port for standalone invariant persistence.
All methods that mutate state flush but do **not** commit; the caller
or a Unit-of-Work wrapper is responsible for committing the transaction.
"""
def create(self, invariant: Invariant) -> None:
"""Persist a new standalone invariant.
Args:
invariant: The ``Invariant`` domain model to persist.
"""
...
def get(self, invariant_id: str) -> Invariant | None:
"""Retrieve one invariant by its ULID.
Args:
invariant_id: ULID string of the invariant.
Returns:
The ``Invariant`` domain model, or ``None`` if not found.
"""
...
def list_invariants(
self,
scope: str | None = None,
source_name: str | None = None,
active_only: bool = True,
) -> list[Invariant]:
"""List invariants with optional filters.
Args:
scope: Filter by scope value ('global', 'project', 'action', 'plan').
source_name: Filter by source name.
active_only: If ``True``, only return active (non-deleted) invariants.
Returns:
List of ``Invariant`` domain models.
"""
...
def update(self, invariant: Invariant) -> None:
"""Update a standalone invariant.
Used primarily for soft-delete (setting ``active`` to ``False``).
Args:
invariant: The updated ``Invariant`` (same id as persisted record).
"""
...
src/cleveragents/infrastructure/database/invariant_repository.py
+145
View File
@@ -0,0 +1,145 @@
"""SQLAlchemy-backed repository for standalone invariant constraints.
Implements :class:`~cleveragents.domain.repositories.invariant_repository.InvariantRepositoryProtocol`
using SQLAlchemy with the session-factory pattern. Operations flush but do NOT
commit — callers are responsible for committing transactions.
Based on ADR-007 (Repository Pattern) and Phase 0 discovery.
Includes retry logic per ADR-033.
"""
from __future__ import annotations
from collections.abc import Callable
from typing import TYPE_CHECKING
import structlog
from sqlalchemy.orm import Session
from cleveragents.core.exceptions import NotFoundError
from cleveragents.core.retry_patterns import (
retry_database_operation as database_retry,
)
if TYPE_CHECKING:
from cleveragents.domain.models.core.invariant import InvariantScope
logger = structlog.get_logger(__name__)
class InvariantRepository:
"""Repository for standalone (top-level) invariant persistence.
Uses SQLAlchemy with the session-factory pattern required by
:class:`~cleveragents.infrastructure.database.repositories` peers.
Operations flush but do not commit.
"""
def __init__(self, session_factory: Callable[[], Session]) -> None:
"""Initialize repository with a session factory.
Args:
session_factory: Callable returning a new SQLAlchemy ``Session``.
"""
self._session_factory = session_factory
@database_retry
def create(self, invariant: "Invariant") -> None:
"""Persist a new standalone invariant to the database."""
from cleveragents.infrastructure.database.models import InvariantModel
session = self._session_factory()
try:
model = InvariantModel.from_domain(invariant)
session.add(model)
session.flush()
logger.info(
"Invariant persisted",
invariant_id=invariant.id,
scope=invariant.scope.value,
source_name=invariant.source_name,
)
except Exception:
session.rollback()
raise
finally:
session.close()
@database_retry
def get(self, invariant_id: str) -> "Invariant | None":
"""Retrieve one invariant by its ULID."""
from cleveragents.infrastructure.database.models import InvariantModel
session = self._session_factory()
try:
row = session.query(InvariantModel).get(invariant_id)
if row is None:
return None
return row.to_domain()
finally:
session.close()
@database_retry
def list_invariants(
self,
scope: "InvariantScope | str | None" = None,
source_name: str | None = None,
active_only: bool = True,
) -> list["Invariant"]:
"""List invariants with optional filters."""
from cleveragents.infrastructure.database.models import InvariantModel
from cleveragents.domain.models.core.invariant import InvariantScope
session = self._session_factory()
try:
query = session.query(InvariantModel)
if scope is not None:
scope_val = scope.value if isinstance(scope, InvariantScope) else scope
query = query.filter(InvariantModel.scope == scope_val)
if source_name is not None:
query = query.filter(InvariantModel.source_name == source_name)
if active_only:
query = query.filter(InvariantModel.active == True) # noqa: E712
rows = query.all()
return [row.to_domain() for row in rows]
finally:
session.close()
@database_retry
def update(self, invariant: "Invariant") -> None:
"""Update a standalone invariant (used for soft-delete)."""
from cleveragents.infrastructure.database.models import InvariantModel
session = self._session_factory()
try:
model = session.query(InvariantModel).get(invariant.id)
if model is None:
raise NotFoundError(
resource_type="invariant",
resource_id=invariant.id,
)
# Update mutable fields
model.text = invariant.text
model.scope = invariant.scope.value
model.source_name = invariant.source_name
model.active = invariant.active
model.non_overridable = invariant.non_overridable
model.created_at = invariant.created_at.isoformat()
session.flush()
logger.info(
"Invariant updated",
invariant_id=invariant.id,
)
except NotFoundError:
raise
except Exception:
session.rollback()
raise
finally:
session.close()
src/cleveragents/infrastructure/database/migrations/versions/m11_001_standalone_invariants.py
+62
View File
@@ -0,0 +1,62 @@
"""Create standalone invariants table for InvariantService persistence.
Adds a new ``invariants`` table for top-level invariant constraints managed
by :class:`~cleveragents.application.services.invariant_service.InvariantService`
via the CLI ``agents invariant add/list/remove`` commands. This table is
separate from action-level (``action_invariants``) and plan-level
(``plan_invariants``) child tables, allowing invariants to persist across
CLI invocations even when not attached to a specific action or plan.
Revision ID: m11_001_standalone_invariants
Revises: m9_003_plan_result_success_column
Create Date: 2026-05-12 00:00:00
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "m11_001_standalone_invariants"
down_revision: str | None = "m9_003_plan_result_success_column"
branch_labels: str | None = None
depends_on: str | None = None
def upgrade() -> None:
"""Create the standalone invariants table and indexes."""
op.create_table(
"invariants",
sa.Column("id", sa.String(26), primary_key=True, nullable=False),
sa.Column("text", sa.Text(), nullable=False),
sa.Column("scope", sa.String(20), nullable=False),
sa.Column("source_name", sa.String(255), nullable=False),
sa.Column(
"created_at",
sa.String(30),
nullable=False,
server_default=sa.func.datetime("now"),
),
sa.Column(
"active", sa.Boolean(), nullable=False, server_default=sa.text("1")
),
sa.Column(
"non_overridable",
sa.Boolean(),
nullable=False,
server_default=sa.text("0"),
),
sa.CheckConstraint(
"scope IN ('global', 'project', 'action', 'plan')",
name="ck_invariants_scope",
),
)
op.create_index("ix_invariants_scope", "invariants", ["scope"])
op.create_index("ix_invariants_source_name", "invariants", ["source_name"])
def downgrade() -> None:
"""Drop the standalone invariants table and indexes."""
op.drop_index("ix_invariants_source_name", table_name="invariants")
op.drop_index("ix_invariants_scope", table_name="invariants")
op.drop_table("invariants")