From ce9ac08bcb60a91eb92810f803aa8cdcb4388b64 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 28 Dec 2025 18:11:43 +0000 Subject: [PATCH 1/4] Initial plan From db25c43dee69f3a95804139edc1243913139f02d Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 28 Dec 2025 18:37:04 +0000 Subject: [PATCH 2/4] Implement Axiom CRUD operations (DL-13) Co-authored-by: spuentesp <112034353+spuentesp@users.noreply.github.com> --- .../src/monitor_data/middleware/auth.py | 2 + .../src/monitor_data/schemas/__init__.py | 15 + .../src/monitor_data/schemas/axioms.py | 120 ++++ .../src/monitor_data/tools/neo4j_tools.py | 385 ++++++++++++ .../tests/test_tools/test_axiom_tools.py | 556 ++++++++++++++++++ 5 files changed, 1078 insertions(+) create mode 100644 packages/data-layer/src/monitor_data/schemas/axioms.py create mode 100644 packages/data-layer/tests/test_tools/test_axiom_tools.py diff --git a/packages/data-layer/src/monitor_data/middleware/auth.py b/packages/data-layer/src/monitor_data/middleware/auth.py index 5dcc057..1bc35cf 100644 --- a/packages/data-layer/src/monitor_data/middleware/auth.py +++ b/packages/data-layer/src/monitor_data/middleware/auth.py @@ -80,6 +80,8 @@ "neo4j_create_axiom": ["CanonKeeper"], "neo4j_get_axiom": ["*"], "neo4j_list_axioms": ["*"], + "neo4j_update_axiom": ["CanonKeeper"], + "neo4j_delete_axiom": ["CanonKeeper"], # ========================================================================= # NEO4J OPERATIONS - Plot Threads # ========================================================================= diff --git a/packages/data-layer/src/monitor_data/schemas/__init__.py b/packages/data-layer/src/monitor_data/schemas/__init__.py index a41bf5f..c0e7e09 100644 --- a/packages/data-layer/src/monitor_data/schemas/__init__.py +++ b/packages/data-layer/src/monitor_data/schemas/__init__.py @@ -46,6 +46,14 @@ MultiverseUpdate, MultiverseResponse, ) +from monitor_data.schemas.axioms import ( + AxiomDomain, + AxiomCreate, + AxiomUpdate, + AxiomResponse, + AxiomFilter, + AxiomListResponse, +) # from monitor_data.schemas.entities import * # from monitor_data.schemas.facts import * @@ -79,4 +87,11 @@ "MultiverseCreate", "MultiverseUpdate", "MultiverseResponse", + # Axiom schemas + "AxiomDomain", + "AxiomCreate", + "AxiomUpdate", + "AxiomResponse", + "AxiomFilter", + "AxiomListResponse", ] diff --git a/packages/data-layer/src/monitor_data/schemas/axioms.py b/packages/data-layer/src/monitor_data/schemas/axioms.py new file mode 100644 index 0000000..079fe22 --- /dev/null +++ b/packages/data-layer/src/monitor_data/schemas/axioms.py @@ -0,0 +1,120 @@ +""" +Pydantic schemas for Axiom operations. + +LAYER: 1 (data-layer) +IMPORTS FROM: External libraries (pydantic, uuid, datetime) and base schemas +CALLED BY: neo4j_tools.py + +These schemas define the data contracts for Axiom CRUD operations. +Axioms represent foundational world rules and constraints tied to universes. +""" + +from datetime import datetime +from typing import Optional, List, Dict, Any +from uuid import UUID +from enum import Enum + +from pydantic import BaseModel, Field + +from monitor_data.schemas.base import AxiomAuthority, CanonLevel + + +# ============================================================================= +# ENUMS +# ============================================================================= + + +class AxiomDomain(str, Enum): + """Domain classification for axioms.""" + + PHYSICS = "physics" + MAGIC = "magic" + SOCIETY = "society" + METAPHYSICS = "metaphysics" + + +# ============================================================================= +# AXIOM SCHEMAS +# ============================================================================= + + +class AxiomCreate(BaseModel): + """Request to create an Axiom.""" + + universe_id: UUID + statement: str = Field( + min_length=1, + max_length=2000, + description="Foundational world rule (e.g., 'magic requires verbal components')", + ) + domain: AxiomDomain + confidence: float = Field( + ge=0.0, + le=1.0, + default=1.0, + description="Confidence level in this axiom's truth", + ) + authority: AxiomAuthority = Field(default=AxiomAuthority.SYSTEM) + canon_level: CanonLevel = Field(default=CanonLevel.CANON) + source_ids: List[UUID] = Field( + default_factory=list, + description="UUIDs of Source nodes that support this axiom (creates SUPPORTED_BY edges)", + ) + + +class AxiomUpdate(BaseModel): + """Request to update an Axiom. + + Only mutable fields can be updated: statement, confidence, canon_level. + """ + + statement: Optional[str] = Field(None, min_length=1, max_length=2000) + confidence: Optional[float] = Field(None, ge=0.0, le=1.0) + canon_level: Optional[CanonLevel] = None + + +class AxiomResponse(BaseModel): + """Response with Axiom data.""" + + id: UUID + universe_id: UUID + statement: str + domain: AxiomDomain + confidence: float + canon_level: CanonLevel + authority: AxiomAuthority + created_at: datetime + # Optional provenance chain (populated by neo4j_get_axiom with provenance) + sources: List[Dict[str, Any]] = Field( + default_factory=list, + description="Source nodes that support this axiom (from SUPPORTED_BY edges)", + ) + + model_config = {"from_attributes": True} + + +class AxiomFilter(BaseModel): + """Filter parameters for listing axioms.""" + + universe_id: Optional[UUID] = None + domain: Optional[AxiomDomain] = None + canon_level: Optional[CanonLevel] = None + confidence_min: Optional[float] = Field(None, ge=0.0, le=1.0) + confidence_max: Optional[float] = Field(None, ge=0.0, le=1.0) + limit: int = Field(default=50, ge=1, le=1000) + offset: int = Field(default=0, ge=0) + sort_by: str = Field( + default="created_at", description="Field to sort by: created_at, confidence" + ) + sort_order: str = Field( + default="desc", description="Sort order: asc, desc", pattern="^(asc|desc)$" + ) + + +class AxiomListResponse(BaseModel): + """Response with list of axioms and pagination info.""" + + axioms: List[AxiomResponse] + total: int + limit: int + offset: int diff --git a/packages/data-layer/src/monitor_data/tools/neo4j_tools.py b/packages/data-layer/src/monitor_data/tools/neo4j_tools.py index 8cccee2..ee1e09c 100644 --- a/packages/data-layer/src/monitor_data/tools/neo4j_tools.py +++ b/packages/data-layer/src/monitor_data/tools/neo4j_tools.py @@ -30,6 +30,13 @@ EntityListResponse, StateTagsUpdate, ) +from monitor_data.schemas.axioms import ( + AxiomCreate, + AxiomUpdate, + AxiomResponse, + AxiomFilter, + AxiomListResponse, +) # ============================================================================= @@ -1091,3 +1098,381 @@ def neo4j_set_state_tags( created_at=e["created_at"], updated_at=e.get("updated_at"), ) + + +# ============================================================================= +# AXIOM OPERATIONS +# ============================================================================= + + +def neo4j_create_axiom(params: AxiomCreate) -> AxiomResponse: + """ + Create a new Axiom node linked to a universe. + + Authority: CanonKeeper only + Use Case: DL-13 + + Args: + params: Axiom creation parameters + + Returns: + AxiomResponse with created axiom data + + Raises: + ValueError: If universe_id doesn't exist or source_ids don't exist + """ + client = get_neo4j_client() + + # Verify universe exists + verify_query = """ + MATCH (u:Universe {id: $universe_id}) + RETURN u.id as id + """ + result = client.execute_read(verify_query, {"universe_id": str(params.universe_id)}) + if not result: + raise ValueError(f"Universe {params.universe_id} not found") + + # Verify all source_ids exist if provided + if params.source_ids: + verify_sources_query = """ + MATCH (s:Source) + WHERE s.id IN $source_ids + RETURN collect(s.id) as found_ids + """ + result = client.execute_read( + verify_sources_query, + {"source_ids": [str(sid) for sid in params.source_ids]}, + ) + found_ids = set(result[0]["found_ids"]) if result else set() + expected_ids = {str(sid) for sid in params.source_ids} + missing_ids = expected_ids - found_ids + if missing_ids: + raise ValueError(f"Source IDs not found: {missing_ids}") + + # Create axiom + axiom_id = uuid4() + created_at = datetime.now(timezone.utc) + + create_query = """ + MATCH (u:Universe {id: $universe_id}) + CREATE (a:Axiom { + id: $id, + universe_id: $universe_id, + statement: $statement, + domain: $domain, + confidence: $confidence, + canon_level: $canon_level, + authority: $authority, + created_at: datetime($created_at) + }) + CREATE (u)-[:HAS_AXIOM]->(a) + RETURN a + """ + client.execute_write( + create_query, + { + "id": str(axiom_id), + "universe_id": str(params.universe_id), + "statement": params.statement, + "domain": params.domain.value, + "confidence": params.confidence, + "canon_level": params.canon_level.value, + "authority": params.authority.value, + "created_at": created_at.isoformat(), + }, + ) + + # Create SUPPORTED_BY edges to sources if provided + if params.source_ids: + link_query = """ + MATCH (a:Axiom {id: $axiom_id}) + MATCH (s:Source) + WHERE s.id IN $source_ids + CREATE (a)-[:SUPPORTED_BY]->(s) + """ + client.execute_write( + link_query, + { + "axiom_id": str(axiom_id), + "source_ids": [str(sid) for sid in params.source_ids], + }, + ) + + return AxiomResponse( + id=axiom_id, + universe_id=params.universe_id, + statement=params.statement, + domain=params.domain, + confidence=params.confidence, + canon_level=params.canon_level, + authority=params.authority, + created_at=created_at, + sources=[], + ) + + +def neo4j_get_axiom(axiom_id: UUID, include_provenance: bool = True) -> Optional[AxiomResponse]: + """ + Get an Axiom by ID with optional provenance chain. + + Authority: Any agent (read-only) + Use Case: DL-13 + + Args: + axiom_id: UUID of the axiom + include_provenance: If True, include SUPPORTED_BY sources in response + + Returns: + AxiomResponse if found, None otherwise + """ + client = get_neo4j_client() + + if include_provenance: + query = """ + MATCH (a:Axiom {id: $id}) + OPTIONAL MATCH (a)-[:SUPPORTED_BY]->(s:Source) + RETURN a, collect(DISTINCT { + id: s.id, + title: s.title, + source_type: s.source_type, + canon_level: s.canon_level + }) as sources + """ + else: + query = """ + MATCH (a:Axiom {id: $id}) + RETURN a, [] as sources + """ + + result = client.execute_read(query, {"id": str(axiom_id)}) + + if not result: + return None + + a = result[0]["a"] + sources = result[0].get("sources", []) + # Filter out sources with None values (from OPTIONAL MATCH when no sources exist) + sources = [s for s in sources if s.get("id") is not None] + + return AxiomResponse( + id=UUID(a["id"]), + universe_id=UUID(a["universe_id"]), + statement=a["statement"], + domain=a["domain"], + confidence=a["confidence"], + canon_level=a["canon_level"], + authority=a["authority"], + created_at=a["created_at"], + sources=sources, + ) + + +def neo4j_list_axioms(filters: Optional[AxiomFilter] = None) -> AxiomListResponse: + """ + List axioms with filtering, pagination, and sorting. + + Authority: Any agent (read-only) + Use Case: DL-13 + + Args: + filters: Optional filter parameters (universe_id, domain, confidence, limit, offset) + + Returns: + AxiomListResponse with axioms and pagination info + """ + client = get_neo4j_client() + + if filters is None: + filters = AxiomFilter() + + # Build WHERE clause + where_clauses = [] + params: Dict[str, Any] = {} + + if filters.universe_id: + where_clauses.append("a.universe_id = $universe_id") + params["universe_id"] = str(filters.universe_id) + + if filters.domain: + where_clauses.append("a.domain = $domain") + params["domain"] = filters.domain.value + + if filters.canon_level: + where_clauses.append("a.canon_level = $canon_level") + params["canon_level"] = filters.canon_level.value + + if filters.confidence_min is not None: + where_clauses.append("a.confidence >= $confidence_min") + params["confidence_min"] = filters.confidence_min + + if filters.confidence_max is not None: + where_clauses.append("a.confidence <= $confidence_max") + params["confidence_max"] = filters.confidence_max + + where_clause = "WHERE " + " AND ".join(where_clauses) if where_clauses else "" + + # Build ORDER BY clause + sort_field_map = {"created_at": "a.created_at", "confidence": "a.confidence"} + sort_field = sort_field_map.get(filters.sort_by, "a.created_at") + sort_order = "DESC" if filters.sort_order == "desc" else "ASC" + + # Count total + count_query = f""" + MATCH (a:Axiom) + {where_clause} + RETURN count(a) as total + """ + count_result = client.execute_read(count_query, params) + total = count_result[0]["total"] + + # Get axioms with pagination + list_query = f""" + MATCH (a:Axiom) + {where_clause} + RETURN a + ORDER BY {sort_field} {sort_order} + SKIP $offset + LIMIT $limit + """ + params["offset"] = filters.offset + params["limit"] = filters.limit + + result = client.execute_read(list_query, params) + + axioms = [] + for record in result: + a = record["a"] + axioms.append( + AxiomResponse( + id=UUID(a["id"]), + universe_id=UUID(a["universe_id"]), + statement=a["statement"], + domain=a["domain"], + confidence=a["confidence"], + canon_level=a["canon_level"], + authority=a["authority"], + created_at=a["created_at"], + sources=[], # Not included in list view for performance + ) + ) + + return AxiomListResponse( + axioms=axioms, total=total, limit=filters.limit, offset=filters.offset + ) + + +def neo4j_update_axiom(axiom_id: UUID, params: AxiomUpdate) -> AxiomResponse: + """ + Update an Axiom's mutable fields. + + Authority: CanonKeeper only + Use Case: DL-13 + + Args: + axiom_id: UUID of the axiom to update + params: Update parameters (statement, confidence, canon_level) + + Returns: + AxiomResponse with updated axiom data + + Raises: + ValueError: If axiom doesn't exist + """ + client = get_neo4j_client() + + # Verify axiom exists + verify_query = """ + MATCH (a:Axiom {id: $id}) + RETURN a.id as id + """ + result = client.execute_read(verify_query, {"id": str(axiom_id)}) + if not result: + raise ValueError(f"Axiom {axiom_id} not found") + + # Build SET clauses for updates + set_clauses = [] + update_params: Dict[str, Any] = {"id": str(axiom_id)} + + if params.statement is not None: + set_clauses.append("a.statement = $statement") + update_params["statement"] = params.statement + + if params.confidence is not None: + set_clauses.append("a.confidence = $confidence") + update_params["confidence"] = params.confidence + + if params.canon_level is not None: + set_clauses.append("a.canon_level = $canon_level") + update_params["canon_level"] = params.canon_level.value + + if not set_clauses: + # No updates, return current state + result = neo4j_get_axiom(axiom_id, include_provenance=False) + if result is None: + raise ValueError(f"Axiom {axiom_id} not found after verification") + return result + + set_clause = ", ".join(set_clauses) + update_query = f""" + MATCH (a:Axiom {{id: $id}}) + SET {set_clause} + RETURN a + """ + + result = client.execute_write(update_query, update_params) + a = result[0]["a"] + + return AxiomResponse( + id=UUID(a["id"]), + universe_id=UUID(a["universe_id"]), + statement=a["statement"], + domain=a["domain"], + confidence=a["confidence"], + canon_level=a["canon_level"], + authority=a["authority"], + created_at=a["created_at"], + sources=[], + ) + + +def neo4j_delete_axiom(axiom_id: UUID) -> Dict[str, Any]: + """ + Soft-delete an Axiom by setting canon_level to 'retconned'. + + Authority: CanonKeeper only + Use Case: DL-13 + + Args: + axiom_id: UUID of the axiom to soft-delete + + Returns: + Dict with deletion status + + Raises: + ValueError: If axiom doesn't exist + """ + client = get_neo4j_client() + + # Verify axiom exists + verify_query = """ + MATCH (a:Axiom {id: $id}) + RETURN a.id as id + """ + result = client.execute_read(verify_query, {"id": str(axiom_id)}) + if not result: + raise ValueError(f"Axiom {axiom_id} not found") + + # Soft-delete by setting canon_level to 'retconned' + delete_query = """ + MATCH (a:Axiom {id: $id}) + SET a.canon_level = 'retconned' + RETURN a + """ + result = client.execute_write(delete_query, {"id": str(axiom_id)}) + + return { + "axiom_id": str(axiom_id), + "deleted": True, + "method": "soft-delete", + "canon_level": "retconned", + } diff --git a/packages/data-layer/tests/test_tools/test_axiom_tools.py b/packages/data-layer/tests/test_tools/test_axiom_tools.py new file mode 100644 index 0000000..3fde10c --- /dev/null +++ b/packages/data-layer/tests/test_tools/test_axiom_tools.py @@ -0,0 +1,556 @@ +""" +Unit tests for Neo4j axiom operations (DL-13). + +Tests cover: +- neo4j_create_axiom (with and without provenance) +- neo4j_get_axiom (with provenance chain) +- neo4j_list_axioms (with filtering) +- neo4j_update_axiom +- neo4j_delete_axiom (soft-delete) +- Axiom lifecycle integration +""" + +from typing import Dict, Any +from unittest.mock import Mock, patch +from uuid import UUID, uuid4 + +import pytest + +from monitor_data.schemas.axioms import ( + AxiomCreate, + AxiomUpdate, + AxiomFilter, + AxiomDomain, +) +from monitor_data.schemas.base import CanonLevel, AxiomAuthority +from monitor_data.tools.neo4j_tools import ( + neo4j_create_axiom, + neo4j_get_axiom, + neo4j_list_axioms, + neo4j_update_axiom, + neo4j_delete_axiom, +) + + +# ============================================================================= +# TEST FIXTURES +# ============================================================================= + + +@pytest.fixture +def axiom_data(universe_data: Dict[str, Any]) -> Dict[str, Any]: + """Provide sample axiom data.""" + return { + "id": str(uuid4()), + "universe_id": universe_data["id"], + "statement": "Magic requires verbal components", + "domain": AxiomDomain.MAGIC.value, + "confidence": 1.0, + "canon_level": CanonLevel.CANON.value, + "authority": AxiomAuthority.SYSTEM.value, + "created_at": "2024-01-01T00:00:00", + } + + +@pytest.fixture +def source_data(universe_data: Dict[str, Any]) -> Dict[str, Any]: + """Provide sample source data.""" + return { + "id": str(uuid4()), + "universe_id": universe_data["id"], + "title": "Core Rulebook", + "source_type": "rulebook", + "canon_level": "authoritative", + } + + +# ============================================================================= +# TESTS: neo4j_create_axiom +# ============================================================================= + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_create_axiom_success( + mock_get_client: Mock, + mock_neo4j_client: Mock, + universe_data: Dict[str, Any], + axiom_data: Dict[str, Any], +): + """Test creating an Axiom with valid parameters.""" + mock_get_client.return_value = mock_neo4j_client + + # Mock universe exists + mock_neo4j_client.execute_read.return_value = [{"id": universe_data["id"]}] + + # Mock axiom creation + mock_neo4j_client.execute_write.return_value = [{"a": axiom_data}] + + params = AxiomCreate( + universe_id=UUID(universe_data["id"]), + statement="Magic requires verbal components", + domain=AxiomDomain.MAGIC, + confidence=1.0, + ) + + result = neo4j_create_axiom(params) + + assert result.statement == "Magic requires verbal components" + assert result.domain == AxiomDomain.MAGIC + assert result.confidence == 1.0 + assert result.canon_level == CanonLevel.CANON + assert result.sources == [] + assert mock_neo4j_client.execute_read.call_count == 1 + assert mock_neo4j_client.execute_write.call_count == 1 + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_create_axiom_with_provenance( + mock_get_client: Mock, + mock_neo4j_client: Mock, + universe_data: Dict[str, Any], + axiom_data: Dict[str, Any], + source_data: Dict[str, Any], +): + """Test creating an Axiom with source provenance.""" + mock_get_client.return_value = mock_neo4j_client + + # Mock universe exists + mock_neo4j_client.execute_read.side_effect = [ + [{"id": universe_data["id"]}], # Universe verification + [{"found_ids": [source_data["id"]]}], # Source verification + ] + + # Mock axiom creation + mock_neo4j_client.execute_write.return_value = [{"a": axiom_data}] + + params = AxiomCreate( + universe_id=UUID(universe_data["id"]), + statement="Magic requires verbal components", + domain=AxiomDomain.MAGIC, + source_ids=[UUID(source_data["id"])], + ) + + result = neo4j_create_axiom(params) + + assert result.statement == "Magic requires verbal components" + assert result.sources == [] # Sources not loaded in create response + # Verify that execute_write was called twice (create + link) + assert mock_neo4j_client.execute_write.call_count == 2 + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_create_axiom_domain( + mock_get_client: Mock, + mock_neo4j_client: Mock, + universe_data: Dict[str, Any], +): + """Test creating axioms with each domain value.""" + mock_get_client.return_value = mock_neo4j_client + + # Mock universe exists + mock_neo4j_client.execute_read.return_value = [{"id": universe_data["id"]}] + + domains = [ + AxiomDomain.PHYSICS, + AxiomDomain.MAGIC, + AxiomDomain.SOCIETY, + AxiomDomain.METAPHYSICS, + ] + + for domain in domains: + axiom_data = { + "id": str(uuid4()), + "universe_id": universe_data["id"], + "statement": f"Test axiom for {domain.value}", + "domain": domain.value, + "confidence": 1.0, + "canon_level": CanonLevel.CANON.value, + "authority": AxiomAuthority.SYSTEM.value, + "created_at": "2024-01-01T00:00:00", + } + mock_neo4j_client.execute_write.return_value = [{"a": axiom_data}] + + params = AxiomCreate( + universe_id=UUID(universe_data["id"]), + statement=f"Test axiom for {domain.value}", + domain=domain, + ) + + result = neo4j_create_axiom(params) + assert result.domain == domain + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_create_axiom_universe_not_found( + mock_get_client: Mock, + mock_neo4j_client: Mock, +): + """Test creating an Axiom with nonexistent universe_id.""" + mock_get_client.return_value = mock_neo4j_client + + # Mock universe not found + mock_neo4j_client.execute_read.return_value = [] + + params = AxiomCreate( + universe_id=uuid4(), + statement="Test axiom", + domain=AxiomDomain.PHYSICS, + ) + + with pytest.raises(ValueError, match="Universe .* not found"): + neo4j_create_axiom(params) + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_create_axiom_source_not_found( + mock_get_client: Mock, + mock_neo4j_client: Mock, + universe_data: Dict[str, Any], +): + """Test creating an Axiom with nonexistent source_id.""" + mock_get_client.return_value = mock_neo4j_client + + # Mock universe exists but source doesn't + mock_neo4j_client.execute_read.side_effect = [ + [{"id": universe_data["id"]}], # Universe found + [{"found_ids": []}], # Source not found + ] + + params = AxiomCreate( + universe_id=UUID(universe_data["id"]), + statement="Test axiom", + domain=AxiomDomain.PHYSICS, + source_ids=[uuid4()], + ) + + with pytest.raises(ValueError, match="Source IDs not found"): + neo4j_create_axiom(params) + + +# ============================================================================= +# TESTS: neo4j_get_axiom +# ============================================================================= + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_get_axiom_with_provenance( + mock_get_client: Mock, + mock_neo4j_client: Mock, + axiom_data: Dict[str, Any], + source_data: Dict[str, Any], +): + """Test retrieving an Axiom with provenance chain.""" + mock_get_client.return_value = mock_neo4j_client + + # Mock axiom with sources + mock_neo4j_client.execute_read.return_value = [ + { + "a": axiom_data, + "sources": [source_data], + } + ] + + result = neo4j_get_axiom(UUID(axiom_data["id"]), include_provenance=True) + + assert result is not None + assert result.id == UUID(axiom_data["id"]) + assert result.statement == axiom_data["statement"] + assert len(result.sources) == 1 + assert result.sources[0]["id"] == source_data["id"] + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_get_axiom_without_provenance( + mock_get_client: Mock, + mock_neo4j_client: Mock, + axiom_data: Dict[str, Any], +): + """Test retrieving an Axiom without provenance.""" + mock_get_client.return_value = mock_neo4j_client + + # Mock axiom without sources + mock_neo4j_client.execute_read.return_value = [ + { + "a": axiom_data, + "sources": [], + } + ] + + result = neo4j_get_axiom(UUID(axiom_data["id"]), include_provenance=False) + + assert result is not None + assert result.id == UUID(axiom_data["id"]) + assert result.sources == [] + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_get_axiom_not_found( + mock_get_client: Mock, + mock_neo4j_client: Mock, +): + """Test retrieving a nonexistent Axiom.""" + mock_get_client.return_value = mock_neo4j_client + + # Mock axiom not found + mock_neo4j_client.execute_read.return_value = [] + + result = neo4j_get_axiom(uuid4()) + + assert result is None + + +# ============================================================================= +# TESTS: neo4j_list_axioms +# ============================================================================= + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_list_axioms_by_domain( + mock_get_client: Mock, + mock_neo4j_client: Mock, + universe_data: Dict[str, Any], +): + """Test listing axioms filtered by domain.""" + mock_get_client.return_value = mock_neo4j_client + + axiom1 = { + "id": str(uuid4()), + "universe_id": universe_data["id"], + "statement": "Magic axiom", + "domain": AxiomDomain.MAGIC.value, + "confidence": 1.0, + "canon_level": CanonLevel.CANON.value, + "authority": AxiomAuthority.SYSTEM.value, + "created_at": "2024-01-01T00:00:00", + } + + # Mock count and list queries + mock_neo4j_client.execute_read.side_effect = [ + [{"total": 1}], # Count + [{"a": axiom1}], # List + ] + + filters = AxiomFilter( + universe_id=UUID(universe_data["id"]), + domain=AxiomDomain.MAGIC, + ) + + result = neo4j_list_axioms(filters) + + assert result.total == 1 + assert len(result.axioms) == 1 + assert result.axioms[0].domain == AxiomDomain.MAGIC + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_list_axioms_by_confidence( + mock_get_client: Mock, + mock_neo4j_client: Mock, + universe_data: Dict[str, Any], +): + """Test listing axioms filtered by confidence range.""" + mock_get_client.return_value = mock_neo4j_client + + axiom1 = { + "id": str(uuid4()), + "universe_id": universe_data["id"], + "statement": "High confidence axiom", + "domain": AxiomDomain.PHYSICS.value, + "confidence": 0.9, + "canon_level": CanonLevel.CANON.value, + "authority": AxiomAuthority.SYSTEM.value, + "created_at": "2024-01-01T00:00:00", + } + + # Mock count and list queries + mock_neo4j_client.execute_read.side_effect = [ + [{"total": 1}], # Count + [{"a": axiom1}], # List + ] + + filters = AxiomFilter( + confidence_min=0.8, + confidence_max=1.0, + ) + + result = neo4j_list_axioms(filters) + + assert result.total == 1 + assert len(result.axioms) == 1 + assert result.axioms[0].confidence >= 0.8 + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_list_axioms_empty( + mock_get_client: Mock, + mock_neo4j_client: Mock, +): + """Test listing axioms when none exist.""" + mock_get_client.return_value = mock_neo4j_client + + # Mock empty result + mock_neo4j_client.execute_read.side_effect = [ + [{"total": 0}], # Count + [], # List + ] + + result = neo4j_list_axioms() + + assert result.total == 0 + assert len(result.axioms) == 0 + + +# ============================================================================= +# TESTS: neo4j_update_axiom +# ============================================================================= + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_update_axiom( + mock_get_client: Mock, + mock_neo4j_client: Mock, + axiom_data: Dict[str, Any], +): + """Test updating an Axiom.""" + mock_get_client.return_value = mock_neo4j_client + + # Mock axiom exists + mock_neo4j_client.execute_read.return_value = [{"id": axiom_data["id"]}] + + # Mock updated axiom + updated_axiom = axiom_data.copy() + updated_axiom["statement"] = "Updated statement" + updated_axiom["confidence"] = 0.8 + mock_neo4j_client.execute_write.return_value = [{"a": updated_axiom}] + + params = AxiomUpdate( + statement="Updated statement", + confidence=0.8, + ) + + result = neo4j_update_axiom(UUID(axiom_data["id"]), params) + + assert result.statement == "Updated statement" + assert result.confidence == 0.8 + assert mock_neo4j_client.execute_write.call_count == 1 + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_update_axiom_not_found( + mock_get_client: Mock, + mock_neo4j_client: Mock, +): + """Test updating a nonexistent Axiom.""" + mock_get_client.return_value = mock_neo4j_client + + # Mock axiom not found + mock_neo4j_client.execute_read.return_value = [] + + params = AxiomUpdate(statement="Test") + + with pytest.raises(ValueError, match="Axiom .* not found"): + neo4j_update_axiom(uuid4(), params) + + +# ============================================================================= +# TESTS: neo4j_delete_axiom +# ============================================================================= + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_delete_axiom_soft_delete( + mock_get_client: Mock, + mock_neo4j_client: Mock, + axiom_data: Dict[str, Any], +): + """Test soft-deleting an Axiom (sets canon_level to retconned).""" + mock_get_client.return_value = mock_neo4j_client + + # Mock axiom exists + mock_neo4j_client.execute_read.return_value = [{"id": axiom_data["id"]}] + + # Mock soft-delete + retconned_axiom = axiom_data.copy() + retconned_axiom["canon_level"] = CanonLevel.RETCONNED.value + mock_neo4j_client.execute_write.return_value = [{"a": retconned_axiom}] + + result = neo4j_delete_axiom(UUID(axiom_data["id"])) + + assert result["deleted"] is True + assert result["method"] == "soft-delete" + assert result["canon_level"] == "retconned" + assert mock_neo4j_client.execute_write.call_count == 1 + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_delete_axiom_not_found( + mock_get_client: Mock, + mock_neo4j_client: Mock, +): + """Test deleting a nonexistent Axiom.""" + mock_get_client.return_value = mock_neo4j_client + + # Mock axiom not found + mock_neo4j_client.execute_read.return_value = [] + + with pytest.raises(ValueError, match="Axiom .* not found"): + neo4j_delete_axiom(uuid4()) + + +# ============================================================================= +# INTEGRATION TESTS +# ============================================================================= + + +@patch("monitor_data.tools.neo4j_tools.get_neo4j_client") +def test_axiom_lifecycle( + mock_get_client: Mock, + mock_neo4j_client: Mock, + universe_data: Dict[str, Any], +): + """Test full axiom lifecycle: create → update → soft-delete.""" + mock_get_client.return_value = mock_neo4j_client + + axiom_id = uuid4() + + # Step 1: Create + mock_neo4j_client.execute_read.return_value = [{"id": universe_data["id"]}] + created_axiom = { + "id": str(axiom_id), + "universe_id": universe_data["id"], + "statement": "Original statement", + "domain": AxiomDomain.PHYSICS.value, + "confidence": 1.0, + "canon_level": CanonLevel.CANON.value, + "authority": AxiomAuthority.SYSTEM.value, + "created_at": "2024-01-01T00:00:00", + } + mock_neo4j_client.execute_write.return_value = [{"a": created_axiom}] + + create_params = AxiomCreate( + universe_id=UUID(universe_data["id"]), + statement="Original statement", + domain=AxiomDomain.PHYSICS, + ) + result = neo4j_create_axiom(create_params) + assert result.statement == "Original statement" + + # Step 2: Update + mock_neo4j_client.execute_read.return_value = [{"id": str(axiom_id)}] + updated_axiom = created_axiom.copy() + updated_axiom["statement"] = "Updated statement" + mock_neo4j_client.execute_write.return_value = [{"a": updated_axiom}] + + update_params = AxiomUpdate(statement="Updated statement") + result = neo4j_update_axiom(axiom_id, update_params) + assert result.statement == "Updated statement" + + # Step 3: Soft-delete + mock_neo4j_client.execute_read.return_value = [{"id": str(axiom_id)}] + retconned_axiom = updated_axiom.copy() + retconned_axiom["canon_level"] = CanonLevel.RETCONNED.value + mock_neo4j_client.execute_write.return_value = [{"a": retconned_axiom}] + + result = neo4j_delete_axiom(axiom_id) + assert result["deleted"] is True + assert result["canon_level"] == "retconned" From 3de08f7c24a0a5cf57d1bc89180a190f7d4fcc63 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 28 Dec 2025 18:38:17 +0000 Subject: [PATCH 3/4] Format neo4j_tools.py with black Co-authored-by: spuentesp <112034353+spuentesp@users.noreply.github.com> --- .../data-layer/src/monitor_data/tools/neo4j_tools.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/packages/data-layer/src/monitor_data/tools/neo4j_tools.py b/packages/data-layer/src/monitor_data/tools/neo4j_tools.py index ee1e09c..cd78dbb 100644 --- a/packages/data-layer/src/monitor_data/tools/neo4j_tools.py +++ b/packages/data-layer/src/monitor_data/tools/neo4j_tools.py @@ -421,9 +421,7 @@ def neo4j_update_universe( set_clause = ", ".join(set_clauses) update_query = ( - "MATCH (u:Universe {id: $id})\n" - "SET " + set_clause + "\n" - "RETURN u" + "MATCH (u:Universe {id: $id})\n" "SET " + set_clause + "\n" "RETURN u" ) result = client.execute_write(update_query, update_params) @@ -1009,9 +1007,7 @@ def neo4j_delete_entity(entity_id: UUID, force: bool = False) -> Dict[str, Any]: } -def neo4j_set_state_tags( - entity_id: UUID, params: StateTagsUpdate -) -> EntityResponse: +def neo4j_set_state_tags(entity_id: UUID, params: StateTagsUpdate) -> EntityResponse: """ Atomically add/remove state tags on an EntityInstance. @@ -1211,7 +1207,9 @@ def neo4j_create_axiom(params: AxiomCreate) -> AxiomResponse: ) -def neo4j_get_axiom(axiom_id: UUID, include_provenance: bool = True) -> Optional[AxiomResponse]: +def neo4j_get_axiom( + axiom_id: UUID, include_provenance: bool = True +) -> Optional[AxiomResponse]: """ Get an Axiom by ID with optional provenance chain. From 698ab4259f89657652dd59cf6b734e695f8beb13 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 28 Dec 2025 18:41:14 +0000 Subject: [PATCH 4/4] Fix hard-coded strings to use CanonLevel enum Co-authored-by: spuentesp <112034353+spuentesp@users.noreply.github.com> --- .../data-layer/src/monitor_data/tools/neo4j_tools.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/packages/data-layer/src/monitor_data/tools/neo4j_tools.py b/packages/data-layer/src/monitor_data/tools/neo4j_tools.py index cd78dbb..de9f362 100644 --- a/packages/data-layer/src/monitor_data/tools/neo4j_tools.py +++ b/packages/data-layer/src/monitor_data/tools/neo4j_tools.py @@ -14,6 +14,7 @@ from uuid import UUID, uuid4 from monitor_data.db.neo4j import get_neo4j_client +from monitor_data.schemas.base import CanonLevel from monitor_data.schemas.universe import ( UniverseCreate, UniverseUpdate, @@ -1463,14 +1464,17 @@ def neo4j_delete_axiom(axiom_id: UUID) -> Dict[str, Any]: # Soft-delete by setting canon_level to 'retconned' delete_query = """ MATCH (a:Axiom {id: $id}) - SET a.canon_level = 'retconned' + SET a.canon_level = $canon_level RETURN a """ - result = client.execute_write(delete_query, {"id": str(axiom_id)}) + result = client.execute_write( + delete_query, + {"id": str(axiom_id), "canon_level": CanonLevel.RETCONNED.value}, + ) return { "axiom_id": str(axiom_id), "deleted": True, "method": "soft-delete", - "canon_level": "retconned", + "canon_level": CanonLevel.RETCONNED.value, }