diff --git a/src/cleveragents/application/services/invariant_service.py b/src/cleveragents/application/services/invariant_service.py index c9cbc0100..b76944c22 100644 --- a/src/cleveragents/application/services/invariant_service.py +++ b/src/cleveragents/application/services/invariant_service.py @@ -110,42 +110,43 @@ class InvariantService: 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: - # Run migrations (or, under tests, the patched template-copy - # fast path) so the ``invariants`` table exists before we hand - # the engine to sessionmaker. Bypassing this leaves the DB - # empty for callers that instantiate InvariantService outside - # the DI container (notably the invariant CLI commands). - try: - from cleveragents.infrastructure.database.migration_runner import ( - MigrationRunner, + with self._lock: + if self._session_factory is None and self._database_url is not None: + # Run migrations (or, under tests, the patched template-copy + # fast path) so the ``invariants`` table exists before we hand + # the engine to sessionmaker. Bypassing this leaves the DB + # empty for callers that instantiate InvariantService outside + # the DI container (notably the invariant CLI commands). + try: + from cleveragents.infrastructure.database.migration_runner import ( + MigrationRunner, + ) + + MigrationRunner(database_url=self._database_url).init_or_upgrade() + except Exception as exc: # pragma: no cover - defensive + self._logger.warning( + "MigrationRunner.init_or_upgrade failed; " + "InvariantService will attempt to proceed", + error=str(exc), + ) + + 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 {}, ) + from sqlalchemy.orm import sessionmaker - MigrationRunner(database_url=self._database_url).init_or_upgrade() - except Exception as exc: # pragma: no cover - defensive - self._logger.warning( - "MigrationRunner.init_or_upgrade failed; " - "InvariantService will attempt to proceed", - error=str(exc), + self._session_factory = sessionmaker( + bind=engine, + expire_on_commit=False, + autoflush=False, + autocommit=False, ) - - 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 {}, - ) - from sqlalchemy.orm import sessionmaker - - self._session_factory = sessionmaker( - bind=engine, - expire_on_commit=False, - autoflush=False, - autocommit=False, - ) assert self._session_factory is not None # Guaranteed by guard above return self._session_factory @@ -215,19 +216,24 @@ class InvariantService: non_overridable=non_overridable, ) - if self._database_url is not None: - session = self._ensure_session_factory()() - try: - from cleveragents.infrastructure.database.models import InvariantModel - - model = InvariantModel.from_domain(invariant) - session.add(model) - session.flush() - session.commit() - finally: - session.close() - with self._lock: + if self._database_url is not None: + session = self._ensure_session_factory()() + try: + from cleveragents.infrastructure.database.models import ( + InvariantModel, + ) + + model = InvariantModel.from_domain(invariant) + session.add(model) + session.flush() + session.commit() + except Exception: + session.rollback() + raise + finally: + session.close() + self._invariants[invariant.id] = invariant self._logger.info( @@ -330,25 +336,30 @@ class InvariantService: # Invariant is frozen (immutable); create a new instance with active=False inactive = inv.model_copy(update={"active": False}) - # Persist to DB when configured - if self._database_url is not None: - session = self._ensure_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() - session.commit() - finally: - session.close() - with self._lock: + # Persist to DB when configured + if self._database_url is not None: + session = self._ensure_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() + session.commit() + except Exception: + session.rollback() + raise + finally: + session.close() + self._invariants[invariant_id] = inactive self._logger.info("Invariant removed (soft-delete)", invariant_id=invariant_id)