fix all tests for observation category

2026-06-21 13:47:35 +00:00 · 2024-12-25 14:16:46 -06:00
parent fd723164ca
commit bb47fbd48b
12 changed files with 536 additions and 136 deletions
@@ -1,53 +0,0 @@
---
-id: 5
-created: '2024-12-24T02:30:29.343588+00:00'
-modified: '2024-12-24T02:30:29.343588+00:00'
-type: test
-tags:
- obsidian
- markdown
- documentation
-created_by: Claude
-status: draft
---
-
-# Obsidian Test Document
-
-This is a test of how documents appear in Obsidian's interface.
-
-## Links and Tags
-We can use:
- Standard markdown links like [Basic Memory](basic-memory)
- Tags like #test #documentation 
- Embeds like ![[basic-memory]]
-
-## Features to Test
-### Knowledge Graph
-This document should show up in the knowledge graph with connections to:
- [[Basic_Memory]] project
- [[Knowledge_Graph_Structure]] which implements it
- [[Development_Process]] that guides it
-
-### Backlinks
-Any document that links to this one should appear in the backlinks panel.
-
-### YAML Frontmatter
-Obsidian should display the frontmatter cleanly at the top of the document.
-
-### Code Blocks
-```python
-def test_function():
-    """Code blocks should have syntax highlighting"""
-    print("Testing display")
-```
-
-### Callouts
-> [!NOTE] 
-> Obsidian supports special callout blocks
-> They help organize important information
-
-### Task Lists
- [x] Create test document
- [x] Add various markdown features
- [ ] View in Obsidian
- [ ] Check graph visualization
@@ -1,6 +1,5 @@
 """Writer for knowledge entity markdown files."""

-from datetime import datetime, UTC
 from typing import Optional, Dict, Any

 import yaml
@@ -14,12 +13,11 @@ class KnowledgeWriter:

    async def format_frontmatter(self, entity: EntityModel) -> dict:
        """Generate frontmatter metadata for entity."""
-        now = datetime.now(UTC).isoformat()
        return {
            "type": entity.entity_type,
-            "id": entity.id,
-            "created": now,
-            "modified": now
+            "id": entity.path_id,
+            "created": entity.created_at.isoformat(),
+            "modified": entity.updated_at.isoformat(),
        }

    async def format_metadata(self, metadata: Optional[Dict[str, Any]] = None) -> str:
@@ -40,37 +38,50 @@ class KnowledgeWriter:
            logger.warning(f"Failed to format metadata YAML: {e}")
            return ""  # Skip metadata on error

-    async def format_content(self, entity: EntityModel, metadata: Optional[Dict[str, Any]] = None) -> str:
+    async def format_content(
+        self, entity: EntityModel, metadata: Optional[Dict[str, Any]] = None
+    ) -> str:
        """Format entity content as markdown."""
        sections = [
-            f"# {entity.name}\n"
+            f"# {entity.name}\n",
+            "",  # Empty line after name
        ]

        if entity.description:
-            sections.extend([
-                entity.description,
-                ""
-            ])
-            
+            sections.extend([entity.description, ""])
+
        if entity.observations:
-            sections.extend([
-                "## Observations",
-                *[f"- {obs.content}" for obs in entity.observations],
-                ""
-            ])
+            sections.extend(
+                [
+                    "## Observations",
+                    "<!-- Format: - [category] Content text #tag1 #tag2 (optional context) -->",
+                    "",  # Empty line after format comment
+                    *[
+                        f"- [{obs.category}] {obs.content}"
+                        + (f" ({obs.context})" if obs.context else "")
+                        for obs in entity.observations
+                    ],
+                    "",
+                ]
+            )

        # Format outgoing and incoming relations separately
        if entity.to_relations or entity.from_relations:
-            sections.append("## Relations")
-            
+            sections.extend(
+                [
+                    "## Relations",
+                    "",  # Empty line after format comment
+                ]
+            )
+
            # Outgoing relations
            for rel in entity.to_relations:
                sections.append(f"- [[{rel.from_entity.name}]] {rel.relation_type}")
