fix(knowledge): 区分普通分割线与元数据头部

backend/app/knowledge/parser.py

@@ -190,11 +190,44 @@ def _frontmatter(markdown: str) -> tuple[str, int] | None:
     offset = content_start
     for raw in markdown[content_start:].splitlines(keepends=True):
         if re.fullmatch(r"(?:---|\.\.\.)[ \t]*", raw.rstrip("\r\n")):
-            return markdown[content_start:offset], offset + len(raw)
+            candidate = markdown[content_start:offset]
+            if not candidate.strip() or _metadata_intent(candidate):
+                return candidate, offset + len(raw)
+            return None  # Ordinary Markdown between thematic breaks.
         offset += len(raw)
+    if not _metadata_intent(markdown[content_start:]):
+        return None
     raise ApiError(422, "INVALID_EMBEDDING_POLICY", "Frontmatter 未闭合，请补全独立一行的结束分隔符后再保存。")
 
 
+def _metadata_intent(content: str) -> bool:
+    """A thematic break alone is not a declaration of YAML metadata."""
+    # An explicit policy must fail closed even when other header lines are broken.
+    fence_marker = None
+    for line in content.splitlines():
+        fence = _FENCE_RE.match(line)
+        if fence_marker is not None:
+            marker = fence.group(1) if fence else ""
+            if marker.startswith(fence_marker[0]) and len(marker) >= len(fence_marker):
+                fence_marker = None
+            continue
+        if fence:
+            fence_marker = fence.group(1)
+            continue
+        if re.match(r"(?i)^[ \t]*[\"']?embedding_local_only[\"']?[ \t]*:", line):
+            return True
+    try:
+        if isinstance(yaml.compose(content, Loader=yaml.SafeLoader), yaml.MappingNode):
+            return True
+    except yaml.YAMLError:
+        pass
+    first = next((line.strip() for line in content.splitlines()
+                  if line.strip() and not line.lstrip().startswith("#")), "")
+    # Preserve errors for incomplete key/value headers, including flow mappings.
+    return bool(re.match(r"(?:[\w.-]+|[\"'][^\"']+[\"'])\s*:(?:\s|$)", first)
+                or (first.startswith("{") and ":" in first))
+
+
 def _utf16_len(text: str) -> int:
     return len(text.encode("utf-16-le")) // 2
 

backend/tests/test_policy_and_migrations.py

@@ -165,3 +165,41 @@ def test_bom_save_and_invalid_update_never_use_remote(monkeypatch):
         assert (get_settings().vault_path/note.file_path).read_text(encoding='utf-8')==markdown
         assert (await note_service.get_note(note.note_id)).markdown==markdown
     asyncio.run(scenario())
+
+
+@pytest.mark.parametrize('markdown', ['---', '---\n\n# Title\n\nNormal body', '---\n\nNormal body\n\n---\n\nLast paragraph', '---\n\n```python\nprint(1)\n```\n---'])
+def test_thematic_breaks_are_not_frontmatter(markdown):
+    note = parse_note(markdown=markdown,file_path='ordinary.md',folder='',created_at=datetime.now(timezone.utc),updated_at=datetime.now(timezone.utc))
+    assert not note.embedding_local_only
+    assert note.blocks[0].content == '---'
+    assert any(block.content == markdown.split('\n\n')[-1] for block in note.blocks) or '```' in markdown
+
+
+@pytest.mark.parametrize('header', ['title: Sample\nembedding_local_only: true', '"embedding_local_only": true', 'title: [broken\nembedding_local_only: true', '{embedding_local_only: true'])
+def test_unclosed_metadata_still_fails_closed(header):
+    with pytest.raises(ApiError) as error:
+        parse_note(markdown='---\n'+header,file_path='private.md',folder='',created_at=datetime.now(timezone.utc),updated_at=datetime.now(timezone.utc))
+    assert error.value.code == 'INVALID_EMBEDDING_POLICY'
+
+
+def test_thematic_break_note_can_save_and_rebuild():
+    import asyncio
+    from app.services import note_service, index_service
+    from app.contracts import IndexRebuildRequest
+    async def scenario():
+        markdown='---\n\n# Title\n\nNormal body'
+        note=await note_service.create_note(title='Divider',markdown=markdown,folder=None,tags=[])
+        assert note.blocks[0].content == '---'
+        assert (await index_service.rebuild(IndexRebuildRequest())).status == 'completed'
+        loaded=await note_service.get_note(note.note_id)
+        assert loaded.markdown == markdown
+        assert [b.content for b in loaded.blocks] == [b.content for b in note.blocks]
+    asyncio.run(scenario())
+
+
+def test_thematic_break_with_policy_example_is_ordinary_markdown():
+    markdown='---\n\n```yaml\nembedding_local_only: true\n```\n\n---\n\nExplanation'
+    note=parse_note(markdown=markdown,file_path='example.md',folder='',created_at=datetime.now(timezone.utc),updated_at=datetime.now(timezone.utc))
+    assert not note.embedding_local_only
+    assert any('embedding_local_only: true' in block.content for block in note.blocks)
+    assert note.blocks[0].content=='---'