diff --git a/src/basic_memory/markdown/markdown_processor.py b/src/basic_memory/markdown/markdown_processor.py
new file mode 100644
index 00000000..64d94b44
--- /dev/null
+++ b/src/basic_memory/markdown/markdown_processor.py
@@ -0,0 +1,156 @@
+"""Process markdown files with structured sections.
+
+This module follows a Read -> Modify -> Write pattern for all file operations:
+1. Read entire file and parse into EntityMarkdown schema
+2. Modify the schema (add relation, update content, etc)
+3. Write entire file atomically using temp file + swap
+
+No in-place updates are performed. Each write reconstructs the entire file from the schema.
+The file format has two distinct types of content:
+1. User content - Free form text that is preserved exactly as written
+2. Structured sections - Observations and Relations that are always formatted 
+   in a standard way and can be overwritten since they're tracked in our schema
+"""
+
+from pathlib import Path
+from typing import Optional
+
+import frontmatter
+from frontmatter import Post
+from loguru import logger
+
+from basic_memory import file_utils
+from basic_memory.markdown.entity_parser import EntityParser
+from basic_memory.markdown.schemas import EntityMarkdown, Observation, Relation
+
+
+class DirtyFileError(Exception):
+    """Raised when attempting to write to a file that has been modified."""
+    pass
+
+
+class MarkdownProcessor:
+    """Process markdown files while preserving content and structure.
+    
+    This class handles the file I/O aspects of our markdown processing. It:
+    1. Uses EntityParser for reading/parsing files into our schema
+    2. Handles writing files with proper frontmatter
+    3. Formats structured sections (observations/relations) consistently
+    4. Preserves user content exactly as written
+    5. Performs atomic writes using temp files
+    
+    It does NOT:
+    1. Modify the schema directly (that's done by services)
+    2. Handle in-place updates (everything is read->modify->write)
+    3. Track schema changes (that's done by the database)
+    """
+
+    def __init__(self, base_path: Path, entity_parser: EntityParser):
+        """Initialize processor with base path and parser."""
+        self.base_path = base_path.resolve()
+        self.entity_parser = entity_parser
+
+    async def read_file(self, path: Path) -> EntityMarkdown:
+        """Read and parse file into EntityMarkdown schema.
+        
+        This is step 1 of our read->modify->write pattern.
+        We use EntityParser to handle all the markdown parsing.
+        """
+        return await self.entity_parser.parse_file(path)
+
+    async def write_file(
+        self,
+        path: Path,
+        markdown: EntityMarkdown,
+        expected_checksum: Optional[str] = None,
+    ) -> str:
+        """Write EntityMarkdown schema back to file.
+        
+        This is step 3 of our read->modify->write pattern.
+        The entire file is rewritten atomically on each update.
+
+        File Structure:
+        ---
+        frontmatter fields
+        ---
+        user content area (preserved exactly)
+        
+        ## Observations (if any)
+        formatted observations
+        
+        ## Relations (if any)
+        formatted relations
+
+        Args:
+            path: Where to write the file
+            markdown: Complete schema to write
+            expected_checksum: If provided, verify file hasn't changed
+
+        Returns:
+            Checksum of written file
+
+        Raises:
+            DirtyFileError: If file has been modified (when expected_checksum provided)
+        """
+        # Dirty check if needed
+        if expected_checksum is not None:
+            current_content = path.read_text()
+            current_checksum = await file_utils.compute_checksum(current_content)
+            if current_checksum != expected_checksum:
+                raise DirtyFileError(f"File {path} has been modified")
+        
+        # Convert frontmatter to dict, dropping None values
+        frontmatter_dict = {
+            "type": markdown.frontmatter.type,
+            "permalink": markdown.frontmatter.permalink,
+            "created": markdown.frontmatter.created.isoformat() if markdown.frontmatter.created else None,
+            "modified": markdown.frontmatter.modified.isoformat() if markdown.frontmatter.modified else None,
+            "tags": markdown.frontmatter.tags,
+        }
+        frontmatter_dict = {k: v for k, v in frontmatter_dict.items() if v is not None}
+        
+        # Start with user content (or minimal title for new files)
+        content = markdown.content.content or f"# {markdown.frontmatter.title}\n"
+        
+        # Add structured sections if present
+        if markdown.content.observations:
+            content += "\n## Observations\n" + self.format_observations(markdown.content.observations)
+        if markdown.content.relations:
+            content += "\n## Relations\n" + self.format_relations(markdown.content.relations)
+        
+        # Create Post object for frontmatter
+        post = Post(content, **frontmatter_dict)
+        final_content = frontmatter.dumps(post)
+        
+        # Write atomically and return checksum
+        path.parent.mkdir(parents=True, exist_ok=True)
+        await file_utils.write_file_atomic(path, final_content)
+        return await file_utils.compute_checksum(final_content)
+
+    def format_observations(self, observations: list[Observation]) -> str:
+        """Format observations section in standard way.
+        
+        Format: - [category] content #tag1 #tag2 (context)
+        """
+        lines = []
+        for obs in observations:
+            line = f"- [{obs.category}] {obs.content}"
+            if obs.tags:
+                line += " " + " ".join(f"#{tag}" for tag in sorted(obs.tags))
+            if obs.context:
+                line += f" ({obs.context})"
+            lines.append(line)
+        return "\n".join(lines) + "\n"
+
+    def format_relations(self, relations: list[Relation]) -> str:
+        """Format relations section in standard way.
+        
+        Format: - relation_type [[target]] (context)
+        """
+        lines = []
+        for rel in relations:
+            line = f"- {rel.type} [[{rel.target}]]"
+            if rel.context:
+                line += f" ({rel.context})"
+            lines.append(line)
+        return "\n".join(lines) + "\n"
\ No newline at end of file
diff --git a/src/basic_memory/markdown/schemas.py b/src/basic_memory/markdown/schemas.py
index 11d3829e..85e5a843 100644
--- a/src/basic_memory/markdown/schemas.py
+++ b/src/basic_memory/markdown/schemas.py
@@ -31,7 +31,7 @@ class EntityFrontmatter(BaseModel):
     permalink: Optional[str] = None
     created: datetime
     modified: datetime