-                
+
            # Incoming relations
            for rel in entity.from_relations:
                sections.append(f"- [[{rel.to_entity.name}]] {rel.relation_type}")
-                
+
            sections.append("")

        if metadata:
@@ -2,12 +2,13 @@

 from basic_memory.models.base import Base
 from basic_memory.models.documents import Document
-from basic_memory.models.knowledge import Entity, Observation, Relation
+from basic_memory.models.knowledge import Entity, Observation, Relation, ObservationCategory

 __all__ = [
    'Base',
    'Document',
    'Entity',
    'Observation', 
+    'ObservationCategory', 
    'Relation'
 ]
@@ -26,3 +26,15 @@ class ObservationRepository(Repository[Observation]):
        query = select(Observation).filter(Observation.context == context)
        result = await self.execute_query(query)
        return result.scalars().all()
+
+    async def find_by_category(self, category: str) -> Sequence[Observation]:
+        """Find observations with a specific context."""
+        query = select(Observation).filter(Observation.category == category)
+        result = await self.execute_query(query)
+        return result.scalars().all()
+
+    async def observation_categories(self) -> Sequence[str]:
+        """Return a list of all observation categories."""
+        query = select(Observation.category).distinct()
+        result = await self.execute_query(query)
+        return result.scalars().all()
@@ -127,6 +127,7 @@ class SearchNodesRequest(BaseModel):
    - Partial word matches
    - Returns full entity objects with relations
    - Includes all matching entities
+    - If a category is specified, only entities with that category are returned

    Example Queries:
    - "memory" - Find entities related to memory systems
