diff --git a/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/in_memory_fact_store.py b/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/in_memory_fact_store.py index 291107b4b..8df564743 100644 --- a/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/in_memory_fact_store.py +++ b/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/in_memory_fact_store.py @@ -9,6 +9,7 @@ from infinity_context_core.features.memory_facts.public import ( FEATURE_ID, FactEligibilityPolicy, + FactRelationSnapshot, FactSupersessionRelation, FactTemporalDecision, FactTemporalDecisionType, @@ -22,6 +23,10 @@ MemoryFactUnitOfWorkFactoryPort, ) +from infinity_context_adapters.features.memory_facts.in_memory_relation_store import ( + InMemoryFactRelationRepository, +) + _FactKey = tuple[str, str, str | None, str] _IdempotencyKey = tuple[str, str, str | None, FactTemporalDecisionType, str] _OperationReceiptKey = tuple[str, str, str | None, str, str] @@ -36,6 +41,7 @@ def __init__(self, facts: Iterable[MemoryFactSnapshot] = ()) -> None: self._decision_idempotency: dict[_IdempotencyKey, str] = {} self._operation_receipts: dict[_OperationReceiptKey, MemoryFactOperationReceipt] = {} self._supersessions: list[FactSupersessionRelation] = [] + self._relations: dict[str, FactRelationSnapshot] = {} self._revision = 0 for fact in facts: self._put(fact, allow_existing=False) @@ -51,6 +57,7 @@ def snapshot( dict[_IdempotencyKey, str], dict[_OperationReceiptKey, MemoryFactOperationReceipt], list[FactSupersessionRelation], + dict[str, FactRelationSnapshot], int, ]: return ( @@ -61,6 +68,7 @@ def snapshot( dict(self._decision_idempotency), dict(self._operation_receipts), list(self._supersessions), + dict(self._relations), self._revision, ) @@ -73,6 +81,7 @@ def replace( decision_idempotency: dict[_IdempotencyKey, str], operation_receipts: dict[_OperationReceiptKey, MemoryFactOperationReceipt], supersessions: list[FactSupersessionRelation], + relations: dict[str, FactRelationSnapshot], expected_revision: int, ) -> None: if self._revision != expected_revision: @@ -84,6 +93,7 @@ def replace( self._decision_idempotency = dict(decision_idempotency) self._operation_receipts = dict(operation_receipts) self._supersessions = list(supersessions) + self._relations = dict(relations) self._revision += 1 def facts(self) -> tuple[MemoryFactSnapshot, ...]: @@ -507,6 +517,7 @@ def __init__(self, state: _InMemoryMemoryFactState | None = None) -> None: self._working_decision_idempotency, self._working_operation_receipts, self._working_supersessions, + self._working_relations, self._base_revision, ) = self._state.snapshot() self.facts = InMemoryMemoryFactRepository.transactional( @@ -523,6 +534,9 @@ def __init__(self, state: _InMemoryMemoryFactState | None = None) -> None: self._working_operation_receipts ) self.supersessions = InMemoryFactSupersessionRepository(self._working_supersessions) + self.relations = InMemoryFactRelationRepository( + self._working_relations, self._working_facts, self._working_supersessions + ) self._committed = False async def __aenter__(self) -> InMemoryMemoryFactUnitOfWork: @@ -554,6 +568,7 @@ async def commit(self) -> None: self._working_decision_idempotency, self._working_operation_receipts, self._working_supersessions, + self._working_relations, self._base_revision, ) self._committed = True @@ -567,6 +582,7 @@ async def rollback(self) -> None: self._working_decision_idempotency, self._working_operation_receipts, self._working_supersessions, + self._working_relations, self._base_revision, ) = self._state.snapshot() self.facts = InMemoryMemoryFactRepository.transactional( @@ -582,6 +598,9 @@ async def rollback(self) -> None: self._working_operation_receipts ) self.supersessions = InMemoryFactSupersessionRepository(self._working_supersessions) + self.relations = InMemoryFactRelationRepository( + self._working_relations, self._working_facts, self._working_supersessions + ) self._committed = False diff --git a/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/in_memory_relation_store.py b/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/in_memory_relation_store.py new file mode 100644 index 000000000..db8a6bfcd --- /dev/null +++ b/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/in_memory_relation_store.py @@ -0,0 +1,155 @@ +"""Minimal transaction-local relation adapter for focused feature tests.""" + +from __future__ import annotations + +from collections.abc import Mapping + +from infinity_context_core.features.memory_facts.public import ( + FactRelationConflict, + FactRelationSnapshot, + FactRelationStatus, + FactRelationType, + FactSupersessionRelation, + MemoryFactScope, + MemoryFactSnapshot, +) + + +class InMemoryFactRelationRepository: + def __init__( + self, + relations: dict[str, FactRelationSnapshot], + facts: Mapping[tuple[str, str, str | None, str], MemoryFactSnapshot], + supersessions: list[FactSupersessionRelation], + ) -> None: + self._relations = relations + self._facts = facts + self._supersessions = supersessions + + def _all(self) -> tuple[FactRelationSnapshot, ...]: + return tuple(self._relations.values()) + tuple( + FactRelationSnapshot( + relation_id=item.relation_id, + space_id=item.scope.space_id, + memory_scope_id=item.scope.memory_scope_id, + thread_id=item.scope.thread_id, + source_fact_id=item.successor_fact_id, + target_fact_id=item.predecessor_fact_id, + relation_type=FactRelationType.SUPERSEDES, + reason=f"temporal_decision:{item.decision_id}", + status=FactRelationStatus.ACTIVE, + observed_at=item.created_at, + valid_from=item.effective_at, + valid_to=None, + created_at=item.created_at, + updated_at=item.created_at, + ) + for item in self._supersessions + ) + + async def get(self, relation_id: str, *, scope: MemoryFactScope) -> FactRelationSnapshot | None: + return next( + ( + item + for item in self._all() + if item.relation_id == relation_id and _in_scope(item, scope) + ), + None, + ) + + async def find_active( + self, *, source_fact_id: str, target_fact_id: str, relation_type: FactRelationType + ) -> FactRelationSnapshot | None: + return next( + ( + item + for item in self._all() + if item.source_fact_id == source_fact_id + and item.target_fact_id == target_fact_id + and item.relation_type == relation_type + and item.status == FactRelationStatus.ACTIVE + ), + None, + ) + + async def create(self, relation: FactRelationSnapshot) -> FactRelationSnapshot: + if any(item.relation_id == relation.relation_id for item in self._all()): + raise FactRelationConflict("Fact relation already exists") + if ( + await self.find_active( + source_fact_id=relation.source_fact_id, + target_fact_id=relation.target_fact_id, + relation_type=relation.relation_type, + ) + is not None + ): + raise FactRelationConflict("Active fact relation already exists") + self._relations[relation.relation_id] = relation + return relation + + async def save(self, relation: FactRelationSnapshot) -> FactRelationSnapshot: + current = await self.get( + relation.relation_id, scope=MemoryFactScope(relation.space_id, relation.memory_scope_id) + ) + if current is None: + raise LookupError("Fact relation not found") + saved = current.delete(now=relation.updated_at) + self._relations[relation.relation_id] = saved + return saved + + async def get_related_fact( + self, fact_id: str, *, scope: MemoryFactScope + ) -> MemoryFactSnapshot | None: + return next( + ( + fact + for fact in self._facts.values() + if fact.identity.fact_id == fact_id + and fact.identity.scope.space_id == scope.space_id + and fact.identity.scope.memory_scope_id == scope.memory_scope_id + ), + None, + ) + + async def list_for_fact( + self, + *, + fact_id: str, + scope: MemoryFactScope, + status: str | None, + limit: int, + enforce_code_scope: bool = False, + repository_id: str | None = None, + code_scope_id: str | None = None, + ) -> tuple[FactRelationSnapshot, ...]: + candidates = [] + for relation in self._all(): + if not _in_scope(relation, scope): + continue + if fact_id not in (relation.source_fact_id, relation.target_fact_id): + continue + if status is not None and relation.status != status: + continue + if enforce_code_scope: + other_id = ( + relation.target_fact_id + if relation.source_fact_id == fact_id + else relation.source_fact_id + ) + other = await self.get_related_fact(other_id, scope=scope) + if other is None or other.visibility.classification == "restricted": + continue + if other.code_scope is not None and not other.code_scope.is_visible_in( + repository_id=repository_id, code_scope_id=code_scope_id + ): + continue + candidates.append(relation) + return tuple( + sorted(candidates, key=lambda item: (item.updated_at, item.relation_id), reverse=True)[ + :limit + ] + ) + + +def _in_scope(relation: FactRelationSnapshot, scope: MemoryFactScope) -> bool: + return (relation.space_id, relation.memory_scope_id) == (scope.space_id, scope.memory_scope_id) diff --git a/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/postgres_fact_store.py b/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/postgres_fact_store.py index 16fafda73..7db75797f 100644 --- a/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/postgres_fact_store.py +++ b/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/postgres_fact_store.py @@ -37,6 +37,9 @@ memory_fact_source_ref_row_to_domain, memory_fact_source_ref_to_row, ) +from infinity_context_adapters.features.memory_facts.postgres_relation_store import ( + PostgresFactRelationRepository, +) from infinity_context_adapters.features.memory_facts.postgres_temporal_decision_store import ( PostgresFactSupersessionRepository, PostgresFactTemporalDecisionRepository, @@ -501,6 +504,7 @@ class PostgresMemoryFactTransaction: def __init__(self, session: AsyncSession, *, now: datetime) -> None: self._session = session + self.relations = PostgresFactRelationRepository(session) self.facts = PostgresMemoryFactStore(session) self.temporal_decisions = PostgresFactTemporalDecisionRepository(session) self.supersessions = PostgresFactSupersessionRepository(session) @@ -555,6 +559,7 @@ async def __aenter__(self) -> PostgresMemoryFactUnitOfWork: self._session = self._session_factory() now = self._clock.now() self._transaction = PostgresMemoryFactTransaction(self._session, now=now) + self.relations = self._transaction.relations self.facts = self._transaction.facts self.temporal_decisions = self._transaction.temporal_decisions self.supersessions = self._transaction.supersessions diff --git a/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/postgres_relation_store.py b/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/postgres_relation_store.py new file mode 100644 index 000000000..29ae83f4d --- /dev/null +++ b/packages/infinity_context_adapters/infinity_context_adapters/features/memory_facts/postgres_relation_store.py @@ -0,0 +1,211 @@ +"""Feature relations over existing rows and the canonical fact session.""" + +from __future__ import annotations + +from infinity_context_core.features.memory_facts.public import ( + FactRelationSnapshot, + FactRelationStatus, + FactRelationType, + MemoryFactScope, + MemoryFactSnapshot, +) +from sqlalchemy import and_, or_, select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import aliased + +from infinity_context_adapters.features.memory_facts.postgres_fact_mapping import ( + memory_fact_row_to_snapshot, +) +from infinity_context_adapters.postgres.fact_selection_conditions import ( + memory_fact_code_scope_conditions, +) +from infinity_context_adapters.postgres.models import ( + MemoryFactRelationRow, + MemoryFactRow, + MemorySourceRefRow, +) + + +class PostgresFactRelationRepository: + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def get(self, relation_id: str, *, scope: MemoryFactScope) -> FactRelationSnapshot | None: + row = ( + await self._session.execute( + select(MemoryFactRelationRow) + .where( + MemoryFactRelationRow.id == relation_id, + MemoryFactRelationRow.space_id == scope.space_id, + MemoryFactRelationRow.memory_scope_id == scope.memory_scope_id, + ) + .execution_options(populate_existing=True) + ) + ).scalar_one_or_none() + return relation_row_to_snapshot(row) if row is not None else None + + async def find_active( + self, *, source_fact_id: str, target_fact_id: str, relation_type: FactRelationType + ) -> FactRelationSnapshot | None: + row = ( + await self._session.execute( + select(MemoryFactRelationRow) + .where( + MemoryFactRelationRow.source_fact_id == source_fact_id, + MemoryFactRelationRow.target_fact_id == target_fact_id, + MemoryFactRelationRow.relation_type == relation_type.value, + MemoryFactRelationRow.status == "active", + ) + .execution_options(populate_existing=True) + ) + ).scalar_one_or_none() + return relation_row_to_snapshot(row) if row is not None else None + + async def create(self, relation: FactRelationSnapshot) -> FactRelationSnapshot: + # The handler holds the scope and endpoint locks through commit. The + # existing partial unique index also protects against non-feature writers. + self._session.add( + MemoryFactRelationRow( + id=relation.relation_id, + space_id=relation.space_id, + memory_scope_id=relation.memory_scope_id, + thread_id=relation.thread_id, + source_fact_id=relation.source_fact_id, + target_fact_id=relation.target_fact_id, + relation_type=relation.relation_type.value, + reason=relation.reason, + status=relation.status.value, + observed_at=relation.observed_at, + valid_from=relation.valid_from, + valid_to=relation.valid_to, + created_at=relation.created_at, + updated_at=relation.updated_at, + ) + ) + await self._session.flush() + return relation + + async def save(self, relation: FactRelationSnapshot) -> FactRelationSnapshot: + row = ( + await self._session.execute( + select(MemoryFactRelationRow) + .where( + MemoryFactRelationRow.id == relation.relation_id, + MemoryFactRelationRow.space_id == relation.space_id, + MemoryFactRelationRow.memory_scope_id == relation.memory_scope_id, + ) + .with_for_update() + .execution_options(populate_existing=True) + ) + ).scalar_one_or_none() + if row is None: + raise LookupError("Fact relation not found") + # Reapply lifecycle to the locked row, preserving immutable temporal audit + # columns and idempotent updated_at even if a legacy unlink raced us. + current = relation_row_to_snapshot(row) + saved = current.delete(now=relation.updated_at) + row.status = saved.status.value + row.updated_at = saved.updated_at + return saved + + async def get_related_fact( + self, fact_id: str, *, scope: MemoryFactScope + ) -> MemoryFactSnapshot | None: + row = ( + await self._session.execute( + select(MemoryFactRow).where( + MemoryFactRow.id == fact_id, + MemoryFactRow.space_id == scope.space_id, + MemoryFactRow.memory_scope_id == scope.memory_scope_id, + ) + ) + ).scalar_one_or_none() + if row is None: + return None + refs = list( + ( + await self._session.execute( + select(MemorySourceRefRow) + .where( + MemorySourceRefRow.fact_id == row.id, + MemorySourceRefRow.fact_version == row.version, + ) + .order_by(MemorySourceRefRow.id) + ) + ).scalars() + ) + return memory_fact_row_to_snapshot(row, refs) + + async def list_for_fact( + self, + *, + fact_id: str, + scope: MemoryFactScope, + status: str | None, + limit: int, + enforce_code_scope: bool = False, + repository_id: str | None = None, + code_scope_id: str | None = None, + ) -> tuple[FactRelationSnapshot, ...]: + conditions = [ + MemoryFactRelationRow.space_id == scope.space_id, + MemoryFactRelationRow.memory_scope_id == scope.memory_scope_id, + or_( + MemoryFactRelationRow.source_fact_id == fact_id, + MemoryFactRelationRow.target_fact_id == fact_id, + ), + ] + if status is not None: + conditions.append(MemoryFactRelationRow.status == status) + statement = select(MemoryFactRelationRow) + if enforce_code_scope: + other = aliased(MemoryFactRow) + statement = statement.join( + other, + or_( + and_( + MemoryFactRelationRow.source_fact_id == fact_id, + other.id == MemoryFactRelationRow.target_fact_id, + ), + and_( + MemoryFactRelationRow.target_fact_id == fact_id, + other.id == MemoryFactRelationRow.source_fact_id, + ), + ), + ) + conditions.extend( + memory_fact_code_scope_conditions( + other, + repository_id=repository_id, + code_scope_id=code_scope_id, + ) + ) + conditions.append(other.classification != "restricted") + rows = ( + await self._session.execute( + statement.where(*conditions) + .order_by(MemoryFactRelationRow.updated_at.desc(), MemoryFactRelationRow.id.desc()) + .limit(limit) + ) + ).scalars() + return tuple(relation_row_to_snapshot(row) for row in rows) + + +def relation_row_to_snapshot(row: MemoryFactRelationRow) -> FactRelationSnapshot: + """Translate persistence at the adapter boundary, including temporal-owned rows.""" + return FactRelationSnapshot( + relation_id=row.id, + space_id=row.space_id, + memory_scope_id=row.memory_scope_id, + thread_id=row.thread_id, + source_fact_id=row.source_fact_id, + target_fact_id=row.target_fact_id, + relation_type=FactRelationType(row.relation_type), + reason=row.reason, + status=FactRelationStatus(row.status), + observed_at=row.observed_at, + valid_from=row.valid_from, + valid_to=row.valid_to, + created_at=row.created_at, + updated_at=row.updated_at, + ) diff --git a/packages/infinity_context_core/infinity_context_core/features/memory_facts/application/__init__.py b/packages/infinity_context_core/infinity_context_core/features/memory_facts/application/__init__.py index 5dc061b70..b5922a593 100644 --- a/packages/infinity_context_core/infinity_context_core/features/memory_facts/application/__init__.py +++ b/packages/infinity_context_core/infinity_context_core/features/memory_facts/application/__init__.py @@ -27,6 +27,18 @@ ListMemoryFactVersionsHandler, MemoryFactReadUseCases, ) +from infinity_context_core.features.memory_facts.application.relations import ( + FactRelationItem, + FactRelationResult, + FactRelationsResult, + LinkFactsCommand, + LinkFactsHandler, + ListFactRelationsHandler, + ListFactRelationsQuery, + UnlinkFactRelationCommand, + UnlinkFactRelationHandler, + link_facts_in_transaction, +) from infinity_context_core.features.memory_facts.application.reviewed_mutations import ( ReviewedFactCandidate, ReviewedFactDecision, @@ -65,6 +77,16 @@ ) __all__ = ( + "LinkFactsCommand", + "LinkFactsHandler", + "ListFactRelationsQuery", + "ListFactRelationsHandler", + "UnlinkFactRelationCommand", + "UnlinkFactRelationHandler", + "FactRelationResult", + "FactRelationItem", + "FactRelationsResult", + "link_facts_in_transaction", "SUPERSESSION_POLICY_VERSION", "FACT_TEMPORAL_MUTATION_POLICY_VERSION", "ConfirmFactCommand", diff --git a/packages/infinity_context_core/infinity_context_core/features/memory_facts/application/relations.py b/packages/infinity_context_core/infinity_context_core/features/memory_facts/application/relations.py new file mode 100644 index 000000000..2294f028f --- /dev/null +++ b/packages/infinity_context_core/infinity_context_core/features/memory_facts/application/relations.py @@ -0,0 +1,195 @@ +"""Bounded generic link/list/unlink; reviewed temporal mutations have separate owners.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Literal + +from infinity_context_core.features.memory_facts.application.locking import ( + memory_fact_identity_lock_key, +) +from infinity_context_core.features.memory_facts.domain.fact import ( + MemoryFactIdentity, + MemoryFactScope, + MemoryFactSnapshot, +) +from infinity_context_core.features.memory_facts.domain.relations import ( + FactRelationSnapshot, + require_generic_relation_type, + require_linkable_facts, +) +from infinity_context_core.features.memory_facts.ports.clock import MemoryFactClockPort +from infinity_context_core.features.memory_facts.ports.ids import MemoryFactIdPort +from infinity_context_core.features.memory_facts.ports.unit_of_work import ( + MemoryFactTransactionPort, + MemoryFactUnitOfWorkFactoryPort, +) + + +@dataclass(frozen=True, slots=True) +class LinkFactsCommand: + source_identity: MemoryFactIdentity + target_identity: MemoryFactIdentity + relation_type: str + reason: str + observed_at: datetime | None = None + valid_from: datetime | None = None + valid_to: datetime | None = None + + +@dataclass(frozen=True, slots=True) +class ListFactRelationsQuery: + identity: MemoryFactIdentity + status: str | None = "active" + limit: int = 50 + enforce_code_scope: bool = False + repository_id: str | None = None + code_scope_id: str | None = None + + +@dataclass(frozen=True, slots=True) +class UnlinkFactRelationCommand: + relation_id: str + scope: MemoryFactScope + + +@dataclass(frozen=True, slots=True) +class FactRelationResult: + relation: FactRelationSnapshot + + +@dataclass(frozen=True, slots=True) +class FactRelationItem: + relation: FactRelationSnapshot + related_fact: MemoryFactSnapshot + direction: Literal["outgoing", "incoming"] + + +@dataclass(frozen=True, slots=True) +class FactRelationsResult: + target: MemoryFactSnapshot + items: tuple[FactRelationItem, ...] + + +@dataclass(frozen=True, slots=True) +class LinkFactsHandler: + uow_factory: MemoryFactUnitOfWorkFactoryPort + clock: MemoryFactClockPort + ids: MemoryFactIdPort + + async def execute(self, command: LinkFactsCommand) -> FactRelationResult: + require_generic_relation_type(command.relation_type) + async with self.uow_factory() as uow: + result = await link_facts_in_transaction( + uow, command, now=self.clock.now(), ids=self.ids + ) + await uow.commit() + return result + + +async def link_facts_in_transaction( + transaction: MemoryFactTransactionPort, + command: LinkFactsCommand, + *, + now: datetime, + ids: MemoryFactIdPort, +) -> FactRelationResult: + """Join a caller-owned transaction without committing or opening another UoW.""" + relation_type = require_generic_relation_type(command.relation_type) + identities = tuple( + sorted( + {command.source_identity, command.target_identity}, key=memory_fact_identity_lock_key + ) + ) + # Lock the memory-scope key, then endpoints in the stable order used by + # audited mutations. Validate exact-thread eligibility before replay or writes. + scopes = sorted( + {(identity.scope.space_id, identity.scope.memory_scope_id) for identity in identities} + ) + for space_id, memory_scope_id in scopes: + await transaction.lock_scope(MemoryFactScope(space_id, memory_scope_id)) + locked = await transaction.facts.get_many_for_update(identities) + by_identity = {fact.identity: fact for fact in locked} + try: + source = by_identity[command.source_identity] + target = by_identity[command.target_identity] + except KeyError as exc: + raise LookupError("Fact not found") from exc + require_linkable_facts(source, target) + existing = await transaction.relations.find_active( + source_fact_id=source.identity.fact_id, + target_fact_id=target.identity.fact_id, + relation_type=relation_type, + ) + if existing is not None: + existing.require_temporal_replay( + observed_at=command.observed_at, + valid_from=command.valid_from, + valid_to=command.valid_to, + ) + return FactRelationResult(existing) + relation = FactRelationSnapshot.create( + relation_id=ids.new_fact_relation_id(), + source=source, + target=target, + relation_type=relation_type, + reason=command.reason, + now=now, + observed_at=command.observed_at, + valid_from=command.valid_from, + valid_to=command.valid_to, + ) + return FactRelationResult(await transaction.relations.create(relation)) + + +@dataclass(frozen=True, slots=True) +class ListFactRelationsHandler: + uow_factory: MemoryFactUnitOfWorkFactoryPort + + async def execute(self, query: ListFactRelationsQuery) -> FactRelationsResult: + if query.limit < 1 or query.limit > 100: + raise ValueError("Fact relation limit must be between 1 and 100") + async with self.uow_factory() as uow: + target = await uow.facts.get(query.identity) + if target is None: + raise LookupError("Fact not found") + relations = await uow.relations.list_for_fact( + fact_id=query.identity.fact_id, + scope=query.identity.scope, + status=query.status, + limit=query.limit, + enforce_code_scope=query.enforce_code_scope, + repository_id=query.repository_id, + code_scope_id=query.code_scope_id, + ) + items: list[FactRelationItem] = [] + for relation in relations: + outgoing = relation.source_fact_id == query.identity.fact_id + other_id = relation.target_fact_id if outgoing else relation.source_fact_id + other = await uow.relations.get_related_fact(other_id, scope=query.identity.scope) + if other is None or other.visibility.status == "deleted": + continue + if other.visibility.classification == "restricted": + continue + items.append( + FactRelationItem(relation, other, "outgoing" if outgoing else "incoming") + ) + return FactRelationsResult(target, tuple(items)) + + +@dataclass(frozen=True, slots=True) +class UnlinkFactRelationHandler: + uow_factory: MemoryFactUnitOfWorkFactoryPort + clock: MemoryFactClockPort + + async def execute(self, command: UnlinkFactRelationCommand) -> FactRelationResult: + async with self.uow_factory() as uow: + scope = MemoryFactScope(command.scope.space_id, command.scope.memory_scope_id) + await uow.lock_scope(scope) + relation = await uow.relations.get(command.relation_id, scope=scope) + if relation is None: + raise LookupError("Fact relation not found") + saved = await uow.relations.save(relation.delete(now=self.clock.now())) + await uow.commit() + return FactRelationResult(saved) diff --git a/packages/infinity_context_core/infinity_context_core/features/memory_facts/domain/__init__.py b/packages/infinity_context_core/infinity_context_core/features/memory_facts/domain/__init__.py index b7ea28382..55038feb8 100644 --- a/packages/infinity_context_core/infinity_context_core/features/memory_facts/domain/__init__.py +++ b/packages/infinity_context_core/infinity_context_core/features/memory_facts/domain/__init__.py @@ -18,6 +18,12 @@ FEATURE_ID, MemoryFactsFeature, ) +from infinity_context_core.features.memory_facts.domain.relations import ( + FactRelationConflict, + FactRelationSnapshot, + FactRelationStatus, + FactRelationType, +) from infinity_context_core.features.memory_facts.domain.selection import ( FactEligibilityAssessment, FactEligibilityPolicy, @@ -58,6 +64,10 @@ ) __all__ = ( + "FactRelationConflict", + "FactRelationSnapshot", + "FactRelationStatus", + "FactRelationType", "FEATURE_ID", "FactCurrentness", "FactCurrentnessAssessment", diff --git a/packages/infinity_context_core/infinity_context_core/features/memory_facts/domain/relations.py b/packages/infinity_context_core/infinity_context_core/features/memory_facts/domain/relations.py new file mode 100644 index 000000000..e3b8ca22e --- /dev/null +++ b/packages/infinity_context_core/infinity_context_core/features/memory_facts/domain/relations.py @@ -0,0 +1,157 @@ +"""Generic relation policy over canonical facts; temporal decisions stay separate.""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from datetime import datetime +from enum import StrEnum + +from infinity_context_core.features.memory_facts.domain.fact import MemoryFactSnapshot + + +class FactRelationType(StrEnum): + SUPPORTS = "supports" + SUPERSEDES = "supersedes" + CONTRADICTS = "contradicts" + DUPLICATES = "duplicates" + REFERENCES = "references" + DEPENDS_ON = "depends_on" + RELATED_TO = "related_to" + + +class FactRelationStatus(StrEnum): + ACTIVE = "active" + DELETED = "deleted" + + +class FactRelationConflict(ValueError): + """A valid request conflicts with canonical relation or fact state.""" + + +@dataclass(frozen=True, slots=True) +class FactRelationSnapshot: + relation_id: str + space_id: str + memory_scope_id: str + source_fact_id: str + target_fact_id: str + relation_type: FactRelationType + reason: str + status: FactRelationStatus + observed_at: datetime + valid_from: datetime | None + valid_to: datetime | None + created_at: datetime + updated_at: datetime + thread_id: str | None = None + + @classmethod + def create( + cls, + *, + relation_id: str, + source: MemoryFactSnapshot, + target: MemoryFactSnapshot, + relation_type: FactRelationType, + reason: str, + now: datetime, + observed_at: datetime | None = None, + valid_from: datetime | None = None, + valid_to: datetime | None = None, + ) -> FactRelationSnapshot: + require_generic_relation_type(relation_type) + require_linkable_facts(source, target) + if source.identity.fact_id == target.identity.fact_id: + raise ValueError("Fact relation requires two distinct facts") + if not reason.strip(): + raise ValueError("Fact relation reason is required") + if valid_from is not None and valid_to is not None: + start, end = comparable_datetimes(valid_from, valid_to) + if end <= start: + raise ValueError("Temporal valid_to must be after valid_from") + return cls( + relation_id=relation_id, + space_id=source.identity.scope.space_id, + memory_scope_id=source.identity.scope.memory_scope_id, + thread_id=source.identity.scope.thread_id, + source_fact_id=source.identity.fact_id, + target_fact_id=target.identity.fact_id, + relation_type=FactRelationType(relation_type), + reason=reason.strip(), + status=FactRelationStatus.ACTIVE, + observed_at=observed_at or now, + valid_from=valid_from, + valid_to=valid_to, + created_at=now, + updated_at=now, + ) + + def delete(self, *, now: datetime) -> FactRelationSnapshot: + if self.relation_type == FactRelationType.SUPERSEDES: + raise ValueError("Supersession relations are immutable; use a compensating decision") + if self.status == FactRelationStatus.DELETED: + return self + return replace(self, status=FactRelationStatus.DELETED, updated_at=now) + + def require_temporal_replay( + self, + *, + observed_at: datetime | None, + valid_from: datetime | None, + valid_to: datetime | None, + ) -> None: + mismatches = [ + name + for name, existing, requested in ( + ("observed_at", self.observed_at, observed_at), + ("valid_from", self.valid_from, valid_from), + ("valid_to", self.valid_to, valid_to), + ) + if requested is not None + and (existing is None or not _datetime_equal(existing, requested)) + ] + if mismatches: + raise FactRelationConflict( + "Active fact relation already exists with different temporal fields: " + + ", ".join(mismatches) + ) + + +def require_generic_relation_type(value: FactRelationType | str) -> FactRelationType: + try: + relation_type = FactRelationType(value) + except ValueError as exc: + raise ValueError("Unknown fact relation type") from exc + if relation_type in {FactRelationType.SUPERSEDES, FactRelationType.CONTRADICTS}: + raise ValueError("Temporal relations require the audited supersede or dispute use case") + return relation_type + + +def require_linkable_facts(source: MemoryFactSnapshot, target: MemoryFactSnapshot) -> None: + source_scope, target_scope = source.identity.scope, target.identity.scope + # Canonical endpoint FKs require the same exact thread, including global. + if (source_scope.space_id, source_scope.memory_scope_id) != ( + target_scope.space_id, + target_scope.memory_scope_id, + ): + raise FactRelationConflict("Fact relations cannot cross memory_scope boundaries") + if source_scope.thread_id != target_scope.thread_id: + raise FactRelationConflict("Fact relations cannot cross thread boundaries") + if any(fact.visibility.status == "deleted" for fact in (source, target)): + raise FactRelationConflict("Deleted facts cannot be linked") + if any(fact.visibility.classification == "restricted" for fact in (source, target)): + raise FactRelationConflict("Restricted facts cannot be linked") + + +def comparable_datetimes(left: datetime, right: datetime) -> tuple[datetime, datetime]: + """Preserve legacy naive timestamp compatibility at the relation boundary.""" + if left.tzinfo is None and right.tzinfo is not None: + left = left.replace(tzinfo=right.tzinfo) + elif left.tzinfo is not None and right.tzinfo is None: + right = right.replace(tzinfo=left.tzinfo) + return left, right + + +def _datetime_equal(left: datetime, right: datetime) -> bool: + left, right = comparable_datetimes(left, right) + return left == right diff --git a/packages/infinity_context_core/infinity_context_core/features/memory_facts/ports/__init__.py b/packages/infinity_context_core/infinity_context_core/features/memory_facts/ports/__init__.py index 43e0e26e7..c6f416a4a 100644 --- a/packages/infinity_context_core/infinity_context_core/features/memory_facts/ports/__init__.py +++ b/packages/infinity_context_core/infinity_context_core/features/memory_facts/ports/__init__.py @@ -21,6 +21,9 @@ MemoryFactListSpec, MemoryFactReadModelPort, ) +from infinity_context_core.features.memory_facts.ports.relations import ( + FactRelationRepositoryPort, +) from infinity_context_core.features.memory_facts.ports.repositories import ( MemoryFactRepositoryPort, ) @@ -38,6 +41,7 @@ ) __all__ = ( + "FactRelationRepositoryPort", "FactSupersessionRepositoryPort", "FactTemporalDecisionRepositoryPort", "MemoryFactClockPort", diff --git a/packages/infinity_context_core/infinity_context_core/features/memory_facts/ports/relations.py b/packages/infinity_context_core/infinity_context_core/features/memory_facts/ports/relations.py new file mode 100644 index 000000000..8314ba3a9 --- /dev/null +++ b/packages/infinity_context_core/infinity_context_core/features/memory_facts/ports/relations.py @@ -0,0 +1,48 @@ +"""Canonical generic relations in the caller's fact transaction.""" + +from __future__ import annotations + +from typing import Protocol + +from infinity_context_core.features.memory_facts.domain.fact import ( + MemoryFactScope, + MemoryFactSnapshot, +) +from infinity_context_core.features.memory_facts.domain.relations import ( + FactRelationSnapshot, + FactRelationType, +) + + +class FactRelationRepositoryPort(Protocol): + async def get(self, relation_id: str, *, scope: MemoryFactScope) -> FactRelationSnapshot | None: + """Read within the supplied space and memory scope (independent of thread).""" + + async def find_active( + self, *, source_fact_id: str, target_fact_id: str, relation_type: FactRelationType + ) -> FactRelationSnapshot | None: + """Read the directed logical key after canonical fact locks are held.""" + + async def create(self, relation: FactRelationSnapshot) -> FactRelationSnapshot: + """Insert using the existing active logical-key uniqueness boundary.""" + + async def save(self, relation: FactRelationSnapshot) -> FactRelationSnapshot: + """Persist relation lifecycle only; never revise a fact or emit an outbox event.""" + + async def get_related_fact( + self, fact_id: str, *, scope: MemoryFactScope + ) -> MemoryFactSnapshot | None: + """Hydrate canonical related evidence within one memory scope.""" + + async def list_for_fact( + self, + *, + fact_id: str, + scope: MemoryFactScope, + status: str | None, + limit: int, + enforce_code_scope: bool = False, + repository_id: str | None = None, + code_scope_id: str | None = None, + ) -> tuple[FactRelationSnapshot, ...]: + """Order by updated_at/id descending; apply code scope before the row limit.""" diff --git a/packages/infinity_context_core/infinity_context_core/features/memory_facts/ports/unit_of_work.py b/packages/infinity_context_core/infinity_context_core/features/memory_facts/ports/unit_of_work.py index 3d84b5c4e..3aae515e0 100644 --- a/packages/infinity_context_core/infinity_context_core/features/memory_facts/ports/unit_of_work.py +++ b/packages/infinity_context_core/infinity_context_core/features/memory_facts/ports/unit_of_work.py @@ -15,6 +15,7 @@ from infinity_context_core.features.memory_facts.ports.outbox import ( MemoryFactOutboxPort, ) +from infinity_context_core.features.memory_facts.ports.relations import FactRelationRepositoryPort from infinity_context_core.features.memory_facts.ports.repositories import ( MemoryFactRepositoryPort, ) @@ -25,6 +26,7 @@ class MemoryFactTransactionPort(Protocol): + relations: FactRelationRepositoryPort facts: MemoryFactRepositoryPort supersessions: FactSupersessionRepositoryPort temporal_decisions: FactTemporalDecisionRepositoryPort diff --git a/packages/infinity_context_core/infinity_context_core/features/memory_facts/public.py b/packages/infinity_context_core/infinity_context_core/features/memory_facts/public.py index cff279f59..a6bcd5ab7 100644 --- a/packages/infinity_context_core/infinity_context_core/features/memory_facts/public.py +++ b/packages/infinity_context_core/infinity_context_core/features/memory_facts/public.py @@ -14,11 +14,18 @@ EndFactValidityCommand, EndFactValidityHandler, EndFactValidityResult, + FactRelationItem, + FactRelationResult, + FactRelationsResult, ForgetFactCommand, ForgetFactHandler, ForgetFactResult, ForgetFactUseCase, GetMemoryFactHandler, + LinkFactsCommand, + LinkFactsHandler, + ListFactRelationsHandler, + ListFactRelationsQuery, ListMemoryFactsHandler, ListMemoryFactVersionsHandler, MemoryFactLifecycleUseCases, @@ -41,10 +48,13 @@ SupersedeFactCommand, SupersedeFactHandler, SupersedeFactResult, + UnlinkFactRelationCommand, + UnlinkFactRelationHandler, UpdateFactCommand, UpdateFactHandler, UpdateFactResult, UpdateFactUseCase, + link_facts_in_transaction, memory_fact_identity_lock_key, ) from infinity_context_core.features.memory_facts.domain import ( @@ -61,6 +71,10 @@ FactLifecycle, FactLifecycleStatus, FactQuality, + FactRelationConflict, + FactRelationSnapshot, + FactRelationStatus, + FactRelationType, FactRetention, FactRevision, FactSupersessionPolicy, @@ -90,6 +104,7 @@ normalize_fact_taxonomy_fields, ) from infinity_context_core.features.memory_facts.ports import ( + FactRelationRepositoryPort, FactSupersessionRepositoryPort, FactTemporalDecisionRepositoryPort, MemoryFactClockPort, @@ -109,6 +124,21 @@ ) __all__ = ( + "FactRelationConflict", + "FactRelationSnapshot", + "FactRelationStatus", + "FactRelationType", + "LinkFactsCommand", + "LinkFactsHandler", + "ListFactRelationsQuery", + "ListFactRelationsHandler", + "UnlinkFactRelationCommand", + "UnlinkFactRelationHandler", + "FactRelationResult", + "FactRelationItem", + "FactRelationsResult", + "link_facts_in_transaction", + "FactRelationRepositoryPort", "FEATURE_ID", "FACT_TEMPORAL_MUTATION_POLICY_VERSION", "SUPERSESSION_POLICY_VERSION", diff --git a/packages/infinity_context_core/infinity_context_core/features/memory_facts/tests/test_application_ports_public_skeleton.py b/packages/infinity_context_core/infinity_context_core/features/memory_facts/tests/test_application_ports_public_skeleton.py index dc165d494..2c359bcb3 100644 --- a/packages/infinity_context_core/infinity_context_core/features/memory_facts/tests/test_application_ports_public_skeleton.py +++ b/packages/infinity_context_core/infinity_context_core/features/memory_facts/tests/test_application_ports_public_skeleton.py @@ -218,6 +218,21 @@ def test_memory_facts_public_api_exports_exact_feature_boundary() -> None: public = importlib.import_module(PUBLIC_MODULE) expected_exports = { + "FactRelationConflict": domain, + "FactRelationSnapshot": domain, + "FactRelationStatus": domain, + "FactRelationType": domain, + "LinkFactsCommand": application, + "LinkFactsHandler": application, + "ListFactRelationsQuery": application, + "ListFactRelationsHandler": application, + "UnlinkFactRelationCommand": application, + "UnlinkFactRelationHandler": application, + "FactRelationResult": application, + "FactRelationItem": application, + "FactRelationsResult": application, + "link_facts_in_transaction": application, + "FactRelationRepositoryPort": ports, "FEATURE_ID": domain, "FACT_TEMPORAL_MUTATION_POLICY_VERSION": application, "SUPERSESSION_POLICY_VERSION": application, diff --git a/tests/adapters/feature_relation_thread_cases.py b/tests/adapters/feature_relation_thread_cases.py new file mode 100644 index 000000000..e19e86310 --- /dev/null +++ b/tests/adapters/feature_relation_thread_cases.py @@ -0,0 +1,106 @@ +"""Shared actual SQL adapter checks, also run against coordinator PostgreSQL.""" + +from dataclasses import replace + +import pytest +from infinity_context_adapters.postgres.models import Base, MemoryFactRelationRow, MemoryThreadRow +from infinity_context_core.features.memory_facts.public import ( + FactRelationConflict, + LinkFactsCommand, + LinkFactsHandler, + MemoryFactIdentity, + MemoryFactScope, + MemoryFactSnapshot, + MemoryFactSourceRef, +) +from sqlalchemy import event, select, text + + +async def exercise_thread_cases(engine, sessions, factory, clock, ids): + scope = MemoryFactScope("thread-case-space", "thread-case-scope") + facts = tuple( + MemoryFactSnapshot( + identity=MemoryFactIdentity(name, replace(scope, thread_id=thread)), + text=name, + source_refs=(MemoryFactSourceRef("manual", name),), + created_at=clock.now(), + updated_at=clock.now(), + ) + for name, thread in ( + ("global-a", None), + ("global-b", None), + ("thread-a", "one"), + ("thread-b", "one"), + ("thread-c", "two"), + ) + ) + async with sessions() as session: + for thread_id in ("one", "two"): + session.add( + MemoryThreadRow( + id=thread_id, + space_id=scope.space_id, + memory_scope_id=scope.memory_scope_id, + external_ref=thread_id, + status="active", + created_at=clock.now(), + updated_at=clock.now(), + ) + ) + await session.commit() + async with factory() as uow: + for fact in facts: + await uow.facts.create(fact) + await uow.commit() + handler = LinkFactsHandler(factory, clock, ids) + for source, target in ((facts[0], facts[1]), (facts[2], facts[3])): + command = LinkFactsCommand(source.identity, target.identity, "supports", "evidence") + result = await handler.execute(command) + assert result.relation.thread_id == source.identity.scope.thread_id + replay = await handler.execute(command) + assert replay.relation.relation_id == result.relation.relation_id + assert replay.relation.thread_id == result.relation.thread_id + async with sessions() as session: + row = await session.get(MemoryFactRelationRow, result.relation.relation_id) + assert row.thread_id == source.identity.scope.thread_id + assert row.thread_scope_key == ( + "global" if row.thread_id is None else "thread:" + row.thread_id + ) + + async def canonical_state(): + async with sessions() as session: + return { + table.name: tuple( + sorted(repr(tuple(row)) for row in (await session.execute(select(table))).all()) + ) + for table in Base.metadata.sorted_tables + } + + before = await canonical_state() + statements = [] + + def capture(conn, cursor, statement, parameters, context, executemany): + statements.append(statement.strip().split()[0].upper()) + + event.listen(engine.sync_engine, "before_cursor_execute", capture) + try: + for source, target in ( + (facts[0], facts[2]), + (facts[2], facts[0]), + (facts[2], facts[4]), + (facts[4], facts[2]), + ): + with pytest.raises(FactRelationConflict, match="cross thread boundaries"): + await handler.execute( + LinkFactsCommand(source.identity, target.identity, "supports", "rejected") + ) + finally: + event.remove(engine.sync_engine, "before_cursor_execute", capture) + assert not {"INSERT", "UPDATE", "DELETE"}.intersection(statements) + assert await canonical_state() == before + + if engine.dialect.name == "sqlite": + async with sessions() as session: + assert ( + await session.execute(text("PRAGMA foreign_key_check(memory_fact_relations)")) + ).all() == [] diff --git a/tests/adapters/test_feature_generic_fact_relations.py b/tests/adapters/test_feature_generic_fact_relations.py new file mode 100644 index 000000000..f685efb06 --- /dev/null +++ b/tests/adapters/test_feature_generic_fact_relations.py @@ -0,0 +1,449 @@ +"""Generic relation parity and transaction behavior through the feature public API.""" + +from __future__ import annotations + +import asyncio +from dataclasses import replace +from datetime import timedelta + +import pytest +from infinity_context_adapters.features.memory_facts.in_memory_fact_store import ( + InMemoryMemoryFactUnitOfWorkFactory, +) +from infinity_context_core.domain.entities import ( + FactRelationType as LegacyType, +) +from infinity_context_core.domain.entities import ( + MemoryFactRelation as LegacyRelation, +) +from infinity_context_core.features.memory_facts.public import ( + FactCodeScopeReference, + FactRelationConflict, + FactRelationSnapshot, + FactRelationStatus, + FactRelationType, + FactSupersessionRelation, + LinkFactsCommand, + LinkFactsHandler, + ListFactRelationsHandler, + ListFactRelationsQuery, + MemoryFactScope, + UnlinkFactRelationCommand, + UnlinkFactRelationHandler, + link_facts_in_transaction, +) +from memory_fact_test_support import EARLIER, LATER, NOW, FakeClock, FakeIds, _fact_snapshot + +SOURCE = _fact_snapshot(fact_id="source") +TARGET = _fact_snapshot(fact_id="target") +COMMAND = LinkFactsCommand(SOURCE.identity, TARGET.identity, "supports", " evidence ") + + +def _handler(factory, *ids): + return LinkFactsHandler(factory, FakeClock(NOW), FakeIds(fact_relation_ids=ids)) + + +def _relation(**changes): + value = FactRelationSnapshot.create( + relation_id="relation", + source=SOURCE, + target=TARGET, + relation_type=FactRelationType.SUPPORTS, + reason=" evidence ", + now=NOW, + ) + return replace(value, **changes) + + +@pytest.mark.parametrize("kind", list(FactRelationType)) +def test_allowed_enum_and_creation_match_legacy_domain(kind): + assert {item.value for item in FactRelationType} == {item.value for item in LegacyType} + if kind in {FactRelationType.SUPERSEDES, FactRelationType.CONTRADICTS}: + with pytest.raises(ValueError, match="audited"): + FactRelationSnapshot.create( + relation_id="relation", + source=SOURCE, + target=TARGET, + relation_type=kind, + reason=" evidence ", + now=NOW, + ) + return + feature = FactRelationSnapshot.create( + relation_id="relation", + source=SOURCE, + target=TARGET, + relation_type=kind, + reason=" evidence ", + now=NOW, + observed_at=EARLIER, + valid_from=EARLIER, + valid_to=LATER, + ) + legacy = LegacyRelation.create( + relation_id="relation", + space_id="space-1", + memory_scope_id="scope-1", + source_fact_id="source", + target_fact_id="target", + relation_type=LegacyType(kind), + reason=" evidence ", + now=NOW, + observed_at=EARLIER, + valid_from=EARLIER, + valid_to=LATER, + ) + for name in ( + "space_id", + "memory_scope_id", + "source_fact_id", + "target_fact_id", + "relation_type", + "reason", + "status", + "observed_at", + "valid_from", + "valid_to", + "created_at", + "updated_at", + ): + assert getattr(feature, name) == getattr(legacy, name) + assert feature.delete(now=LATER).updated_at == legacy.delete(now=LATER).updated_at + deleted = feature.delete(now=LATER) + assert deleted.delete(now=LATER + timedelta(days=1)) is deleted + + +@pytest.mark.parametrize( + "changes,match", + [ + ({"relation_type": "wrong"}, "Unknown"), + ({"relation_type": "supersedes"}, "audited"), + ({"relation_type": "contradicts"}, "audited"), + ({"target_identity": SOURCE.identity}, "distinct"), + ({"reason": " "}, "reason"), + ({"valid_from": NOW, "valid_to": NOW}, "after"), + ({"valid_from": LATER, "valid_to": NOW}, "after"), + ], +) +def test_invalid_link_rolls_back(changes, match): + async def run(): + factory = InMemoryMemoryFactUnitOfWorkFactory((SOURCE, TARGET)) + with pytest.raises(ValueError, match=match): + await _handler(factory, "relation").execute(replace(COMMAND, **changes)) + result = await ListFactRelationsHandler(factory).execute( + ListFactRelationsQuery(SOURCE.identity) + ) + assert result.items == () + assert factory.facts == (SOURCE, TARGET) + assert factory.outbox_messages == () + + asyncio.run(run()) + + +@pytest.mark.parametrize( + "change,match", + [ + ({"status": "deleted"}, "Deleted"), + ({"classification": "restricted"}, "Restricted"), + ], +) +@pytest.mark.parametrize("endpoint", ["source", "target"]) +def test_ineligible_endpoint_rejected_even_on_replay(change, match, endpoint): + async def run(): + factory = InMemoryMemoryFactUnitOfWorkFactory((SOURCE, TARGET)) + handler = _handler(factory, "relation") + await handler.execute(COMMAND) + fact = SOURCE if endpoint == "source" else TARGET + async with factory() as uow: + await uow.facts.save( + replace(fact, visibility=replace(fact.visibility, version=2, **change)) + ) + await uow.commit() + with pytest.raises(FactRelationConflict, match=match): + await handler.execute(COMMAND) + + asyncio.run(run()) + + +@pytest.mark.parametrize( + "scope", [MemoryFactScope("other", "scope-1"), MemoryFactScope("space-1", "other")] +) +def test_cross_scope_rejected(scope): + async def run(): + other = replace(TARGET, identity=replace(TARGET.identity, scope=scope)) + factory = InMemoryMemoryFactUnitOfWorkFactory((SOURCE, other)) + with pytest.raises(FactRelationConflict, match="boundaries"): + await _handler(factory, "relation").execute( + replace(COMMAND, target_identity=other.identity) + ) + + asyncio.run(run()) + + +def test_replay_temporal_conflict_unlink_and_relink_leave_facts_untouched(): + async def run(): + factory = InMemoryMemoryFactUnitOfWorkFactory((SOURCE, TARGET)) + handler = _handler(factory, "first", "second") + command = replace(COMMAND, observed_at=EARLIER, valid_from=NOW, valid_to=LATER) + first = await handler.execute(command) + assert first.relation.reason == "evidence" + assert await handler.execute(replace(COMMAND, reason="")) == first + assert ( + await handler.execute(replace(command, observed_at=EARLIER.replace(tzinfo=None))) + == first + ) + for field in ("observed_at", "valid_from", "valid_to"): + with pytest.raises(FactRelationConflict, match=field): + await handler.execute(replace(command, **{field: LATER + timedelta(days=1)})) + unlink = UnlinkFactRelationHandler(factory, FakeClock(LATER)) + deleted = await unlink.execute(UnlinkFactRelationCommand("first", SOURCE.identity.scope)) + assert deleted.relation.status == FactRelationStatus.DELETED + assert ( + await unlink.execute(UnlinkFactRelationCommand("first", SOURCE.identity.scope)) + == deleted + ) + second = await handler.execute(COMMAND) + assert second.relation.relation_id == "second" + for fact in (SOURCE, TARGET): + async with factory() as uow: + assert await uow.facts.list_versions(fact.identity) == (fact,) + assert factory.facts == (SOURCE, TARGET) + assert factory.outbox_messages == () + assert factory.temporal_decisions == () + + asyncio.run(run()) + + +def test_shared_transaction_does_not_commit_and_conflicting_snapshots_cannot_duplicate(): + async def run(): + factory = InMemoryMemoryFactUnitOfWorkFactory((SOURCE, TARGET)) + async with factory() as uow: + await link_facts_in_transaction( + uow, COMMAND, now=NOW, ids=FakeIds(fact_relation_ids=("rollback",)) + ) + query = ListFactRelationsQuery(SOURCE.identity) + assert (await ListFactRelationsHandler(factory).execute(query)).items == () + async with factory() as first, factory() as second: + for transaction, relation_id in ((first, "first"), (second, "second")): + await link_facts_in_transaction( + transaction, COMMAND, now=NOW, ids=FakeIds(fact_relation_ids=(relation_id,)) + ) + await first.commit() + with pytest.raises(ValueError, match="transaction conflict"): + await second.commit() + assert len((await ListFactRelationsHandler(factory).execute(query)).items) == 1 + + asyncio.run(run()) + + +@pytest.mark.parametrize("limit", [0, 101]) +def test_limit_validation(limit): + with pytest.raises(ValueError, match="between 1 and 100"): + asyncio.run( + ListFactRelationsHandler(InMemoryMemoryFactUnitOfWorkFactory()).execute( + ListFactRelationsQuery(SOURCE.identity, limit=limit) + ) + ) + + +def test_missing_and_wrong_scope_do_not_expose_relations(): + async def run(): + factory = InMemoryMemoryFactUnitOfWorkFactory((SOURCE, TARGET)) + await _handler(factory, "relation").execute(COMMAND) + with pytest.raises(LookupError): + await UnlinkFactRelationHandler(factory, FakeClock(NOW)).execute( + UnlinkFactRelationCommand("relation", MemoryFactScope("other", "scope-1")) + ) + missing = replace(SOURCE.identity, fact_id="missing") + with pytest.raises(LookupError): + await _handler(factory).execute(replace(COMMAND, source_identity=missing)) + with pytest.raises(LookupError): + await ListFactRelationsHandler(factory).execute(ListFactRelationsQuery(missing)) + + asyncio.run(run()) + + +def test_order_direction_status_code_scope_and_post_limit_eligibility(): + async def run(): + # Restricted/deleted filtering intentionally follows the legacy row-limit + # boundary except restricted facts excluded by the code-scope SQL join. + restricted = replace( + _fact_snapshot(fact_id="restricted"), + visibility=replace(SOURCE.visibility, classification="restricted"), + ) + deleted = replace( + _fact_snapshot(fact_id="deleted"), + visibility=replace(SOURCE.visibility, status="deleted"), + ) + project = replace( + _fact_snapshot(fact_id="project"), code_scope=FactCodeScopeReference("repo", "branch") + ) + unknown = replace( + _fact_snapshot(fact_id="unknown"), + visibility=replace(SOURCE.visibility, classification="unknown", status="disputed"), + ) + factory = InMemoryMemoryFactUnitOfWorkFactory( + (SOURCE, TARGET, restricted, deleted, project, unknown) + ) + async with factory() as uow: + for relation in ( + _relation(relation_id="a"), + _relation(relation_id="b", source_fact_id="target", target_fact_id="source"), + _relation(relation_id="c", target_fact_id="project"), + _relation(relation_id="d", target_fact_id="deleted"), + _relation(relation_id="e", target_fact_id="restricted"), + _relation(relation_id="f", target_fact_id="unknown"), + _relation( + relation_id="g", target_fact_id="target", status=FactRelationStatus.DELETED + ), + ): + # Seeding existing rows includes facts that became ineligible later. + uow.relations._relations[relation.relation_id] = relation + await uow.commit() + handler = ListFactRelationsHandler(factory) + result = await handler.execute(ListFactRelationsQuery(SOURCE.identity)) + assert [item.relation.relation_id for item in result.items] == ["f", "c", "b", "a"] + assert [item.direction for item in result.items] == [ + "outgoing", + "outgoing", + "incoming", + "outgoing", + ] + assert result.items[-1].related_fact.source_refs == TARGET.source_refs + assert (await handler.execute(ListFactRelationsQuery(SOURCE.identity, limit=2))).items[ + 0 + ].relation.relation_id == "f" + assert ( + len((await handler.execute(ListFactRelationsQuery(SOURCE.identity, limit=2))).items) + == 1 + ) + for repo, branch, expected in [ + (None, None, ["f", "b", "a"]), + ("repo", None, ["f", "b", "a"]), + ("repo", "branch", ["f", "c", "b", "a"]), + ("other", "branch", ["f", "b", "a"]), + ]: + selected = await handler.execute( + ListFactRelationsQuery( + SOURCE.identity, + enforce_code_scope=True, + repository_id=repo, + code_scope_id=branch, + ) + ) + assert [item.relation.relation_id for item in selected.items] == expected + deleted_result = await handler.execute( + ListFactRelationsQuery(SOURCE.identity, status="deleted") + ) + assert [item.relation.relation_id for item in deleted_result.items] == ["g"] + assert ( + len((await handler.execute(ListFactRelationsQuery(SOURCE.identity, status=None))).items) + == 5 + ) + assert ( + await handler.execute(ListFactRelationsQuery(SOURCE.identity, status="unknown")) + ).items == () + + asyncio.run(run()) + + +def test_temporal_owned_rows_visible_but_supersession_unlink_is_immutable(): + async def run(): + factory = InMemoryMemoryFactUnitOfWorkFactory((SOURCE, TARGET)) + async with factory() as uow: + await uow.supersessions.create( + FactSupersessionRelation( + "supersession", + SOURCE.identity.scope, + "source", + 2, + "target", + 2, + NOW, + "decision", + NOW, + ) + ) + await uow.relations.create( + _relation(relation_id="contradiction", relation_type=FactRelationType.CONTRADICTS) + ) + await uow.commit() + items = ( + await ListFactRelationsHandler(factory).execute(ListFactRelationsQuery(SOURCE.identity)) + ).items + assert {item.relation.relation_type for item in items} == { + FactRelationType.SUPERSEDES, + FactRelationType.CONTRADICTS, + } + unlink = UnlinkFactRelationHandler(factory, FakeClock(LATER)) + with pytest.raises(ValueError, match="immutable"): + await unlink.execute(UnlinkFactRelationCommand("supersession", SOURCE.identity.scope)) + await unlink.execute(UnlinkFactRelationCommand("contradiction", SOURCE.identity.scope)) + assert factory.facts == (SOURCE, TARGET) + + asyncio.run(run()) + + +def test_link_locks_scope_then_unique_facts_in_stable_identity_order(): + async def run(): + factory = InMemoryMemoryFactUnitOfWorkFactory((SOURCE, TARGET)) + async with factory() as uow: + calls = [] + get_many = uow.facts.get_many_for_update + + async def lock_scope(scope): + calls.append(("scope", scope)) + + async def get_many_for_update(identities): + calls.append(("facts", identities)) + return await get_many(identities) + + uow.lock_scope = lock_scope + uow.facts.get_many_for_update = get_many_for_update + await link_facts_in_transaction( + uow, + replace(COMMAND, source_identity=TARGET.identity, target_identity=SOURCE.identity), + now=NOW, + ids=FakeIds(fact_relation_ids=("reverse",)), + ) + assert calls == [ + ("scope", SOURCE.identity.scope), + ("facts", (SOURCE.identity, TARGET.identity)), + ] + + asyncio.run(run()) + + +@pytest.mark.parametrize( + "source_thread,target_thread", [(None, "one"), ("one", None), ("one", "two")] +) +def test_thread_mismatch_rejected_before_replay(source_thread, target_thread): + async def run(): + source = replace( + SOURCE, + identity=replace( + SOURCE.identity, scope=replace(SOURCE.identity.scope, thread_id=source_thread) + ), + ) + target = replace( + TARGET, + identity=replace( + TARGET.identity, scope=replace(TARGET.identity.scope, thread_id=target_thread) + ), + ) + factory = InMemoryMemoryFactUnitOfWorkFactory((source, target)) + # Simulate an old invalid row: endpoint validation must still precede replay. + async with factory() as uow: + uow.relations._relations["relation"] = _relation() + await uow.commit() + with pytest.raises(FactRelationConflict, match="cross thread boundaries"): + await _handler(factory).execute( + replace(COMMAND, source_identity=source.identity, target_identity=target.identity) + ) + assert factory.facts == (source, target) + assert factory.outbox_messages == () + assert factory.temporal_decisions == () + async with factory() as uow: + assert await uow.relations.get("relation", scope=source.identity.scope) == _relation() + + asyncio.run(run()) diff --git a/tests/adapters/test_feature_relation_postgres_adapter.py b/tests/adapters/test_feature_relation_postgres_adapter.py new file mode 100644 index 000000000..545b8c19d --- /dev/null +++ b/tests/adapters/test_feature_relation_postgres_adapter.py @@ -0,0 +1,173 @@ +"""SQL adapter parity on SQLite; live PostgreSQL acceptance is a separate test.""" + +from __future__ import annotations + +import asyncio +from dataclasses import replace +from datetime import timedelta + +from infinity_context_adapters.features.memory_facts.postgres_fact_store import ( + PostgresMemoryFactTransaction, + PostgresMemoryFactUnitOfWorkFactory, +) +from infinity_context_adapters.postgres import ( + build_async_engine, + build_session_factory, + create_schema, +) +from infinity_context_adapters.postgres.fact_repositories import ( + PostgresFactRelationRepository as LegacyRelationRepository, +) +from infinity_context_adapters.postgres.models import ( + MemoryFactRelationRow, + MemoryFactRow, + MemoryFactVersionRow, + MemoryOutboxRow, + MemorySourceRefRow, +) +from infinity_context_core.features.memory_facts.public import ( + FactCodeScopeReference, + FactRelationType, + LinkFactsCommand, + LinkFactsHandler, + ListFactRelationsHandler, + ListFactRelationsQuery, + MemoryFactSourceRef, + UnlinkFactRelationCommand, + UnlinkFactRelationHandler, + link_facts_in_transaction, +) +from memory_fact_test_support import NOW, FakeClock, FakeIds, _fact_snapshot +from sqlalchemy import func, select + +from tests.adapters.feature_relation_thread_cases import exercise_thread_cases + + +def test_existing_rows_session_rollback_hydration_and_legacy_order(tmp_path): + async def run(): + engine = build_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'relations.db'}") + try: + await create_schema(engine) + sessions = build_session_factory(engine) + factory = PostgresMemoryFactUnitOfWorkFactory( + session_factory=sessions, clock=FakeClock(NOW) + ) + facts = tuple( + replace( + _fact_snapshot(fact_id=name), + source_refs=(MemoryFactSourceRef("manual", name, char_start=1, char_end=5),), + code_scope=FactCodeScopeReference("repo", "branch") + if name == "project" + else None, + ) + for name in ("source", "target", "project") + ) + async with factory() as uow: + for fact in facts: + await uow.facts.create(fact) + await uow.commit() + command = LinkFactsCommand(facts[0].identity, facts[1].identity, "supports", "evidence") + handler = LinkFactsHandler( + factory, FakeClock(NOW), FakeIds(fact_relation_ids=("a", "b", "c")) + ) + first = await handler.execute(command) + await handler.execute(replace(command, target_identity=facts[2].identity)) + await handler.execute( + replace( + command, source_identity=facts[1].identity, target_identity=facts[0].identity + ) + ) + replay = await handler.execute(replace(command, reason="different")) + assert replay.relation.relation_id == first.relation.relation_id + reader = ListFactRelationsHandler(factory) + for repo, branch in [ + (None, None), + ("repo", None), + ("repo", "branch"), + ("other", "branch"), + ]: + query = ListFactRelationsQuery( + facts[0].identity, + enforce_code_scope=True, + repository_id=repo, + code_scope_id=branch, + ) + result = await reader.execute(query) + async with sessions() as session: + legacy = await LegacyRelationRepository(session).list_for_fact( + fact_id="source", + status="active", + limit=50, + enforce_code_scope=True, + repository_id=repo, + code_scope_id=branch, + ) + assert [item.relation.relation_id for item in result.items] == [ + str(item.id) for item in legacy + ] + assert all(item.related_fact.source_refs[0].char_end == 5 for item in result.items) + # Excluded project relation ranks first; filtering must precede LIMIT 1. + async with sessions() as session: + project_row = await session.get(MemoryFactRelationRow, "b") + project_row.updated_at = NOW + timedelta(days=1) + await session.commit() + selected = await reader.execute( + ListFactRelationsQuery(facts[0].identity, limit=1, enforce_code_scope=True) + ) + assert [item.relation.relation_id for item in selected.items] == ["c"] + # Joining the supplied session must roll back with unrelated canonical work. + async with sessions() as session: + transaction = PostgresMemoryFactTransaction(session, now=NOW) + assert transaction.relations._session is session + await link_facts_in_transaction( + transaction, + replace(command, relation_type="references"), + now=NOW, + ids=FakeIds(fact_relation_ids=("rollback",)), + ) + unrelated = await session.get(MemoryFactRow, "source") + unrelated.text = "uncommitted canonical change" + await session.flush() + assert unrelated.text != facts[0].text + await session.rollback() + async with sessions() as session: + assert await session.get(MemoryFactRelationRow, "rollback") is None + assert (await session.get(MemoryFactRow, "source")).text == facts[0].text + assert await session.scalar(select(func.count()).select_from(MemoryFactRow)) == 3 + assert ( + await session.scalar(select(func.count()).select_from(MemoryFactVersionRow)) + == 3 + ) + assert ( + await session.scalar(select(func.count()).select_from(MemorySourceRefRow)) == 3 + ) + assert await session.scalar(select(func.count()).select_from(MemoryOutboxRow)) == 0 + # Existing temporal rows are readable; unlink cannot overwrite their audit. + row = await session.get(MemoryFactRelationRow, "a") + row.relation_type = "supersedes" + await session.commit() + async with factory() as uow: + relation = await uow.relations.get("a", scope=facts[0].identity.scope) + assert relation.relation_type == FactRelationType.SUPERSEDES + import pytest + + with pytest.raises(ValueError, match="immutable"): + await UnlinkFactRelationHandler(factory, FakeClock(NOW)).execute( + UnlinkFactRelationCommand("a", facts[0].identity.scope) + ) + await UnlinkFactRelationHandler(factory, FakeClock(NOW)).execute( + UnlinkFactRelationCommand("b", facts[0].identity.scope) + ) + async with sessions() as session: + assert (await session.get(MemoryFactRelationRow, "b")).status == "deleted" + await exercise_thread_cases( + engine, + sessions, + factory, + FakeClock(NOW), + FakeIds(fact_relation_ids=("global-link", "thread-link")), + ) + finally: + await engine.dispose() + + asyncio.run(run()) diff --git a/tests/architecture/test_feature_owned_vertical_slices.py b/tests/architecture/test_feature_owned_vertical_slices.py index 97182e14c..52a827781 100644 --- a/tests/architecture/test_feature_owned_vertical_slices.py +++ b/tests/architecture/test_feature_owned_vertical_slices.py @@ -511,6 +511,21 @@ def test_memory_facts_public_api_is_importable_and_narrow() -> None: assert public.FEATURE_ID == "memory_facts" assert public.MemoryFactsFeature().feature_id == "memory_facts" assert public.__all__ == ( + "FactRelationConflict", + "FactRelationSnapshot", + "FactRelationStatus", + "FactRelationType", + "LinkFactsCommand", + "LinkFactsHandler", + "ListFactRelationsQuery", + "ListFactRelationsHandler", + "UnlinkFactRelationCommand", + "UnlinkFactRelationHandler", + "FactRelationResult", + "FactRelationItem", + "FactRelationsResult", + "link_facts_in_transaction", + "FactRelationRepositoryPort", "FEATURE_ID", "FACT_TEMPORAL_MUTATION_POLICY_VERSION", "SUPERSESSION_POLICY_VERSION", diff --git a/tests/architecture/test_generic_relation_public_boundary.py b/tests/architecture/test_generic_relation_public_boundary.py new file mode 100644 index 000000000..e4f9177da --- /dev/null +++ b/tests/architecture/test_generic_relation_public_boundary.py @@ -0,0 +1,46 @@ +"""Additive feature relation exports stay typed and provider independent.""" + +from __future__ import annotations + +import importlib +import inspect +from typing import get_type_hints + + +def test_relation_public_symbols_resolve_to_their_feature_layers(): + root = "infinity_context_core.features.memory_facts" + public = importlib.import_module(f"{root}.public") + exports = { + "domain": ( + "FactRelationConflict", + "FactRelationSnapshot", + "FactRelationStatus", + "FactRelationType", + ), + "ports": ("FactRelationRepositoryPort",), + "application": ( + "LinkFactsCommand", + "LinkFactsHandler", + "ListFactRelationsQuery", + "ListFactRelationsHandler", + "UnlinkFactRelationCommand", + "UnlinkFactRelationHandler", + "FactRelationResult", + "FactRelationItem", + "FactRelationsResult", + "link_facts_in_transaction", + ), + } + for layer, names in exports.items(): + module = importlib.import_module(f"{root}.{layer}") + for name in names: + value = getattr(public, name) + assert value is getattr(module, name) + assert name in public.__all__ + assert value.__module__.startswith(f"{root}.{layer}.") + if inspect.isclass(value): + assert all("Any" not in str(hint) for hint in get_type_hints(value).values()) + assert ( + get_type_hints(public.MemoryFactTransactionPort)["relations"] + is public.FactRelationRepositoryPort + ) diff --git a/tests/e2e/test_feature_generic_relations_postgres.py b/tests/e2e/test_feature_generic_relations_postgres.py new file mode 100644 index 000000000..79ef91a66 --- /dev/null +++ b/tests/e2e/test_feature_generic_relations_postgres.py @@ -0,0 +1,198 @@ +"""Disposable PostgreSQL acceptance for the generic feature relation transaction. + +Pending execution until the coordinator supplies an explicitly test-only database. +SQLite adapter tests do not establish these lock/index/rollback guarantees. +""" + +from __future__ import annotations + +import asyncio +import os +from datetime import UTC, datetime, timedelta +from time import monotonic +from uuid import uuid4 + +import pytest +from infinity_context_adapters.features.memory_facts.id_generator import MemoryFactIdAdapter +from infinity_context_adapters.features.memory_facts.postgres_fact_store import ( + PostgresMemoryFactUnitOfWorkFactory, +) +from infinity_context_adapters.noop import SystemClock +from infinity_context_adapters.postgres import ( + build_async_engine, + build_session_factory, + upgrade_schema, +) +from infinity_context_adapters.postgres.models import ( + MemoryFactRelationRow, + MemoryFactRow, + MemoryFactVersionRow, + MemoryOutboxRow, + MemorySourceRefRow, +) +from infinity_context_core.features.memory_facts.public import ( + FactRelationConflict, + LinkFactsCommand, + LinkFactsHandler, + MemoryFactIdentity, + MemoryFactScope, + MemoryFactSnapshot, + MemoryFactSourceRef, + UnlinkFactRelationCommand, + UnlinkFactRelationHandler, + link_facts_in_transaction, +) +from postgres_test_database import PostgresTestDatabase +from sqlalchemy import func, select, text +from sqlalchemy.exc import IntegrityError + +from tests.adapters.feature_relation_thread_cases import exercise_thread_cases + +NOW = datetime(2026, 9, 8, tzinfo=UTC) + + +def test_postgres_generic_relation_contention_uniqueness_and_atomicity(): + url = os.getenv("INFINITY_CONTEXT_TEST_POSTGRES_URL") + if not url: + pytest.skip("Coordinator must supply INFINITY_CONTEXT_TEST_POSTGRES_URL (test-only)") + asyncio.run(_exercise(url)) + + +async def _exercise(url): + database = PostgresTestDatabase.from_url( + url, prefix="generic_relations", asyncpg=pytest.importorskip("asyncpg") + ) + await database.recreate() + engine = build_async_engine(database.app_url) + pending = None + try: + await upgrade_schema(engine) + sessions = build_session_factory(engine) + factory = PostgresMemoryFactUnitOfWorkFactory(session_factory=sessions, clock=SystemClock()) + scope = MemoryFactScope("relation-space", "relation-scope") + facts = tuple( + MemoryFactSnapshot( + identity=MemoryFactIdentity(name, scope), + text=f"Evidence {name}", + source_refs=(MemoryFactSourceRef("manual", name),), + created_at=NOW, + updated_at=NOW, + ) + for name in ("source", "target") + ) + async with factory() as uow: + for fact in facts: + await uow.facts.create(fact) + await uow.commit() + ids = MemoryFactIdAdapter(lambda prefix: f"{prefix}-{uuid4().hex}") + command = LinkFactsCommand(facts[0].identity, facts[1].identity, "supports", "evidence") + async with factory() as first, factory() as second: + winner = await link_facts_in_transaction(first, command, now=NOW, ids=ids) + second_pid = await second._session.scalar(text("SELECT pg_backend_pid()")) + pending = asyncio.create_task( + link_facts_in_transaction(second, command, now=NOW, ids=ids) + ) + deadline = monotonic() + 5 + async with sessions() as observer: + while monotonic() < deadline: + blocked = await observer.scalar( + text( + "SELECT wait_event_type = 'Lock' FROM pg_stat_activity WHERE pid = :pid" + ), + {"pid": second_pid}, + ) + if blocked: + break + assert not pending.done(), "second writer escaped the canonical lock" + await asyncio.sleep(0.02) + await observer.rollback() + else: + pytest.fail("second writer did not demonstrate PostgreSQL lock contention") + await first.commit() + replay = await asyncio.wait_for(pending, timeout=5) + pending = None + await second.commit() + assert winner == replay + handler = LinkFactsHandler(factory, SystemClock(), ids) + with pytest.raises(FactRelationConflict, match="observed_at"): + await handler.execute( + LinkFactsCommand( + facts[0].identity, + facts[1].identity, + "supports", + "evidence", + observed_at=NOW + timedelta(days=1), + ) + ) + # The existing partial index also rejects a writer bypassing feature locks. + async with sessions() as session: + existing = await session.get(MemoryFactRelationRow, winner.relation.relation_id) + session.add( + MemoryFactRelationRow( + id="bypass-duplicate", + space_id=scope.space_id, + memory_scope_id=scope.memory_scope_id, + source_fact_id="source", + target_fact_id="target", + relation_type="supports", + reason="bypass", + status="active", + observed_at=NOW, + created_at=NOW, + updated_at=NOW, + ) + ) + assert existing is not None + with pytest.raises(IntegrityError) as error: + await session.flush() + assert "uq_memory_fact_relation_active" in str(error.value) + await session.rollback() + # Caller-owned transaction failure leaves no relation, revision or outbox write. + with pytest.raises(RuntimeError, match="injected"): + async with factory() as uow: + await link_facts_in_transaction( + uow, + LinkFactsCommand( + facts[0].identity, facts[1].identity, "references", "rollback" + ), + now=NOW, + ids=ids, + ) + unrelated = await uow._session.get(MemoryFactRow, "source") + unrelated.text = "uncommitted canonical change" + await uow._session.flush() + raise RuntimeError("injected failure before commit") + async with sessions() as session: + assert (await session.get(MemoryFactRow, "source")).text == facts[0].text + assert ( + await session.scalar(select(func.count()).select_from(MemoryFactRelationRow)) == 1 + ) + assert await session.scalar(select(func.count()).select_from(MemoryFactVersionRow)) == 2 + assert await session.scalar(select(func.count()).select_from(MemorySourceRefRow)) == 2 + assert await session.scalar(select(func.count()).select_from(MemoryOutboxRow)) == 0 + unlink = UnlinkFactRelationHandler(factory, SystemClock()) + deleted = await unlink.execute( + UnlinkFactRelationCommand(winner.relation.relation_id, scope) + ) + assert ( + await unlink.execute(UnlinkFactRelationCommand(winner.relation.relation_id, scope)) + == deleted + ) + relinked = await handler.execute(command) + assert relinked.relation.relation_id != winner.relation.relation_id + async with sessions() as session: + assert ( + await session.scalar( + select(func.count()) + .select_from(MemoryFactRelationRow) + .where(MemoryFactRelationRow.status == "active") + ) + == 1 + ) + await exercise_thread_cases(engine, sessions, factory, SystemClock(), ids) + finally: + if pending is not None: + pending.cancel() + await asyncio.gather(pending, return_exceptions=True) + await engine.dispose() + await database.drop()