fix(core): preserve malformed frontmatter during sync

src/basic_memory/services/file_service.py

@@ -465,7 +465,6 @@ async def update_frontmatter_with_result(
 
         Raises:
             FileOperationError: If file operations fail
-            ParseError: If frontmatter parsing fails
         """
         # Convert string to Path if needed
         path_obj = self.base_path / path if isinstance(path, str) else path
@@ -481,15 +480,15 @@ async def update_frontmatter_with_result(
             if file_utils.has_frontmatter(content):
                 try:
                     current_fm = file_utils.parse_frontmatter(content)
-                    content = file_utils.remove_frontmatter(content)
-                except (ParseError, yaml.YAMLError) as e:  # pragma: no cover
-                    # Log warning and treat as plain markdown without frontmatter
-                    logger.warning(  # pragma: no cover
-                        f"Failed to parse YAML frontmatter in {full_path}: {e}. "
-                        "Treating file as plain markdown without frontmatter."
-                    )
-                    # Keep full content, treat as having no frontmatter
-                    current_fm = {}  # pragma: no cover
+                except ParseError as e:
+                    # Trigger: a fenced frontmatter block cannot be parsed safely.
+                    # Why: Markdown is authoritative, and a partial update cannot know
+                    # which malformed metadata fields the user intended to preserve.
+                    # Outcome: reject the rewrite before any bytes are changed.
+                    raise FileOperationError(
+                        f"Refusing to update malformed frontmatter in {full_path}: {e}"
+                    ) from e
+                content = file_utils.remove_frontmatter(content)
 
             # Update frontmatter
             new_fm = {**current_fm, **updates}
@@ -522,14 +521,14 @@ async def update_frontmatter_with_result(
                 content=content_for_checksum,
             )
 
+        except FileOperationError:
+            raise
         except Exception as e:  # pragma: no cover
-            # Only log real errors (not YAML parsing, which is handled above)
-            if not isinstance(e, (ParseError, yaml.YAMLError)):
-                logger.error(
-                    "Failed to update frontmatter",
-                    path=str(full_path),
-                    error=str(e),
-                )
+            logger.error(
+                "Failed to update frontmatter",
+                path=str(full_path),
+                error=str(e),
+            )
             raise FileOperationError(f"Failed to update frontmatter: {e}")
 
     async def update_frontmatter(self, path: FilePath, updates: Dict[str, Any]) -> str:

tests/indexing/test_batch_indexer.py

@@ -655,6 +655,58 @@ async def test_batch_indexer_uses_parsed_markdown_body_for_malformed_frontmatter
     assert entity is not None
 
 
+@pytest.mark.asyncio
+async def test_batch_indexer_reports_malformed_yaml_without_rewriting_source(
+    app_config,
+    entity_service,
+    entity_repository,
+    relation_repository,
+    search_service,
+    file_service,
+    project_config,
+):
+    path = "notes/malformed-yaml.md"
+    original_content = dedent(
+        """\
+        ---
+        title: Important
+        description: Agent context: urgent: keep
+        tags: [critical]
+        ---
+
+        # Important
+
+        The source file must remain authoritative.
+        """
+    )
+    await _create_file(project_config.home / path, original_content)
+    batch_indexer = _make_batch_indexer(
+        app_config,
+        entity_service,
+        entity_repository,
+        relation_repository,
+        search_service,
+        file_service,
+    )
+
+    result = await batch_indexer.index_files(
+        {path: await _load_input(file_service, path)},
+        max_concurrent=1,
+        parse_max_concurrent=1,
+    )
+
+    assert result.indexed == []
+    assert len(result.errors) == 1
+    error_path, error = result.errors[0]
+    assert error_path == path
+    assert "Refusing to update malformed frontmatter" in error
+    assert (project_config.home / path).read_text(encoding="utf-8") == original_content
+
+    async with db.scoped_session(search_service.session_maker) as session:
+        entity = await entity_repository.get_by_file_path(session, path)
+    assert entity is None
+
+
 @pytest.mark.asyncio
 async def test_batch_indexer_re_raises_fatal_sync_errors(
     app_config,

tests/services/test_file_service.py

@@ -200,6 +200,32 @@ async def fake_write_file_atomic(path: Path, content: str) -> None:
         setattr(result, "checksum", "changed")
 
 
+@pytest.mark.asyncio
+async def test_update_frontmatter_rejects_malformed_yaml_without_changing_file(
+    tmp_path: Path, file_service: FileService
+):
+    """A partial metadata update must not replace malformed user frontmatter."""
+    test_path = tmp_path / "note.md"
+    original_content = (
+        "---\n"
+        "title: Important\n"
+        "description: Agent context: urgent: keep\n"
+        "tags: [critical]\n"
+        "---\n\n"
+        "# Note\n"
+        "Body\n"
+    )
+    test_path.write_text(original_content, encoding="utf-8")
+
+    with pytest.raises(FileOperationError, match="Refusing to update malformed frontmatter"):
+        await file_service.update_frontmatter_with_result(
+            test_path,
+            {"permalink": "notes/important"},
+        )
+
+    assert test_path.read_text(encoding="utf-8") == original_content
+
+
 @pytest.mark.asyncio
 async def test_read_file_content(tmp_path: Path, file_service: FileService):
     """Test read_file_content returns just the content without checksum."""