@@ -143,6 +144,7 @@ class SearchNodesRequest(BaseModel):
    """

    query: Annotated[str, MinLen(1), MaxLen(200)]
+    category: Optional[ObservationCategory] = None


 class OpenNodesRequest(BaseModel):
@@ -7,6 +7,7 @@ from loguru import logger
 from basic_memory.models import Entity as EntityModel
 from basic_memory.services.exceptions import EntityNotFoundError
 from basic_memory.services.observation_service import ObservationService
+from basic_memory.schemas.request import ObservationCreate
 from .relations import RelationOperations


@@ -18,9 +19,22 @@ class ObservationOperations(RelationOperations):
        self.observation_service = observation_service

    async def add_observations(
-        self, path_id: str, observations: List[str], context: str | None = None
+        self, 
+        path_id: str, 
+        observations: List[ObservationCreate], 
+        context: str | None = None
    ) -> EntityModel:
-        """Add observations to entity and update its file."""
+        """Add observations to entity and update its file.
+        
+        Observations are added with their categories and written to both
+        the database and markdown file. The file format is:
+        - [category] Content text #tag1 #tag2 (optional context)
+        
+        Args:
+            path_id: Entity path ID
+            observations: List of observations with categories
+            context: Optional shared context for all observations
+        """
        logger.debug(f"Adding observations to entity {path_id}")

        try:
@@ -33,25 +47,34 @@ class ObservationOperations(RelationOperations):
            await self.observation_service.add_observations(entity.id, observations, context)

            # Get updated entity
-            updated_entity = await self.entity_service.get_by_path_id(path_id)
+            entity = await self.entity_service.get_by_path_id(path_id)

            # Write updated file and checksum
            _, checksum = await self.write_entity_file(entity)
            await self.entity_service.update_entity(path_id, {"checksum": checksum})

-            # query to fetch all relations
+            # Return final entity with all updates and relations
            return await self.entity_service.get_by_path_id(path_id)

        except Exception as e:
            logger.error(f"Failed to add observations: {e}")
            raise

-    async def delete_observations(self, path_id: str, observations: List[str]) -> EntityModel:
-        """Delete observations from entity and update its file."""
+    async def delete_observations(
+        self,
+        path_id: str, 
+        observations: List[str]
+    ) -> EntityModel:
+        """Delete observations from entity and update its file.
+        
+        Args:
+            path_id: Entity path ID
+            observations: List of observation contents to delete
+        """
        logger.debug(f"Deleting observations from entity {path_id}")

        try:
-            # Get updated entity
+            # Get entity
            entity = await self.entity_service.get_by_path_id(path_id)
            if not entity:
                raise EntityNotFoundError(f"Entity not found: {path_id}")
@@ -63,9 +86,9 @@ class ObservationOperations(RelationOperations):
            _, checksum = await self.write_entity_file(entity)
            await self.entity_service.update_entity(path_id, {"checksum": checksum})

-            # Get final entity with all updates
+            # Return final entity with all updates
            return await self.entity_service.get_by_path_id(path_id)

        except Exception as e:
            logger.error(f"Failed to delete observations: {e}")
-            raise
+            raise
@@ -1,6 +1,6 @@
 """Service for managing observations in the database."""

-from typing import List, Sequence
+from typing import List, Sequence, Optional

 from loguru import logger
 from sqlalchemy import select
@@ -8,6 +8,8 @@ from sqlalchemy import select
 from basic_memory.models import Observation as ObservationModel
 from basic_memory.repository.observation_repository import ObservationRepository
 from .service import BaseService
+from ..schemas.base import ObservationCategory
+from ..schemas.request import ObservationCreate


 class ObservationService(BaseService[ObservationRepository]):
@@ -20,13 +22,22 @@ class ObservationService(BaseService[ObservationRepository]):
        super().__init__(observation_repository)

    async def add_observations(
-        self, entity_id: int, observations: List[str], context: str | None = None
+        self,
+        entity_id: int,
+        observations: List[str | ObservationCreate],
+        context: Optional[str] = None,
    ) -> Sequence[ObservationModel]:
        """Add multiple observations to an entity."""
        logger.debug(f"Adding {len(observations)} observations to entity: {entity_id}")
        return await self.repository.create_all(
            [
-                dict(entity_id=entity_id, content=observation, context=context)
+                # unpack the ObservationCreate values if present
+                dict(
+                    entity_id=entity_id,
+                    content=getattr(observation, "content", observation), 
+                    context=context,
+                    category=getattr(observation, "category", None),
+                )
                for observation in observations
            ]
        )
@@ -46,14 +57,20 @@ class ObservationService(BaseService[ObservationRepository]):
        logger.debug(f"Deleting all observations for entity: {entity_id}")
        return await self.repository.delete_by_fields(entity_id=entity_id)

-    async def search_observations(self, query: str) -> List[ObservationModel]:
+    async def search_observations(self, query: str, category: Optional[ObservationCategory] = None) -> List[ObservationModel]:
        """Search for observations across all entities."""
        logger.debug(f"Searching observations with query: {query}")
-        result = await self.repository.execute_query(
-            select(ObservationModel).filter(
-                ObservationModel.content.contains(query) | ObservationModel.context.contains(query)
-            )
+
+        # Build base query
+        statement = select(ObservationModel).filter(
+            ObservationModel.content.contains(query) | ObservationModel.context.contains(query)
        )
+
+        # Add category filter if specified
+        if category:
+            statement = statement.filter(ObservationModel.category == category)
+
+        result = await self.repository.execute_query(statement)
        observations = result.scalars().all()
        return [ObservationModel(content=obs.content) for obs in observations]

@@ -61,3 +78,13 @@ class ObservationService(BaseService[ObservationRepository]):
        """Get all observations with a specific context."""
        logger.debug(f"Getting observations for context: {context}")
        return await self.repository.find_by_context(context)
+
+    async def get_observations_by_category(self, category: ObservationCategory) -> Sequence[ObservationModel]:
+        """Get all observations with a specific context."""
+        logger.debug(f"Getting observations for context: {category}")
+        return await self.repository.find_by_category(category)
+
+    async def observation_categories(self) -> Sequence[str]:
+        """Get all observation categories."""
+        logger.debug("Getting observations categories")
+        return await self.repository.observation_categories()
@@ -41,7 +41,14 @@ async def create_entity(client) -> EntityResponse:
 async def add_observations(client, path_id: str) -> List[ObservationResponse]:
    response = await client.post(
        "/knowledge/observations",
-        json={"path_id": path_id, "observations": ["First observation", "Second observation"]},
+        json={
+            "path_id": path_id,
+            "observations": [
+                {"content": "First observation", "category": "tech"},
+                {"content": "Second observation", "category": "note"},
+            ],
+            "context": "something special"
+        },
    )
    # Verify observations were added
    assert response.status_code == 200
@@ -407,9 +414,10 @@ async def test_full_knowledge_flow(client: AsyncClient):
        json={
            "path_id": "test/main_entity",
            "observations": [
-                "Connected to first related entity",
-                "Connected to second related entity",
+                {"content": "Connected to first related entity", "category": "tech"},
+                {"content": "Connected to second related entity", "category": "note"},
            ],
+            "context": "testing the flow"
        },
    )

@@ -426,7 +434,9 @@ async def test_full_knowledge_flow(client: AsyncClient):
    # 6. Search should find all related entities
    search = await client.post("/knowledge/search", json={"query": "Related"})
    matches = search.json()["matches"]
-    assert len(matches) == 3  # Should find both related entities, and the main one with the observation
+    assert (
+        len(matches) == 3
+    )  # Should find both related entities, and the main one with the observation

    # 7. Delete main entity
    response = await client.post(
@@ -1,9 +1,14 @@
 """Tests for knowledge entity writer."""
+from datetime import datetime, UTC

 import pytest

 from basic_memory.markdown.knowledge_writer import KnowledgeWriter
-from basic_memory.models import Entity as EntityModel, Observation, Relation
+from basic_memory.models import (
+    Entity as EntityModel,
+    Observation,
+    Relation, ObservationCategory,
+)


@pytest.fixture
@@ -15,17 +20,40 @@ def writer():
@pytest.fixture
 def test_entity():
    """Create test entity with observations and relations."""
+    now = datetime.now(UTC)
    # Create main entity
-    entity = EntityModel(id=123, name="TestEntity", entity_type="test", description="A test entity")
+    entity = EntityModel(
+        id=1,
+        path_id="test/test_entity",
+        name="TestEntity",
+        entity_type="test",
+        description="A test entity",
+        created_at=now,
+        updated_at=now
+    )

-    # Add observations
+    # Add observations with categories and context
    entity.observations = [
-        Observation(content="First observation"),
-        Observation(content="Second observation"),
+        Observation(
+            content="Technical implementation detail",
+            category=ObservationCategory.TECH.value,
+            context="Initial implementation"
+        ),
+        Observation(
+            content="Design pattern choice",
+            category=ObservationCategory.DESIGN.value
+        ),
    ]

    # Create related entity
-    other_entity = EntityModel(id=456, name="OtherEntity", entity_type="test")
+    other_entity = EntityModel(
+        id=2,
+        path_id="test/other_entity",
+        name="OtherEntity",
+        entity_type="test",
+        created_at=now,
+        updated_at=now
+    )

    # Create relation from main entity to other
    relation = Relation(from_entity=entity, to_entity=other_entity, relation_type="relates_to")
@@ -40,28 +68,78 @@ async def test_format_frontmatter(writer: KnowledgeWriter, test_entity: EntityMo
    frontmatter = await writer.format_frontmatter(test_entity)

    assert frontmatter["type"] == "test"
-    assert frontmatter["id"] == 123
+    assert frontmatter["id"] == "test/test_entity"
    assert isinstance(frontmatter["created"], str)
    assert isinstance(frontmatter["modified"], str)


@pytest.mark.asyncio
-async def test_format_content_basic(writer: KnowledgeWriter, test_entity: EntityModel):
-    """Test basic content formatting without metadata."""
+async def test_format_content_with_categories(writer: KnowledgeWriter, test_entity: EntityModel):
+    """Test content formatting with categorized observations."""
    content = await writer.format_content(test_entity)

-    # Check sections
-    assert content.startswith("# TestEntity\n")
-    assert "A test entity" in content
+    # Check observations section header and format comment
    assert "## Observations" in content
-    assert "- First observation" in content
-    assert "- Second observation" in content
-    assert "## Relations" in content
-    assert "- [[OtherEntity]] relates_to" in content
+    assert "<!-- Format: - [category] Content text #tag1 #tag2 (optional context) -->" in content

-    # Should not have metadata section
-    assert "# Metadata" not in content
-    assert "```yml" not in content
+    # Check formatted observations
+    assert "- [tech] Technical implementation detail (Initial implementation)" in content
+    assert "- [design] Design pattern choice" in content
+
+
+@pytest.mark.asyncio
+async def test_format_content_default_category(writer: KnowledgeWriter):
+    """Test formatting observation with default category."""
+    entity = EntityModel(id=1, name="Test", entity_type="test")
+    entity.observations = [
+        Observation(content="Simple note", category=ObservationCategory.NOTE.value) 
+    ]
+
+    content = await writer.format_content(entity)
+    assert "- [note] Simple note" in content
+
+
+@pytest.mark.asyncio
+async def test_format_content_context_handling(writer: KnowledgeWriter):
+    """Test formatting observations with different context scenarios."""
+    entity = EntityModel(id=1, name="Test", entity_type="test")
+    entity.observations = [
+        # With context
+        Observation(
+            content="With context",
+            category=ObservationCategory.TECH.value,
+            context="Important context"
+        ),
+        # Without context
+        Observation(
+            content="No context",
+            category=ObservationCategory.TECH.value
+        ),
+    ]
+
+    content = await writer.format_content(entity)
+    assert "- [tech] With context (Important context)" in content
+    assert "- [tech] No context" in content
+    assert "No context ()" not in content  # Shouldn't have empty parentheses
+
+
+@pytest.mark.asyncio
+async def test_format_content_sections_order(writer: KnowledgeWriter, test_entity: EntityModel):
+    """Test proper order and spacing of sections with new format."""
+    content = await writer.format_content(test_entity)
+    lines = content.split("\n")
+
+    # Find key sections
+    title_idx = next(i for i, line in enumerate(lines) if line.startswith("# "))
+    obs_idx = next(i for i, line in enumerate(lines) if line.strip() == "## Observations")
+    format_idx = next(i for i, line in enumerate(lines) if "<!-- Format:" in line)
+    first_obs_idx = next(i for i, line in enumerate(lines) if line.startswith("- ["))
+
+    # Verify order and spacing
+    assert obs_idx > title_idx  # Observations after title
+    assert format_idx == obs_idx + 1  # Format comment right after header
+    assert lines[format_idx + 1] == ""  # Blank line after format comment
+    assert first_obs_idx == format_idx + 2  # First observation after blank line


@pytest.mark.asyncio
@@ -119,18 +197,3 @@ async def test_format_content_minimal_entity(writer: KnowledgeWriter):
    # Should have title only
    assert content.strip() == "# Minimal"

-
-@pytest.mark.asyncio
-async def test_content_section_spacing(writer: KnowledgeWriter, test_entity: EntityModel):
-    """Test proper spacing between sections."""
-    content = await writer.format_content(test_entity)
-    lines = content.split("\n")
-
-    # Find section headers
-    section_indexes = [i for i, line in enumerate(lines) if line.startswith("##")]
-
-    for idx in section_indexes:
-        # Should be blank line before header
-        assert lines[idx - 1] == "", f"No blank line before section at line {idx}"
-        # Content should start right after header
-        assert lines[idx + 1].strip(), f"No content after section at line {idx}"
@@ -149,3 +149,149 @@ async def test_delete_observation_by_content(session_maker: async_sessionmaker,
    remaining = await repo.find_by_entity(entity.id)
    assert len(remaining) == 1
    assert remaining[0].content == "Keep this observation"
+
+
+@pytest.mark.asyncio
+async def test_find_by_category(session_maker: async_sessionmaker, repo):
+    """Test finding observations by their category."""
+    # Create test entity
+    async with db.scoped_session(session_maker) as session:
+        entity = Entity(
+            name="test_entity",
+            entity_type="test",
+            description="Test entity",
+            path_id="test/test_entity"
+        )
+        session.add(entity)
+        await session.flush()
+
+        # Create test observations with different categories
+        observations = [
+            Observation(
+                entity_id=entity.id,
+                content="Tech observation",
+                category="tech"
+            ),
+            Observation(
+                entity_id=entity.id,
+                content="Design observation",
+                category="design"
+            ),
+            Observation(
+                entity_id=entity.id,
+                content="Another tech observation",
+                category="tech"
+            )
+        ]
+        session.add_all(observations)
+        await session.commit()
+
+    # Find tech observations
+    tech_obs = await repo.find_by_category("tech")
+    assert len(tech_obs) == 2
+    assert all(obs.category == "tech" for obs in tech_obs)
+    assert set(obs.content for obs in tech_obs) == {
+        "Tech observation",
+        "Another tech observation"
+    }
+
+    # Find design observations
+    design_obs = await repo.find_by_category("design")
+    assert len(design_obs) == 1
+    assert design_obs[0].category == "design"
+    assert design_obs[0].content == "Design observation"
+
+    # Search for non-existent category
+    missing_obs = await repo.find_by_category("missing")
+    assert len(missing_obs) == 0
+
+
+@pytest.mark.asyncio
+async def test_observation_categories(session_maker: async_sessionmaker, repo):
+    """Test retrieving distinct observation categories."""
+    # Create test entity
+    async with db.scoped_session(session_maker) as session:
+        entity = Entity(
+            name="test_entity",
+            entity_type="test",
+            description="Test entity",
+            path_id="test/test_entity"
+        )
+        session.add(entity)
+        await session.flush()
+
+        # Create observations with various categories
+        observations = [
+            Observation(
+                entity_id=entity.id,
+                content="First tech note",
+                category="tech"
+            ),
+            Observation(
+                entity_id=entity.id,
+                content="Second tech note",
+                category="tech"  # Duplicate category
+            ),
+            Observation(
+                entity_id=entity.id,
+                content="Design note",
+                category="design"
+            ),
+            Observation(
+                entity_id=entity.id,
+                content="Feature note",
+                category="feature"
+            )
+        ]
+        session.add_all(observations)
+        await session.commit()
+
+    # Get distinct categories
+    categories = await repo.observation_categories()
+
+    # Should have unique categories in a deterministic order
+    assert set(categories) == {"tech", "design", "feature"}
+
+
+@pytest.mark.asyncio
+async def test_find_by_category_with_empty_db(repo):
+    """Test category operations with an empty database."""
+    # Find by category should return empty list
+    obs = await repo.find_by_category("tech")
+    assert len(obs) == 0
+
+    # Get categories should return empty list
+    categories = await repo.observation_categories()
+    assert len(categories) == 0
+
+
+@pytest.mark.asyncio
+async def test_find_by_category_case_sensitivity(session_maker: async_sessionmaker, repo):
+    """Test how category search handles case sensitivity."""
+    async with db.scoped_session(session_maker) as session:
+        entity = Entity(
+            name="test_entity",
+            entity_type="test",
+            description="Test entity",
+            path_id="test/test_entity"
+        )
+        session.add(entity)
+        await session.flush()
+
+        # Create a test observation
+        obs = Observation(
+            entity_id=entity.id,
+            content="Tech note",
+            category="tech"  # lowercase in database
+        )
+        session.add(obs)
+        await session.commit()
+
+    # Search should work regardless of case
+    # Note: If we want case-insensitive search, we'll need to update the query
+    # For now, this test documents the current behavior
+    exact_match = await repo.find_by_category("tech")
+    assert len(exact_match) == 1
+
+    upper_case = await repo.find_by_category("TECH")
+    assert len(upper_case) == 0  # Currently case-sensitive
@@ -8,8 +8,10 @@ from sqlalchemy.exc import IntegrityError

 from basic_memory.models import Entity as EntityModel
 from basic_memory.schemas import Entity as EntitySchema, Relation as RelationSchema
+from basic_memory.schemas.base import ObservationCategory
+from basic_memory.schemas.request import ObservationCreate
 from basic_memory.services import EntityService
-from basic_memory.services.exceptions import EntityNotFoundError, FileOperationError, EntityCreationError
+from basic_memory.services.exceptions import EntityNotFoundError, FileOperationError
 from basic_memory.services.knowledge import KnowledgeService


@@ -91,7 +93,7 @@ async def test_create_relations(knowledge_service: KnowledgeService, entity_serv


@pytest.mark.asyncio
-async def test_add_observations(knowledge_service: KnowledgeService):
+async def test_add_observations_observation(knowledge_service: KnowledgeService):
    """Should add observations and update entity file."""
    # Create test entity
    entity = await knowledge_service.create_entity(
@@ -99,21 +101,34 @@ async def test_add_observations(knowledge_service: KnowledgeService):
    )

    # Add observations
-    observations = ["Test observation 1", "Test observation 2"]
+    observations = [
+        ObservationCreate(content="Test observation 1", category=ObservationCategory.TECH),
+        ObservationCreate(content="Test observation 2", category=ObservationCategory.DESIGN),
+    ]
+    context = "Test context"
    updated_entity = await knowledge_service.add_observations(
-        entity.path_id, observations, "Test context"
+        entity.path_id, observations, context
    )

    # Verify observations in DB
    assert len(updated_entity.observations) == 2
    assert updated_entity.observations[0].content == "Test observation 1"
+    assert updated_entity.observations[0].category == "tech"
+    assert updated_entity.observations[0].context == context
    assert updated_entity.observations[1].content == "Test observation 2"
-
+    assert updated_entity.observations[1].category == "design"
+    assert updated_entity.observations[1].context == context
+    
    # Verify file was updated
    file_path = knowledge_service.get_entity_path(updated_entity)
    content, _ = await knowledge_service.file_service.read_file(file_path)
+    
    for obs in observations:
-        assert obs in content
+        expected_line = f"- [{obs.category.value}] {obs.content} ({context})"
+        assert expected_line in content
+
+    # Also verify the Observations section header exists
+    assert "## Observations" in content


@pytest.mark.asyncio
@@ -6,6 +6,8 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

 from basic_memory.models import Entity, Observation
 from basic_memory.repository.observation_repository import ObservationRepository
+from basic_memory.schemas.base import ObservationCategory
+from basic_memory.schemas.request import ObservationCreate
 from basic_memory.services.observation_service import ObservationService


@@ -157,3 +159,144 @@ async def test_get_observations_by_context(
    assert len(results) == 1
    assert results[0].content == "Contextual observation"
    assert results[0].context == "test_context"
+
+"""Additional tests for ObservationService category support."""
+
+
+
+@pytest.mark.asyncio
+async def test_add_observations_with_categories(
+    observation_service: ObservationService,
+    test_entity: Entity
+):
+    """Test adding observations with specific categories."""
+    observations = [
+        ObservationCreate(content="Tech observation", category=ObservationCategory.TECH),
+        ObservationCreate(content="Design observation", category=ObservationCategory.DESIGN),
+    ]
+
+    result = await observation_service.add_observations(test_entity.id, observations)
+
+    assert len(result) == 2
+    assert result[0].category == ObservationCategory.TECH.value
+    assert result[1].category == ObservationCategory.DESIGN.value
+
+
+@pytest.mark.asyncio
+async def test_search_observations_by_category(
+    observation_service: ObservationService,
+    test_entity: Entity
+):
+    """Test searching observations with category filter."""
+    # Add observations with different categories
+    observations = [
+        ObservationCreate(content="Tech implementation note", category=ObservationCategory.TECH),
+        ObservationCreate(content="Design pattern choice", category=ObservationCategory.DESIGN),
+        ObservationCreate(content="Another tech detail", category=ObservationCategory.TECH),
+    ]
+    await observation_service.add_observations(test_entity.id, observations)
+
+    # Search with category filter
+    tech_results = await observation_service.search_observations(
+        query="note",
+        category=ObservationCategory.TECH
+    )
+    assert len(tech_results) == 1
+    assert tech_results[0].content == "Tech implementation note"
+
+    # Search without category filter
+    all_results = await observation_service.search_observations(
+        query="tech"
+    )
+    assert len(all_results) == 2
+
+
+@pytest.mark.asyncio
+async def test_get_observations_by_category(
+    observation_service: ObservationService,
+    test_entity: Entity
+):
+    """Test retrieving observations by category."""
+    # Add observations with different categories
+    observations = [
+        ObservationCreate(content="First tech note", category=ObservationCategory.TECH),
+        ObservationCreate(content="Design decision", category=ObservationCategory.DESIGN),
+        ObservationCreate(content="Second tech note", category=ObservationCategory.TECH),
+    ]
+    await observation_service.add_observations(test_entity.id, observations)
+
+    # Get tech observations
+    tech_obs = await observation_service.get_observations_by_category(ObservationCategory.TECH)
+    assert len(tech_obs) == 2
+    assert all(obs.category == ObservationCategory.TECH.value for obs in tech_obs)
+
+    # Get observations for unused category
+    feature_obs = await observation_service.get_observations_by_category(ObservationCategory.FEATURE)
+    assert len(feature_obs) == 0
+
+
+@pytest.mark.asyncio
+async def test_observation_categories(
+    observation_service: ObservationService,
+    test_entity: Entity
+):
+    """Test retrieving distinct observation categories."""
+    # Add observations with various categories
+    observations = [
+        ObservationCreate(content="Tech note", category=ObservationCategory.TECH),
+        ObservationCreate(content="Design note", category=ObservationCategory.DESIGN),
+        ObservationCreate(content="Another tech note", category=ObservationCategory.TECH),
+        ObservationCreate(content="Feature note", category=ObservationCategory.FEATURE),
+    ]
+    await observation_service.add_observations(test_entity.id, observations)
+
+    # Get categories
+    categories = await observation_service.observation_categories()
+    assert set(categories) == {
+        ObservationCategory.TECH.value,
+        ObservationCategory.DESIGN.value,
+        ObservationCategory.FEATURE.value
+    }
+
+
+@pytest.mark.asyncio
+async def test_search_observations_empty_category(
+    observation_service: ObservationService,
+    test_entity: Entity
+):
+    """Test search behavior with empty/invalid category."""
+    # Add some observations
+    observations = [
+        ObservationCreate(content="Tech note", category=ObservationCategory.TECH),
+    ]
+    await observation_service.add_observations(test_entity.id, observations)
+
+    # Search with empty category should return all matches
+    results = await observation_service.search_observations(
+        query="note",
+        category=None
+    )
+    assert len(results) == 1
+
+    # Search with non-existent category should return empty
+    results = await observation_service.search_observations(
+        query="note",
+        category=ObservationCategory.ISSUE
+    )
+    assert len(results) == 0
+
+
+@pytest.mark.asyncio
+async def test_default_category_behavior(
+    observation_service: ObservationService,
+    test_entity: Entity
+):
+    """Test default category assignment."""
+    # Add observation without explicit category
+    observations = [
+        ObservationCreate(content="Simple note")  # No category specified
+    ]
+    result = await observation_service.add_observations(test_entity.id, observations)
+
+    assert len(result) == 1
+    assert result[0].category == ObservationCategory.NOTE.value  # Should use default