fix(mcp): recover edit_note when file exists on disk but is not indexed (#934)

Closes #581

Signed-off-by: phernandez <paul@basicmemory.com>
This commit is contained in:
Paul Hernandez
2026-06-10 14:04:59 -05:00
committed by GitHub
parent df485aa5a4
commit db578ccfdb
10 changed files with 1200 additions and 9 deletions
@@ -10,10 +10,18 @@ Key improvements:
- Simplified caching strategies
"""
import os
import pathlib
from fastapi import APIRouter, HTTPException, Response, Path
from loguru import logger
import logfire
from basic_memory.ignore_utils import (
IGNORED_PATH_REJECTION_DETAIL,
load_gitignore_patterns,
should_ignore_path,
)
from basic_memory.deps import (
EntityServiceV2ExternalDep,
SearchServiceV2ExternalDep,
@@ -24,6 +32,7 @@ from basic_memory.deps import (
EntityRepositoryV2ExternalDep,
RelationRepositoryV2ExternalDep,
ProjectExternalIdPathDep,
SyncServiceV2ExternalDep,
TaskSchedulerDep,
)
from basic_memory.schemas import DeleteEntitiesResponse
@@ -40,8 +49,10 @@ from basic_memory.schemas.v2 import (
MoveDirectoryRequestV2,
DeleteDirectoryRequestV2,
OrphanEntitiesResponse,
SyncFileRequest,
)
from basic_memory.schemas.response import DirectoryMoveResult, DirectoryDeleteResult
from basic_memory.utils import validate_project_path
router = APIRouter(prefix="/knowledge", tags=["knowledge-v2"])
@@ -236,6 +247,187 @@ async def resolve_identifier(
return result
## Single-file sync endpoint
def _canonical_file_path(home: pathlib.Path, segments: list[str]) -> str | None:
"""Resolve the actual on-disk casing of a file path under the project home.
Trigger: case-insensitive filesystems (macOS/Windows) pass existence checks for
wrong-cased paths like 'notes/Disk-Note.md' when the file is 'notes/disk-note.md'.
Why: indexing the caller-supplied casing misses the existing DB row keyed by the
on-disk path and inserts a duplicate entity under the wrong-cased path.
Outcome: each segment is matched against real directory entries — exact name first
(so distinct case-variant files on case-sensitive filesystems stay distinct),
then a unique case-insensitive match. Returns None when any segment cannot be
matched to exactly one entry, including missing files. Traversal stops at the
project boundary: a directory whose resolved path escapes the project home is
never scanned.
"""
resolved_home = home.resolve()
current = home
canonical_segments: list[str] = []
for segment in segments:
# Trigger: a previously matched segment may be a symlink whose target lies
# outside the project root (e.g. wrong-cased 'LINK' matched the on-disk
# 'link' -> /tmp/outside on a case-sensitive filesystem).
# Why: os.scandir follows symlinked directories, so continuing would read
# directory contents outside the project boundary even though the
# post-canonicalization containment check rejects the request later.
# Outcome: bail before scanning the moment resolution escapes the home.
if not current.resolve().is_relative_to(resolved_home):
return None
try:
with os.scandir(current) as entries_iter:
entries = [entry.name for entry in entries_iter]
except OSError:
# A parent segment resolved to a non-directory (or vanished): no canonical
# path exists for the remaining segments.
return None
if segment in entries:
matched = segment
else:
matches = [entry for entry in entries if entry.lower() == segment.lower()]
if len(matches) != 1:
return None
matched = matches[0]
canonical_segments.append(matched)
current = current / matched
return "/".join(canonical_segments)
@router.post("/sync-file", response_model=EntityResponseV2)
async def sync_file(
data: SyncFileRequest,
project_id: ProjectExternalIdPathDep,
sync_service: SyncServiceV2ExternalDep,
project_config: ProjectConfigV2ExternalDep,
search_service: SearchServiceV2ExternalDep,
app_config: AppConfigDep,
) -> EntityResponseV2:
"""Index a single markdown file that exists on disk but is not indexed yet.
Recovery path for files written directly to disk before the watcher indexed
them (#581): callers such as edit_note can index the exact file and retry
identifier resolution without running a full project sync.
Args:
data: Request containing the markdown file path relative to project root
Returns:
The indexed entity
Raises:
HTTPException: 400 if the path escapes the project root, contains
non-normalized segments, matches the project ignore rules, or is
not markdown, 404 if the file does not exist on disk
"""
with logfire.span(
"api.request.knowledge.sync_file",
entrypoint="api",
domain="knowledge",
action="sync_file",
):
logger.info(f"API v2 request: sync_file file_path='{data.file_path}'")
if not validate_project_path(data.file_path, project_config.home):
raise HTTPException(
status_code=400,
detail=f"File path '{data.file_path}' is not allowed - "
"paths must stay within project boundaries",
)
# Trigger: segments like './' or '//' survive the traversal check above
# Why: a non-normalized path would index under a non-canonical DB key
# Outcome: reject fail-fast instead of guessing the canonical form
segments = data.file_path.replace("\\", "/").split("/")
if any(segment in ("", ".") for segment in segments):
raise HTTPException(
status_code=400,
detail=f"File path '{data.file_path}' is not normalized - "
"segments like './' or '//' are not allowed",
)
# Canonicalize to the actual on-disk casing so the DB lookup below hits the
# row keyed by the real path instead of inserting a wrong-cased duplicate.
file_path = _canonical_file_path(project_config.home, segments)
if file_path is None:
raise HTTPException(
status_code=404, detail=f"File not found on disk: '{data.file_path}'"
)
# Trigger: canonicalization rewrote a segment to its on-disk form, and that
# segment may be a symlink. The pre-check above validated the ORIGINAL
# request path — on a case-sensitive filesystem 'LINK/secret.md' does not
# exist, so resolve() cannot follow the real 'link' symlink and the check
# passes even when 'link' points outside the project root.
# Why: indexing through an escaping symlink would read and index content
# outside the project boundary — and even an is_file() existence probe on
# the joined path would follow the symlink and stat its target, so
# containment must hold BEFORE any filesystem probe that follows symlinks.
# Path.resolve() only walks symlink names (readlink); it never opens or
# stats the final target, so it is safe to run pre-containment.
# Outcome: the canonical path is re-validated and the fully-resolved absolute
# target must stay inside the resolved project home; escapes get a 400
# before the file-existence probe below ever touches the target.
resolved_target = (project_config.home / file_path).resolve()
if not validate_project_path(file_path, project_config.home) or not (
resolved_target.is_relative_to(project_config.home.resolve())
):
raise HTTPException(
status_code=400,
detail=f"File path '{data.file_path}' is not allowed - "
"paths must stay within project boundaries",
)
# Containment holds, so probing the resolved target cannot leave the project.
if not resolved_target.is_file():
raise HTTPException(
status_code=404, detail=f"File not found on disk: '{data.file_path}'"
)
# Trigger: the canonical path matches the .bmignore / project .gitignore rules
# Why: scan and watch flows filter ignored files before they ever reach the
# indexer; indexing one here would bypass the ignored-file contract and
# make hidden or gitignored content searchable
# Outcome: the same should_ignore_path() rules apply to single-file sync
ignore_patterns = load_gitignore_patterns(project_config.home)
if should_ignore_path(
project_config.home / file_path, project_config.home, ignore_patterns
):
raise HTTPException(
status_code=400,
detail=f"File path '{data.file_path}' {IGNORED_PATH_REJECTION_DETAIL} "
"and cannot be indexed",
)
if not sync_service.file_service.is_markdown(file_path):
raise HTTPException(
status_code=400,
detail=f"Only markdown files can be indexed: '{data.file_path}'",
)
# Trigger: the file may already have a DB row (e.g. modified on disk after indexing)
# Why: the indexer needs to know whether to insert or update the entity
# Outcome: new is computed from the database instead of assumed by the caller
existing = await sync_service.entity_repository.get_by_file_path(file_path)
synced = await sync_service.sync_one_markdown_file(
file_path, new=existing is None, index_search=True
)
# Trigger: semantic search is enabled and the entity index was just refreshed
# Why: the project sync flow awaits sync_entity_vectors_batch() inline after
# indexing changed files (SyncService.sync); without the single-entity
# equivalent, a note recovered via sync-file stays missing or stale in
# semantic search until a later edit or full project sync
# Outcome: vectors refresh synchronously before the response returns,
# mirroring the sync flow instead of the out-of-band scheduler
if app_config.semantic_search_enabled:
await search_service.sync_entity_vectors_batch([synced.entity.id])
result = EntityResponseV2.model_validate(synced.entity)
logger.info(
f"API v2 response: sync_file file_path='{file_path}' external_id={result.external_id}"
)
return result
## Read endpoints
+9
View File
@@ -7,6 +7,15 @@ from typing import Set
from basic_memory.config import resolve_data_dir
# Marker shared by the API ignored-path rejection detail and MCP-side error handling.
# The sync-file endpoint embeds it in its 400 detail and edit_note's disk recovery
# matches on it, so "exists but ignored" stays distinguishable from generic rejections
# without duplicating message text across layers.
IGNORED_PATH_REJECTION_DETAIL = (
"matches Basic Memory ignore rules (.bmignore or project .gitignore)"
)
# Common directories and patterns to ignore by default
# These are used as fallback if .bmignore doesn't exist
DEFAULT_IGNORE_PATTERNS = {
+29
View File
@@ -276,6 +276,35 @@ class KnowledgeClient:
)
return DirectoryDeleteResult.model_validate(response.json())
# --- Single-file sync ---
async def sync_file(self, file_path: str) -> EntityResponse:
"""Index a markdown file that exists on disk but is not indexed yet.
Args:
file_path: Markdown file path relative to the project root
Returns:
EntityResponse for the indexed entity
Raises:
ToolError: If the file does not exist on disk or indexing fails
"""
with logfire.span(
"mcp.client.knowledge.sync_file",
client_name="knowledge",
operation="sync_file",
):
response = await call_post(
self.http_client,
f"{self._base_path}/sync-file",
json={"file_path": file_path},
client_name="knowledge",
operation="sync_file",
path_template="/v2/projects/{project_id}/knowledge/sync-file",
)
return EntityResponse.model_validate(response.json())
# --- Orphan detection ---
async def get_orphans(self) -> list[GraphNode]:
+104 -8
View File
@@ -1,13 +1,19 @@
"""Edit note tool for Basic Memory MCP server."""
from typing import Annotated, Optional, Literal
from typing import TYPE_CHECKING, Annotated, Optional, Literal
import logfire
from httpx import HTTPStatusError
from loguru import logger
from fastmcp import Context
from mcp.server.fastmcp.exceptions import ToolError
from pydantic import AliasChoices, Field
if TYPE_CHECKING: # pragma: no cover
from basic_memory.mcp.clients import KnowledgeClient
from basic_memory.config import ConfigManager
from basic_memory.ignore_utils import IGNORED_PATH_REJECTION_DETAIL
from basic_memory.mcp.project_context import (
_workspace_identifier_discovery_available,
detect_project_from_memory_url_prefix,
@@ -16,6 +22,7 @@ from basic_memory.mcp.project_context import (
resolve_project_and_path,
)
from basic_memory.mcp.server import mcp
from basic_memory.mcp.tools.utils import _extract_response_data, _response_detail_text
from basic_memory.schemas.base import Entity
from basic_memory.schemas.response import EntityResponse
from basic_memory.services.link_resolver import (
@@ -52,6 +59,79 @@ def _parse_identifier_to_title_and_directory(identifier: str) -> tuple[str, str]
return title, directory
# Suffixes mimetypes maps to text/markdown (extension matching is case-insensitive),
# mirroring FileService.is_markdown which gates the sync-file endpoint server-side.
_MARKDOWN_SUFFIXES = (".md", ".markdown")
async def _resolve_after_disk_recovery(
knowledge_client: "KnowledgeClient",
identifier: str,
) -> Optional[str]:
"""Recover from a resolution miss when the note exists on disk but is not indexed.
Trigger: identifier resolution failed with "not found", but the identifier may map
to a markdown file written directly to disk before the watcher indexed it (#581).
Why: editing an on-disk note should not require a manual full sync or watcher restart.
Outcome: the single file is indexed server-side and resolution is retried exactly
once. Returns None when the identifier does not map to an indexable file on
disk, so the caller keeps its existing not-found handling.
"""
# Try the identifier as-is first so existing .markdown/.MD files are found; only
# fall back to appending markdown suffixes (".md" first, then ".markdown") when
# the identifier does not already carry one, so 'notes/foo.markdown' never becomes
# 'notes/foo.markdown.md' and a stem identifier still reaches 'notes/foo.markdown'.
candidates = [identifier]
if not identifier.lower().endswith(_MARKDOWN_SUFFIXES):
candidates.extend(f"{identifier}{suffix}" for suffix in _MARKDOWN_SUFFIXES)
for candidate in candidates:
try:
synced = await knowledge_client.sync_file(candidate)
except ToolError as sync_error:
# Trigger: the sync-file request failed
# Why: 400/404 are the expected "nothing to recover" rejections (missing
# file, traversal, non-markdown) — except the ignored-path 400, which
# means the file exists on disk but the ignore rules forbid indexing
# it, so falling through to auto-create would silently shadow the
# file. Anything else — auth, server, transport-level failures — is a
# real error that must not be masked as a not-found miss.
# Outcome: ignored-path rejections raise a clear ToolError; other expected
# rejections try the next candidate or fall through to the caller's
# existing not-found behavior; unexpected failures propagate.
cause = sync_error.__cause__
candidate_rejected = isinstance(
cause, HTTPStatusError
) and cause.response.status_code in (400, 404)
if not candidate_rejected:
raise
detail = _response_detail_text(_extract_response_data(cause.response)) or ""
if IGNORED_PATH_REJECTION_DETAIL in detail:
raise ToolError(
f"Note file '{candidate}' exists on disk but {IGNORED_PATH_REJECTION_DETAIL} "
"and will not be edited"
) from sync_error
logger.debug(f"edit_note disk recovery skipped for '{candidate}': {sync_error}")
continue
# Trigger: sync-file succeeded and returned the indexed entity.
# Why: the server may have canonicalized the path casing (notes/Disk-Note ->
# notes/disk-note.md), so strictly re-resolving the raw identifier can
# still miss the entity we just indexed.
# Outcome: use the entity identity from the sync-file response directly; only
# fall back to a strict re-resolve when an older server omits external_id,
# and let that re-resolve fail loudly instead of guessing.
if synced.external_id:
logger.info(
f"edit_note indexed unindexed file '{candidate}' as entity {synced.external_id}"
)
return synced.external_id
logger.info(f"edit_note indexed unindexed file '{candidate}'; retrying resolution")
return await knowledge_client.resolve_entity(identifier, strict=True)
return None
def _compose_workspace_project_route(
*,
workspace: Optional[str],
@@ -126,7 +206,8 @@ The note with identifier '{identifier}' could not be found. The `find_replace` a
## Suggestions to try:
1. **Use append/prepend instead**: These operations will create the note automatically if it doesn't exist
2. **Search for the note first**: Use `search_notes("{project or "project-name"}", "{identifier.split("/")[-1]}")` to find similar notes with exact identifiers
3. **Try different exact identifier formats**:
3. **File exists on disk but is not indexed yet?**: edit_note indexes the file automatically when the identifier matches its path (e.g. 'folder/note' for 'folder/note.md'). If your identifier is a title or differs from the file path, run a sync (`basic-memory sync`) or wait for the file watcher, then retry
4. **Try different exact identifier formats**:
- If you used a permalink like "folder/note-title", try the exact title: "{identifier.split("/")[-1].replace("-", " ").title()}"
- If you used a title, try the exact permalink format: "{identifier.lower().replace(" ", "-")}"
- Use `read_note("{project or "project-name"}", "{identifier}")` first to verify the note exists and get the exact identifier
@@ -343,7 +424,9 @@ async def edit_note(
Note:
Edit operations require exact identifier matches. If unsure, use read_note() or
search_notes() first to find the correct identifier. The tool provides detailed
search_notes() first to find the correct identifier. When the identifier looks
like a file path and the file exists on disk but is not indexed yet, edit_note
indexes that file automatically and retries the edit. The tool provides detailed
error messages with suggestions if operations fail.
"""
# Resolve effective default: allow MCP clients to send null for optional int field
@@ -465,14 +548,27 @@ async def edit_note(
strict=True,
)
except Exception as resolve_error:
# Trigger: entity does not exist yet
# Why: append/prepend can meaningfully create a new note from the content,
# while find_replace/replace_section require existing content to modify
# Outcome: note is created via the same path as write_note
error_msg = str(resolve_error).lower()
is_not_found = "entity not found" in error_msg or "not found" in error_msg
if is_not_found and operation in ("append", "prepend"):
# Trigger: resolution missed but the file may already exist on disk
# Why: files written directly to disk are invisible to identifier
# resolution until indexed; editing them should just work (#581)
# Outcome: the single file is indexed and resolution retried once
recovered_entity_id: str | None = None
if is_not_found:
recovered_entity_id = await _resolve_after_disk_recovery(
knowledge_client, entity_identifier
)
if recovered_entity_id is not None:
entity_id = recovered_entity_id
elif is_not_found and operation in ("append", "prepend"):
# Trigger: entity does not exist yet (on disk or in the index)
# Why: append/prepend can meaningfully create a new note from the
# content, while find_replace/replace_section require existing
# content to modify
# Outcome: note is created via the same path as write_note
title, directory = _parse_identifier_to_title_and_directory(identifier)
# Validate directory path (same security check as write_note)
+2
View File
@@ -9,6 +9,7 @@ from basic_memory.schemas.v2.entity import (
DeleteDirectoryRequestV2,
ProjectResolveRequest,
ProjectResolveResponse,
SyncFileRequest,
)
from basic_memory.schemas.v2.graph import (
GraphEdge,
@@ -31,6 +32,7 @@ __all__ = [
"DeleteDirectoryRequestV2",
"ProjectResolveRequest",
"ProjectResolveResponse",
"SyncFileRequest",
"GraphEdge",
"GraphNode",
"GraphResponse",
+15
View File
@@ -54,6 +54,21 @@ class EntityResolveResponse(BaseModel):
)
class SyncFileRequest(BaseModel):
"""Request to index a single markdown file that exists on disk.
Used as a recovery path when an identifier fails resolution but maps to a
file written directly to disk that the watcher has not indexed yet (#581).
"""
file_path: str = Field(
...,
description="Markdown file path to index (relative to project root)",
min_length=1,
max_length=500,
)
class MoveEntityRequestV2(BaseModel):
"""V2 request schema for moving an entity to a new file location.
@@ -4,6 +4,8 @@ Integration tests for edit_note MCP tool.
Tests the complete edit note workflow: MCP client -> MCP server -> FastAPI -> database
"""
from pathlib import Path
import pytest
from fastmcp import Client
@@ -788,3 +790,39 @@ async def test_edit_note_append_autocreate_does_not_fuzzy_match(mcp_server, app,
error_text = edit_result2.content[0].text
assert "Edit Failed" in error_text
@pytest.mark.asyncio
async def test_edit_note_recovers_file_on_disk_not_indexed(mcp_server, app, test_project):
"""edit_note should index and edit a markdown file written directly to disk (#581).
Common flow: a file is written straight to the project directory and edit_note is
called before the watcher indexes it. The tool must recover by indexing the single
file and retrying resolution instead of failing with "Entity not found".
"""
note_path = Path(test_project.path) / "direct" / "disk-note.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Disk Note\n\nstatus: draft\n", encoding="utf-8")
async with Client(mcp_server) as client:
edit_result = await client.call_tool(
"edit_note",
{
"project": test_project.name,
"identifier": "direct/disk-note",
"operation": "find_replace",
"content": "status: final",
"find_text": "status: draft",
},
)
edit_text = edit_result.content[0].text
assert "Edited note (find_replace)" in edit_text
read_result = await client.call_tool(
"read_note",
{"project": test_project.name, "identifier": "direct/disk-note"},
)
content = read_result.content[0].text
assert "status: final" in content
assert "status: draft" not in content
+515
View File
@@ -1,17 +1,23 @@
"""Tests for V2 knowledge graph API routes (ID-based endpoints)."""
import os
from datetime import datetime, timezone
from pathlib import Path
import uuid
import pytest
from httpx import AsyncClient
from basic_memory.api.v2.routers.knowledge_router import _canonical_file_path
from basic_memory.ignore_utils import get_bmignore_path
from basic_memory.models import Entity as EntityModel, Project
from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.repository.project_repository import ProjectRepository
from basic_memory.repository.search_repository_base import VectorSyncBatchResult
from basic_memory.schemas import DeleteEntitiesResponse
from basic_memory.schemas.response import DirectoryMoveResult, DirectoryDeleteResult
from basic_memory.schemas.v2 import EntityResponseV2, EntityResolveResponse
from basic_memory.services.search_service import SearchService
@pytest.mark.asyncio
@@ -955,3 +961,512 @@ async def test_entity_response_includes_user_tracking_fields(client: AsyncClient
assert "last_updated_by" in body
assert body["created_by"] is None
assert body["last_updated_by"] is None
## Single-file sync endpoint tests
@pytest.mark.asyncio
async def test_sync_file_indexes_file_on_disk(
client: AsyncClient, v2_project_url, test_project: Project
):
"""A markdown file written directly to disk becomes resolvable after sync-file (#581)."""
note_path = Path(test_project.path) / "incoming" / "disk-note.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Disk Note\n\nWritten directly to disk.\n", encoding="utf-8")
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "incoming/disk-note.md"},
)
assert response.status_code == 200
entity = EntityResponseV2.model_validate(response.json())
assert entity.file_path == "incoming/disk-note.md"
# The file is now resolvable by path, which is what edit_note retries with
resolve_response = await client.post(
f"{v2_project_url}/knowledge/resolve",
json={"identifier": "incoming/disk-note.md", "strict": True},
)
assert resolve_response.status_code == 200
resolved = EntityResolveResponse.model_validate(resolve_response.json())
assert resolved.external_id == entity.external_id
@pytest.mark.asyncio
async def test_sync_file_already_indexed_is_idempotent(
client: AsyncClient, v2_project_url, test_project: Project
):
"""sync-file on an already indexed, unchanged file returns the existing entity."""
entity_data = {
"title": "AlreadyIndexed",
"directory": "test",
"content": "Already indexed content",
}
create_response = await client.post(f"{v2_project_url}/knowledge/entities", json=entity_data)
assert create_response.status_code == 200
created = EntityResponseV2.model_validate(create_response.json())
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": created.file_path},
)
assert response.status_code == 200
synced = EntityResponseV2.model_validate(response.json())
assert synced.external_id == created.external_id
assert synced.file_path == created.file_path
@pytest.mark.asyncio
async def test_sync_file_syncs_vectors_when_semantic_enabled(
client: AsyncClient,
v2_project_url,
test_project: Project,
app_config,
monkeypatch: pytest.MonkeyPatch,
):
"""sync-file refreshes semantic vectors for the synced entity.
Mirrors the inline sync_entity_vectors_batch() pass the project sync flow runs
after indexing changed files (SyncService.sync); without it, a note recovered
via sync-file stays missing from semantic search until a later edit or full
sync. Fixtures run with semantic search disabled, so enable it here and stub
the service-level vector batch (like test_search_service.py::test_reindex_vectors
stubs the repository batch) to exercise the wiring without the embedding stack.
"""
app_config.semantic_search_enabled = True
synced_batches: list[list[int]] = []
async def stub_sync_entity_vectors_batch(
self, entity_ids: list[int], progress_callback=None
) -> VectorSyncBatchResult:
synced_batches.append(list(entity_ids))
return VectorSyncBatchResult(
entities_total=len(entity_ids),
entities_synced=len(entity_ids),
entities_failed=0,
)
# The router builds its SearchService per request, so patch the class method
# rather than a fixture instance.
monkeypatch.setattr(SearchService, "sync_entity_vectors_batch", stub_sync_entity_vectors_batch)
note_path = Path(test_project.path) / "incoming" / "semantic-note.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Semantic Note\n\nNeeds vectors after recovery.\n", encoding="utf-8")
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "incoming/semantic-note.md"},
)
assert response.status_code == 200
entity = EntityResponseV2.model_validate(response.json())
assert synced_batches == [[entity.id]]
@pytest.mark.asyncio
async def test_sync_file_skips_vector_sync_when_semantic_disabled(
client: AsyncClient,
v2_project_url,
test_project: Project,
app_config,
monkeypatch: pytest.MonkeyPatch,
):
"""sync-file does not touch the vector pipeline when semantic search is disabled."""
assert app_config.semantic_search_enabled is False
synced_batches: list[list[int]] = []
async def stub_sync_entity_vectors_batch(
self, entity_ids: list[int], progress_callback=None
) -> VectorSyncBatchResult:
synced_batches.append(list(entity_ids))
return VectorSyncBatchResult(
entities_total=len(entity_ids),
entities_synced=len(entity_ids),
entities_failed=0,
)
monkeypatch.setattr(SearchService, "sync_entity_vectors_batch", stub_sync_entity_vectors_batch)
note_path = Path(test_project.path) / "incoming" / "plain-note.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Plain Note\n\nNo vectors needed.\n", encoding="utf-8")
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "incoming/plain-note.md"},
)
assert response.status_code == 200
assert synced_batches == []
@pytest.mark.asyncio
async def test_sync_file_missing_file_returns_404(client: AsyncClient, v2_project_url):
"""sync-file fails fast when the file does not exist on disk."""
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "missing/never-written.md"},
)
assert response.status_code == 404
assert "File not found on disk" in response.json()["detail"]
@pytest.mark.asyncio
async def test_sync_file_rejects_path_traversal(client: AsyncClient, v2_project_url):
"""sync-file rejects paths that escape the project root."""
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "../outside-project.md"},
)
assert response.status_code == 400
assert "project boundaries" in response.json()["detail"]
@pytest.mark.asyncio
async def test_sync_file_rejects_symlink_escape(
client: AsyncClient, v2_project_url, test_project: Project, entity_repository
):
"""sync-file rejects paths whose canonical target escapes the project via symlink.
The exact-cased request ('link/secret.md') is rejected by the pre-canonicalization
boundary check: the path exists, so resolve() follows the symlink and detects the
escape. The wrong-cased request ('LINK/secret.md') is the regression case — on a
case-sensitive filesystem that path does not exist, the pre-check resolves it
lexically and passes; canonicalization then matches the real 'link' segment but
stops at the project boundary and reports the path as not found (404). On
case-insensitive filesystems the pre-check catches both with 400.
"""
project_path = Path(test_project.path)
outside_dir = project_path.parent / "sync-file-outside"
outside_dir.mkdir(parents=True, exist_ok=True)
(outside_dir / "secret.md").write_text(
"# Outside\n\nMust never be indexed.\n", encoding="utf-8"
)
(project_path / "link").symlink_to(outside_dir, target_is_directory=True)
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "link/secret.md"},
)
assert response.status_code == 400
assert "project boundaries" in response.json()["detail"]
# 400 (pre-check, case-insensitive FS) or 404 (canonicalization stops at the
# boundary, case-sensitive FS) — either way the escape is rejected.
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "LINK/secret.md"},
)
assert response.status_code in (400, 404)
# Nothing outside the project root was indexed
assert await entity_repository.find_all() == []
@pytest.mark.asyncio
async def test_sync_file_symlink_escape_never_scans_outside_directory(
client: AsyncClient,
v2_project_url,
test_project: Project,
entity_repository,
monkeypatch: pytest.MonkeyPatch,
):
"""Canonicalization never scans directories outside the project root.
Rejecting the request is not enough: before this fix, the wrong-cased request
('LINK/secret.md') passed the pre-check on a case-sensitive filesystem, and
canonicalization followed the escaping 'link' symlink with os.scandir before the
post-canonicalization containment check fired — an information touch outside the
boundary. We spy on os.scandir and assert the outside directory is never scanned.
The spy is what makes this test meaningful on case-insensitive macOS too, where
the request is already rejected by the pre-check: the assertion proves no layer
scanned past the boundary either way.
"""
project_path = Path(test_project.path)
outside_dir = (project_path.parent / "sync-file-outside-scan").resolve()
outside_dir.mkdir(parents=True, exist_ok=True)
(outside_dir / "secret.md").write_text(
"# Outside\n\nMust never be scanned.\n", encoding="utf-8"
)
(project_path / "link").symlink_to(outside_dir, target_is_directory=True)
real_scandir = os.scandir
scanned: list[Path] = []
def recording_scandir(path=".", *args, **kwargs):
# Record the resolved path so a scandir on the symlinked 'link' directory
# shows up as the outside directory it actually reads.
scanned.append(Path(path).resolve())
return real_scandir(path, *args, **kwargs)
monkeypatch.setattr(os, "scandir", recording_scandir)
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "LINK/secret.md"},
)
assert response.status_code in (400, 404)
assert outside_dir not in scanned
assert await entity_repository.find_all() == []
@pytest.mark.asyncio
async def test_sync_file_symlink_escape_never_probes_outside_target(
client: AsyncClient,
v2_project_url,
test_project: Project,
entity_repository,
monkeypatch: pytest.MonkeyPatch,
):
"""The containment check runs before any filesystem probe that follows symlinks.
Regression: a wrong-cased request ('SECRET.md') canonicalizes onto an in-project
FILE symlink ('secret.md' -> outside target) on a case-sensitive filesystem.
Before the fix, the endpoint's is_file() existence probe ran before the
resolved-containment check, so it followed the symlink and stat'ed the target
outside the project boundary (and the response flipped between 404 and 400
depending on whether the external target existed). We spy on Path.is_file and
assert no probe ever resolves to the outside target; the escape is always
rejected with 400 — by the pre-check on case-insensitive filesystems, by the
post-canonicalization containment check on case-sensitive ones.
"""
project_path = Path(test_project.path)
outside_dir = (project_path.parent / "sync-file-outside-probe").resolve()
outside_dir.mkdir(parents=True, exist_ok=True)
outside_target = outside_dir / "secret.md"
outside_target.write_text("# Outside\n\nMust never be probed.\n", encoding="utf-8")
(project_path / "secret.md").symlink_to(outside_target)
real_is_file = Path.is_file
probed: list[Path] = []
def recording_is_file(self, *args, **kwargs):
# Record the resolved path so a probe on the in-project symlink name shows
# up as the outside target it would actually stat.
probed.append(self.resolve())
return real_is_file(self, *args, **kwargs)
monkeypatch.setattr(Path, "is_file", recording_is_file)
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "SECRET.md"},
)
assert response.status_code == 400
assert "project boundaries" in response.json()["detail"]
assert outside_target not in probed
assert await entity_repository.find_all() == []
def test_canonical_file_path_stops_at_project_boundary(tmp_path: Path):
"""_canonical_file_path bails before descending past the resolved project home.
Exercised directly (not via the endpoint) so the boundary bail is covered on
case-insensitive filesystems too, where the endpoint pre-check rejects the
request before canonicalization runs.
"""
home = tmp_path / "project"
home.mkdir()
outside_dir = tmp_path / "outside"
outside_dir.mkdir()
(outside_dir / "secret.md").write_text("# Outside\n", encoding="utf-8")
(home / "link").symlink_to(outside_dir, target_is_directory=True)
assert _canonical_file_path(home, ["link", "secret.md"]) is None
# Symlinks that stay inside the project keep canonicalizing as before.
real_dir = home / "real"
real_dir.mkdir()
(real_dir / "inside.md").write_text("# Inside\n", encoding="utf-8")
(home / "alias").symlink_to(real_dir, target_is_directory=True)
assert _canonical_file_path(home, ["alias", "inside.md"]) == "alias/inside.md"
@pytest.mark.asyncio
async def test_sync_file_symlink_inside_project_still_indexes(
client: AsyncClient, v2_project_url, test_project: Project
):
"""A symlinked directory that stays inside the project is still accepted.
Pre-existing behavior we preserve: the containment check follows the symlink,
sees the resolved target inside the project root, and indexes the entity under
the requested (symlinked) path — only escapes outside the root are rejected.
"""
project_path = Path(test_project.path)
real_dir = project_path / "real"
real_dir.mkdir(parents=True, exist_ok=True)
(real_dir / "inside.md").write_text("# Inside\n\nReachable via alias.\n", encoding="utf-8")
(project_path / "alias").symlink_to(real_dir, target_is_directory=True)
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "alias/inside.md"},
)
assert response.status_code == 200
entity = EntityResponseV2.model_validate(response.json())
assert entity.file_path == "alias/inside.md"
@pytest.mark.asyncio
async def test_sync_file_wrong_cased_path_does_not_create_duplicate(
client: AsyncClient, v2_project_url, test_project: Project, entity_repository
):
"""A wrong-cased path resolves to the canonical on-disk file without duplicating it.
On case-insensitive filesystems (macOS/Windows) a wrong-cased path passes existence
checks; without canonicalization the indexer would insert a second entity keyed by
the wrong-cased path. The endpoint matches real directory entries, so the request
behaves identically on case-sensitive and case-insensitive filesystems.
"""
note_path = Path(test_project.path) / "notes" / "disk-note.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Disk Note\n\nWritten directly to disk.\n", encoding="utf-8")
first = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "notes/disk-note.md"},
)
assert first.status_code == 200
canonical = EntityResponseV2.model_validate(first.json())
assert canonical.file_path == "notes/disk-note.md"
second = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "notes/Disk-Note.md"},
)
assert second.status_code == 200
synced = EntityResponseV2.model_validate(second.json())
assert synced.file_path == "notes/disk-note.md"
assert synced.external_id == canonical.external_id
entities = await entity_repository.find_all()
assert [entity.file_path for entity in entities] == ["notes/disk-note.md"]
@pytest.mark.asyncio
async def test_sync_file_rejects_non_normalized_segments(
client: AsyncClient, v2_project_url, test_project: Project
):
"""sync-file rejects './' and '//' style segments instead of indexing them verbatim."""
note_path = Path(test_project.path) / "notes" / "disk-note.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Disk Note\n", encoding="utf-8")
for non_normalized in ("./notes/disk-note.md", "notes//disk-note.md"):
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": non_normalized},
)
assert response.status_code == 400
assert "not normalized" in response.json()["detail"]
@pytest.mark.asyncio
async def test_sync_file_directory_returns_404(
client: AsyncClient, v2_project_url, test_project: Project
):
"""sync-file refuses a path that canonicalizes to a directory instead of a file."""
(Path(test_project.path) / "just-a-directory").mkdir(parents=True, exist_ok=True)
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "just-a-directory"},
)
assert response.status_code == 404
assert "File not found on disk" in response.json()["detail"]
@pytest.mark.asyncio
async def test_sync_file_path_through_file_returns_404(
client: AsyncClient, v2_project_url, test_project: Project
):
"""sync-file fails fast when a parent segment resolves to a file, not a directory."""
note_path = Path(test_project.path) / "notes" / "disk-note.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Disk Note\n", encoding="utf-8")
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "notes/disk-note.md/child.md"},
)
assert response.status_code == 404
assert "File not found on disk" in response.json()["detail"]
@pytest.mark.asyncio
async def test_sync_file_rejects_hidden_file(
client: AsyncClient, v2_project_url, test_project: Project
):
"""sync-file refuses hidden files, matching the default '.*' ignore pattern."""
hidden_path = Path(test_project.path) / ".secrets.md"
hidden_path.write_text("# Hidden\n\nShould never be indexed.\n", encoding="utf-8")
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": ".secrets.md"},
)
assert response.status_code == 400
assert "ignore rules" in response.json()["detail"]
@pytest.mark.asyncio
async def test_sync_file_rejects_gitignored_file(
client: AsyncClient, v2_project_url, test_project: Project
):
"""sync-file honors the project .gitignore, matching scan/watch filtering."""
project_path = Path(test_project.path)
(project_path / ".gitignore").write_text("private/\n", encoding="utf-8")
note_path = project_path / "private" / "secret.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Secret\n\nGitignored content.\n", encoding="utf-8")
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "private/secret.md"},
)
assert response.status_code == 400
assert "ignore rules" in response.json()["detail"]
@pytest.mark.asyncio
async def test_sync_file_rejects_bmignored_file(
client: AsyncClient, v2_project_url, test_project: Project
):
"""sync-file honors user .bmignore patterns, matching scan/watch filtering."""
bmignore_path = get_bmignore_path()
bmignore_path.parent.mkdir(parents=True, exist_ok=True)
bmignore_path.write_text("drafts-wip\n", encoding="utf-8")
note_path = Path(test_project.path) / "drafts-wip" / "scratch.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Scratch\n\nBmignored content.\n", encoding="utf-8")
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "drafts-wip/scratch.md"},
)
assert response.status_code == 400
assert "ignore rules" in response.json()["detail"]
@pytest.mark.asyncio
async def test_sync_file_rejects_non_markdown(
client: AsyncClient, v2_project_url, test_project: Project
):
"""sync-file only indexes markdown notes."""
file_path = Path(test_project.path) / "data" / "records.csv"
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text("a,b,c\n", encoding="utf-8")
response = await client.post(
f"{v2_project_url}/knowledge/sync-file",
json={"file_path": "data/records.csv"},
)
assert response.status_code == 400
assert "Only markdown files" in response.json()["detail"]
+30
View File
@@ -133,6 +133,36 @@ class TestKnowledgeClient:
result = await client.resolve_entity("my-note")
assert result == "entity-uuid-123"
@pytest.mark.asyncio
async def test_sync_file(self, monkeypatch):
"""Test sync_file posts the file path to the sync-file endpoint."""
from basic_memory.mcp.clients import knowledge as knowledge_mod
mock_response = MagicMock()
mock_response.json.return_value = {
"permalink": "notes/disk-note",
"title": "Disk Note",
"file_path": "notes/disk-note.md",
"note_type": "note",
"content_type": "text/markdown",
"observations": [],
"relations": [],
"created_at": "2024-01-01T00:00:00",
"updated_at": "2024-01-01T00:00:00",
}
async def mock_call_post(client, url, **kwargs):
assert "/v2/projects/proj-123/knowledge/sync-file" in url
assert kwargs.get("json") == {"file_path": "notes/disk-note.md"}
return mock_response
monkeypatch.setattr(knowledge_mod, "call_post", mock_call_post)
mock_http = MagicMock()
client = KnowledgeClient(mock_http, "proj-123")
result = await client.sync_file("notes/disk-note.md")
assert result.file_path == "notes/disk-note.md"
@pytest.mark.asyncio
async def test_get_orphans_validates_response(self, monkeypatch):
"""Orphan responses are validated into GraphNode objects."""
+266 -1
View File
@@ -1,10 +1,14 @@
"""Tests for the edit_note MCP tool."""
from pathlib import Path
from unittest.mock import patch
import httpx
import pytest
from mcp.server.fastmcp.exceptions import ToolError
from basic_memory.mcp.tools.edit_note import edit_note
from basic_memory.mcp.clients import KnowledgeClient
from basic_memory.mcp.tools.edit_note import _resolve_after_disk_recovery, edit_note
from basic_memory.mcp.tools.read_note import read_note
from basic_memory.mcp.tools.write_note import write_note
@@ -1263,3 +1267,264 @@ async def test_edit_note_skips_detection_when_project_id_provided(
assert isinstance(result, str)
assert "Edited note (append)" in result
assert f"project: {test_project.name}" in result
@pytest.mark.asyncio
async def test_edit_note_find_replace_recovers_file_on_disk_not_indexed(client, test_project):
"""find_replace should index and edit a file written directly to disk (#581)."""
note_path = Path(test_project.path) / "notes" / "disk-note.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Disk Note\n\nstatus: draft\n", encoding="utf-8")
result = await edit_note(
project=test_project.name,
identifier="notes/disk-note",
operation="find_replace",
content="status: final",
find_text="status: draft",
)
assert isinstance(result, str)
assert "Edited note (find_replace)" in result
assert "status: final" in note_path.read_text(encoding="utf-8")
@pytest.mark.asyncio
async def test_edit_note_recovers_identifier_with_md_extension(client, test_project):
"""An identifier already ending in .md should recover via the exact file path (#581)."""
note_path = Path(test_project.path) / "notes" / "exact-path.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Exact Path\n\nversion: v1\n", encoding="utf-8")
result = await edit_note(
project=test_project.name,
identifier="notes/exact-path.md",
operation="find_replace",
content="version: v2",
find_text="version: v1",
)
assert isinstance(result, str)
assert "Edited note (find_replace)" in result
assert "version: v2" in note_path.read_text(encoding="utf-8")
@pytest.mark.asyncio
async def test_edit_note_recovers_identifier_with_markdown_extension(client, test_project):
"""A .markdown identifier must recover via the exact path, not '<path>.markdown.md' (#581)."""
note_path = Path(test_project.path) / "notes" / "alt-suffix.markdown"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Alt Suffix\n\nstate: pending\n", encoding="utf-8")
result = await edit_note(
project=test_project.name,
identifier="notes/alt-suffix.markdown",
operation="find_replace",
content="state: done",
find_text="state: pending",
)
assert isinstance(result, str)
assert "Edited note (find_replace)" in result
assert "state: done" in note_path.read_text(encoding="utf-8")
@pytest.mark.asyncio
async def test_edit_note_refuses_ignored_on_disk_file(client, test_project):
"""An on-disk file matched by .gitignore must be refused, not shadowed by auto-create."""
project_path = Path(test_project.path)
(project_path / ".gitignore").write_text("private/\n", encoding="utf-8")
note_path = project_path / "private" / "secret.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
original_content = "# Secret\n\nGitignored content.\n"
note_path.write_text(original_content, encoding="utf-8")
result = await edit_note(
project=test_project.name,
identifier="private/secret",
operation="append",
content="\nShould never be written.",
)
assert isinstance(result, str)
assert "ignore rules" in result
assert "will not be edited" in result
assert "Edited note" not in result
assert "Created note" not in result
# The ignored file is untouched and auto-create did not shadow it with a new entity
assert note_path.read_text(encoding="utf-8") == original_content
assert [entry.name for entry in (project_path / "private").iterdir()] == ["secret.md"]
@pytest.mark.asyncio
async def test_edit_note_append_recovers_file_on_disk_instead_of_autocreate(client, test_project):
"""append to an unindexed on-disk file should edit it, not auto-create a replacement (#581)."""
note_path = Path(test_project.path) / "notes" / "disk-append.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Disk Append\n\nOriginal disk content.\n", encoding="utf-8")
result = await edit_note(
project=test_project.name,
identifier="notes/disk-append",
operation="append",
content="\nAppended line.",
)
assert isinstance(result, str)
assert "Edited note (append)" in result
assert "Created note" not in result
final_content = note_path.read_text(encoding="utf-8")
assert "Original disk content." in final_content
assert "Appended line." in final_content
@pytest.mark.asyncio
async def test_edit_note_append_recovers_markdown_suffix_file_from_stem(client, test_project):
"""A stem identifier for an on-disk .markdown file edits it, not auto-creates .md (#581).
Recovery probes the identifier as-is, then '.md', then '.markdown'; without the
'.markdown' probe, append would auto-create 'notes/alt-stem.md' next to the real
file instead of editing it.
"""
note_path = Path(test_project.path) / "notes" / "alt-stem.markdown"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Alt Stem\n\nOriginal markdown-suffix content.\n", encoding="utf-8")
result = await edit_note(
project=test_project.name,
identifier="notes/alt-stem",
operation="append",
content="\nAppended line.",
)
assert isinstance(result, str)
assert "Edited note (append)" in result
assert "Created note" not in result
final_content = note_path.read_text(encoding="utf-8")
assert "Original markdown-suffix content." in final_content
assert "Appended line." in final_content
# The real file was edited in place; no shadow .md entity was created beside it
assert [entry.name for entry in note_path.parent.iterdir()] == ["alt-stem.markdown"]
@pytest.mark.asyncio
async def test_edit_note_append_recovers_wrong_cased_identifier(client, test_project):
"""A wrong-cased identifier edits the canonical on-disk file after recovery (#581).
The sync-file endpoint canonicalizes casing by matching real directory entries,
so syncing 'notes/Disk-Note.md' indexes 'notes/disk-note.md' identically on
case-sensitive (CI) and case-insensitive (macOS) filesystems — no filesystem
probe is needed here. The regression: the retry used to strictly re-resolve the
raw wrong-cased identifier, which can miss the just-indexed canonical entity;
the fix returns the entity identity straight from the sync-file response.
"""
note_path = Path(test_project.path) / "notes" / "disk-note.md"
note_path.parent.mkdir(parents=True, exist_ok=True)
note_path.write_text("# Disk Note\n\nOriginal cased content.\n", encoding="utf-8")
result = await edit_note(
project=test_project.name,
identifier="notes/Disk-Note",
operation="append",
content="\nAppended line.",
)
assert isinstance(result, str)
assert "Edited note (append)" in result
assert "Created note" not in result
final_content = note_path.read_text(encoding="utf-8")
assert "Original cased content." in final_content
assert "Appended line." in final_content
# The canonical file was edited; no wrong-cased duplicate was created beside it
assert [entry.name for entry in note_path.parent.iterdir()] == ["disk-note.md"]
@pytest.mark.asyncio
async def test_resolve_after_disk_recovery_falls_back_to_strict_resolve():
"""Older servers that omit external_id from sync-file trigger a strict re-resolve.
The recovery path prefers the entity identity from the sync-file response; when a
server predates that field, the only safe option is a strict re-resolve of the
raw identifier (which fails loudly on a miss instead of guessing).
"""
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/sync-file"):
return httpx.Response(
200,
json={
"permalink": "notes/old-server-note",
"title": "Old Server Note",
"file_path": "notes/old-server-note.md",
"note_type": "note",
"content_type": "text/markdown",
"observations": [],
"relations": [],
"created_at": "2024-01-01T00:00:00",
"updated_at": "2024-01-01T00:00:00",
},
)
assert request.url.path.endswith("/resolve")
return httpx.Response(200, json={"external_id": "resolved-entity-uuid"})
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as http_client:
knowledge_client = KnowledgeClient(http_client, "project-external-id")
result = await _resolve_after_disk_recovery(knowledge_client, "notes/old-server-note")
assert result == "resolved-entity-uuid"
@pytest.mark.asyncio
async def test_resolve_after_disk_recovery_propagates_unexpected_errors():
"""Server-side failures during disk recovery must not be masked as a not-found miss.
Only 400/404 sync-file rejections mean "nothing to recover"; a 500 (or auth
failure) would otherwise be swallowed and edit_note would continue into
auto-create with a misleading not-found error.
"""
def server_error(request: httpx.Request) -> httpx.Response:
return httpx.Response(500, json={"detail": "boom"})
transport = httpx.MockTransport(server_error)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as http_client:
knowledge_client = KnowledgeClient(http_client, "project-external-id")
with pytest.raises(ToolError, match="boom"):
await _resolve_after_disk_recovery(knowledge_client, "notes/unlucky-note")
@pytest.mark.asyncio
async def test_edit_note_append_traversal_identifier_is_blocked(client, test_project):
"""A traversal identifier must be rejected by both disk recovery and auto-create."""
result = await edit_note(
project=test_project.name,
identifier="../escape-note",
operation="append",
content="should never be written",
)
assert isinstance(result, str)
assert "# Error" in result
assert "paths must stay within project boundaries" in result
assert not (Path(test_project.path).parent / "escape-note.md").exists()
@pytest.mark.asyncio
async def test_edit_note_append_traversal_identifier_json_error(client, test_project):
"""JSON mode reports a structured security error for traversal identifiers."""
result = await edit_note(
project=test_project.name,
identifier="../escape-json-note",
operation="append",
content="should never be written",
output_format="json",
)
assert isinstance(result, dict)
assert result["error"] == "SECURITY_VALIDATION_ERROR"
assert result["fileCreated"] is False