-    tags: List[str]
+    tags: Optional[List[str]] = None
 
 
 class EntityContent(BaseModel):
diff --git a/tests/markdown/test_markdown_processor.py b/tests/markdown/test_markdown_processor.py
new file mode 100644
index 00000000..dd98496e
--- /dev/null
+++ b/tests/markdown/test_markdown_processor.py
@@ -0,0 +1,201 @@
+"""Tests for MarkdownProcessor.
+
+Tests focus on the Read -> Modify -> Write pattern and content preservation.
+"""
+
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+
+from basic_memory.markdown.entity_parser import EntityParser
+from basic_memory.markdown.markdown_processor import MarkdownProcessor, DirtyFileError
+from basic_memory.markdown.schemas import (
+    EntityMarkdown,
+    EntityFrontmatter,
+    EntityContent,
+    Observation,
+    Relation,
+)
+
+
+@pytest.fixture
+def processor(tmp_path: Path, entity_parser: EntityParser) -> MarkdownProcessor:
+    """Create MarkdownProcessor with temp path."""
+    return MarkdownProcessor(tmp_path, entity_parser)
+
+
+@pytest.mark.asyncio
+async def test_write_new_minimal_file(processor: MarkdownProcessor, tmp_path: Path):
+    """Test creating new file with just title."""
+    path = tmp_path / "test.md"
+    
+    # Create minimal markdown schema
+    markdown = EntityMarkdown(
+        frontmatter=EntityFrontmatter(
+            type="note",
+            permalink="test",
+            title="Test Note",
+            created=datetime(2024, 1, 1),
+            modified=datetime(2024, 1, 1),
+            tags=["test"],
+        ),
+        content=EntityContent(content=""),
+    )
+    
+    # Write file
+    checksum = await processor.write_file(path, markdown)
+    
+    # Read back and verify
+    content = path.read_text()
+    assert "---" in content  # Has frontmatter
+    assert "type: note" in content
+    assert "permalink: test" in content
+    assert "# Test Note" in content  # Added title
+    assert "tags:" in content
+    assert "- test" in content
+
+    # Should not have empty sections
+    assert "## Observations" not in content
+    assert "## Relations" not in content
+
+
+@pytest.mark.asyncio
+async def test_write_new_file_with_content(processor: MarkdownProcessor, tmp_path: Path):
+    """Test creating new file with content and sections."""
+    path = tmp_path / "test.md"
+    
+    # Create markdown with content and sections
+    markdown = EntityMarkdown(
+        frontmatter=EntityFrontmatter(
+            type="note",
+            permalink="test",
+            title="Test Note",
+            created=datetime(2024, 1, 1),
+            modified=datetime(2024, 1, 1),
+        ),
+        content=EntityContent(
+            content="# Custom Title\n\nMy content here.\nMultiple lines.",
+            observations=[
+                Observation(
+                    content="Test observation",
+                    category="tech",
+                    tags=["test"],
+                    context="test context",
+                ),
+            ],
+            relations=[
+                Relation(
+                    type="relates_to",
+                    target="other-note",
+                    context="test relation",
+                ),
+            ],
+        ),
+    )
+    
+    # Write file
+    checksum = await processor.write_file(path, markdown)
+    
+    # Read back and verify
+    content = path.read_text()
+    
+    # Check content preserved exactly
+    assert "# Custom Title" in content
+    assert "My content here." in content
+    assert "Multiple lines." in content
+    
+    # Check sections formatted correctly
+    assert "## Observations" in content
+    assert "- [tech] Test observation #test (test context)" in content
+    
+    assert "## Relations" in content
+    assert "- relates_to [[other-note]] (test relation)" in content
+
+
+@pytest.mark.asyncio
+async def test_update_preserves_content(processor: MarkdownProcessor, tmp_path: Path):
+    """Test that updating file preserves existing content."""
+    path = tmp_path / "test.md"
+    
+    # Create initial file
+    initial = EntityMarkdown(
+        frontmatter=EntityFrontmatter(
+            type="note",
+            permalink="test",
+            title="Test Note",
+            created=datetime(2024, 1, 1),
+            modified=datetime(2024, 1, 1),
+        ),
+        content=EntityContent(
+            content="# My Note\n\nOriginal content here.",
+            observations=[
+                Observation(content="First observation", category="note"),
+            ],
+        ),
+    )
+    
+    checksum = await processor.write_file(path, initial)
+    
+    # Update with new observation
+    updated = EntityMarkdown(
+        frontmatter=initial.frontmatter,
+        content=EntityContent(
+            content=initial.content.content,  # Preserve original content
+            observations=[
+                initial.content.observations[0],  # Keep original observation
+                Observation(content="Second observation", category="tech"),  # Add new one
+            ],
+        ),
+    )
+    
+    # Update file
+    new_checksum = await processor.write_file(path, updated, expected_checksum=checksum)
+    
+    # Read back and verify
+    result = await processor.read_file(path)
+    
+    # Original content preserved
+    assert "Original content here." in result.content.content
+    
+    # Both observations present
+    assert len(result.content.observations) == 2
+    assert any(o.content == "First observation" for o in result.content.observations)
+    assert any(o.content == "Second observation" for o in result.content.observations)
+
+
+@pytest.mark.asyncio
+async def test_dirty_file_detection(processor: MarkdownProcessor, tmp_path: Path):
+    """Test detection of file modifications."""
+    path = tmp_path / "test.md"
+    
+    # Create initial file
+    initial = EntityMarkdown(
+        frontmatter=EntityFrontmatter(
+            type="note",
+            permalink="test",
+            title="Test Note",
+            created=datetime(2024, 1, 1),
+            modified=datetime(2024, 1, 1),
+        ),
+        content=EntityContent(content="Initial content"),
+    )
+    
+    checksum = await processor.write_file(path, initial)
+    
+    # Modify file directly
+    path.write_text(path.read_text() + "\nModified!")
+    
+    # Try to update with old checksum
+    update = EntityMarkdown(
+        frontmatter=initial.frontmatter,
+        content=EntityContent(content="New content"),
+    )
+    
+    # Should raise DirtyFileError
+    with pytest.raises(DirtyFileError):
+        await processor.write_file(path, update, expected_checksum=checksum)
+    
+    # Should succeed without checksum
+    new_checksum = await processor.write_file(path, update)
+    assert new_checksum != checksum
\ No newline at end of file