From 6574cb0515544e59b2830c47ca21f0c761b1498e Mon Sep 17 00:00:00 2001 From: phernandez Date: Tue, 14 Jul 2026 23:08:54 -0500 Subject: [PATCH 1/2] fix(mcp): validate all schema-covered types when schema_validate gets no arguments schema_validate() called without note_type or identifier returned an empty report, which the MCP tool rendered as the misleading "No Notes Found of Type 'unknown'" guidance (#1013). Router: when neither parameter is given, enumerate every note type that has a schema note (matching the frontmatter entity field against stored note types via generate_permalink normalization, the same rule implicit type lookup uses), validate all of them, and return an aggregated ValidationReport with a new per-type type_summaries breakdown. The shared per-entity validation loop is extracted into _validate_note_entities. MCP tool: detect all-types mode up front. With no schemas defined, return "No Schemas Defined" guidance (or a clear JSON error) instead of the 'unknown' message; otherwise render the report with a By Type section (e.g. person: 1/1 valid, meeting: no notes). The CLI's 'bm schema validate' with no target flows through the same path and now works as its help text always claimed. Fixes #1013 Co-Authored-By: Claude Fable 5 Signed-off-by: phernandez --- .../api/v2/routers/schema_router.py | 150 ++++++++++++++---- src/basic_memory/mcp/tools/schema.py | 54 ++++++- src/basic_memory/schemas/schema.py | 15 ++ tests/api/v2/test_schema_router.py | 133 ++++++++++++++++ tests/mcp/test_tool_schema.py | 111 +++++++++++++ 5 files changed, 434 insertions(+), 29 deletions(-) diff --git a/src/basic_memory/api/v2/routers/schema_router.py b/src/basic_memory/api/v2/routers/schema_router.py index 85f225874..46850dea8 100644 --- a/src/basic_memory/api/v2/routers/schema_router.py +++ b/src/basic_memory/api/v2/routers/schema_router.py @@ -30,6 +30,7 @@ FieldResultResponse, FieldFrequencyResponse, DriftFieldResponse, + TypeValidationSummary, ) from basic_memory.schema.resolver import resolve_schema from basic_memory.schema.validator import validate_note @@ -140,7 +141,9 @@ async def validate_schema( ): """Validate notes against their resolved schemas. - Validates a specific note (by identifier) or all notes of a given type. + Validates a specific note (by identifier), all notes of a given type, or — + when neither is provided — all notes of every type that has a schema + defined, with a per-type breakdown in type_summaries. Returns warnings/errors based on the schema's validation mode. Schema definitions are read directly from their files to ensure the @@ -191,41 +194,57 @@ async def search_fn(query: str) -> list[dict]: ) # --- Batch validation by note type --- - entities = await _find_by_note_type(session, entity_repository, note_type) if note_type else [] - - for entity in entities: - frontmatter = _entity_frontmatter(entity) - schema_ref = frontmatter.get("schema") - - async def search_fn(query: str) -> list[dict]: - entities = await _find_schema_entities( - session, - entity_repository, - query, - allow_reference_match=isinstance(schema_ref, str) and query == schema_ref, - ) - return [await _schema_frontmatter_from_file(file_service, e) for e in entities] + if note_type: + entities = await _find_by_note_type(session, entity_repository, note_type) + results = await _validate_note_entities(session, entity_repository, file_service, entities) + return ValidationReport( + note_type=note_type, + total_notes=len(results), + total_entities=len(entities), + valid_count=sum(1 for r in results if r.passed), + warning_count=sum(len(r.warnings) for r in results), + error_count=sum(len(r.errors) for r in results), + results=results, + ) - schema_def = await resolve_schema(frontmatter, search_fn) - if schema_def: - result = validate_note( - entity.title or entity.permalink or entity.file_path, - schema_def, - _entity_observations(entity), - _entity_relations(entity), - frontmatter=frontmatter, + # --- All-types validation --- + # Trigger: neither identifier nor note_type was provided + # Why: schema_validate() with no arguments should check every note type that + # has a schema defined instead of returning an empty report (#1013) + # Outcome: aggregated report with a per-type breakdown in type_summaries + covered_types = await _schema_covered_note_types(session, entity_repository) + + type_summaries: list[TypeValidationSummary] = [] + total_entities = 0 + for target_type, stored_note_types in covered_types.items(): + entities = [] + for stored_type in stored_note_types: + entities.extend(await _find_by_note_type(session, entity_repository, stored_type)) + type_results = await _validate_note_entities( + session, entity_repository, file_service, entities + ) + type_summaries.append( + TypeValidationSummary( + note_type=target_type, + total_notes=len(type_results), + total_entities=len(entities), + valid_count=sum(1 for r in type_results if r.passed), + warning_count=sum(len(r.warnings) for r in type_results), + error_count=sum(len(r.errors) for r in type_results), ) - results.append(_to_note_validation_response(result)) + ) + results.extend(type_results) + total_entities += len(entities) - valid = sum(1 for r in results if r.passed) return ValidationReport( - note_type=note_type, + note_type=None, total_notes=len(results), - total_entities=len(entities), - valid_count=valid, + total_entities=total_entities, + valid_count=sum(1 for r in results if r.passed), warning_count=sum(len(r.warnings) for r in results), error_count=sum(len(r.errors) for r in results), results=results, + type_summaries=type_summaries, ) @@ -337,6 +356,81 @@ async def search_fn(query: str) -> list[dict]: # --- Helpers --- +async def _validate_note_entities( + session: AsyncSession, + entity_repository: EntityRepositoryV2ExternalDep, + file_service: FileServiceV2ExternalDep, + entities: list[Entity], +) -> list[NoteValidationResponse]: + """Validate a batch of note entities against their resolved schemas. + + Entities whose frontmatter resolves to no schema are skipped, which is why + a report's total_notes can be lower than its total_entities. + """ + results: list[NoteValidationResponse] = [] + + for entity in entities: + frontmatter = _entity_frontmatter(entity) + schema_ref = frontmatter.get("schema") + + async def search_fn(query: str) -> list[dict]: + found = await _find_schema_entities( + session, + entity_repository, + query, + allow_reference_match=isinstance(schema_ref, str) and query == schema_ref, + ) + return [await _schema_frontmatter_from_file(file_service, e) for e in found] + + schema_def = await resolve_schema(frontmatter, search_fn) + if schema_def: + result = validate_note( + entity.title or entity.permalink or entity.file_path, + schema_def, + _entity_observations(entity), + _entity_relations(entity), + frontmatter=frontmatter, + ) + results.append(_to_note_validation_response(result)) + + return results + + +async def _schema_covered_note_types( + session: AsyncSession, + entity_repository: EntityRepositoryV2ExternalDep, +) -> dict[str, list[str]]: + """Map each schema-declared target type to the stored note_type values it covers. + + Schema notes declare their target via the frontmatter `entity` field. Stored + note types are snake_case while schema authors may write "Person" or "person", + so both sides are compared through generate_permalink normalization — the same + matching rule _find_schema_entities uses for implicit type lookup. Targets with + no matching notes map to an empty list so they still appear in the report. + """ + schema_query = entity_repository.select().where(Entity.note_type == "schema") + schema_result = await entity_repository.execute_query(session, schema_query) + + # normalized target -> raw value as the schema author wrote it (first seen wins) + targets: dict[str, str] = {} + for schema_entity in schema_result.scalars().all(): + target = (schema_entity.entity_metadata or {}).get("entity") + if isinstance(target, str) and target: + targets.setdefault(generate_permalink(target), target) + + # Column-only select: skip eager-load options, they only apply to full-entity queries + type_query = entity_repository.select(Entity.note_type).distinct() + type_result = await entity_repository.execute_query( + session, type_query, use_query_options=False + ) + stored_types = [t for t in type_result.scalars().all() if t] + + return { + raw: [t for t in stored_types if generate_permalink(t) == normalized] + for normalized, raw in sorted(targets.items()) + } + + async def _find_by_note_type( session: AsyncSession, entity_repository: EntityRepositoryV2ExternalDep, diff --git a/src/basic_memory/mcp/tools/schema.py b/src/basic_memory/mcp/tools/schema.py index 08ae03ba8..c99701b6f 100644 --- a/src/basic_memory/mcp/tools/schema.py +++ b/src/basic_memory/mcp/tools/schema.py @@ -32,6 +32,19 @@ def _format_validation_report(report: ValidationReport) -> str: ) lines.append("") + # --- Per-type breakdown (all-types mode) --- + if report.type_summaries: + lines.append("## By Type") + lines.append("") + for summary in report.type_summaries: + if summary.total_entities == 0: + lines.append(f"- **{summary.note_type}**: no notes") + else: + lines.append( + f"- **{summary.note_type}**: {summary.valid_count}/{summary.total_notes} valid" + ) + lines.append("") + # --- Per-note results --- for r in report.results: status = "valid" if r.passed else "INVALID" @@ -167,6 +180,26 @@ def _no_notes_guidance(note_type: str, tool_name: str) -> str: ) +def _no_schemas_defined_guidance(tool_name: str) -> str: + """Build guidance string when validating all types but no schemas exist. + + Used by schema_validate when called without arguments in a project that + has no schema notes — there is nothing to validate yet. + """ + return ( + f"# No Schemas Defined\n\n" + f"`{tool_name}` was called without `note_type` or `identifier`, which " + f"validates every note type that has a schema — but this project has " + f"no schema notes yet, so there is nothing to validate.\n\n" + f"## Next Steps\n\n" + f'1. **Infer a schema** — run `schema_infer("")` to analyze ' + f"existing notes and get a suggested schema\n" + f"2. **Create a schema note** — write a note with `type: schema` and an " + f"`entity` field naming the note type it validates\n" + f"3. **Re-run** — call `{tool_name}()` again once a schema exists\n" + ) + + def _no_schema_guidance(note_type: str, tool_name: str) -> str: """Build guidance string when no schema exists for a note type. @@ -224,7 +257,9 @@ async def schema_validate( ) -> ValidationReport | str | dict: """Validate notes against their resolved schema. - Validates a specific note (by identifier) or all notes of a given type. + Validates a specific note (by identifier), all notes of a given type, or — + when called with neither — all notes of every type that has a schema + defined, with a per-type breakdown. Returns warnings/errors based on the schema's validation mode. Schemas are resolved in priority order: @@ -258,6 +293,9 @@ async def schema_validate( # Validate a specific note schema_validate(identifier="people/paul-graham") + # Validate every type that has a schema defined + schema_validate() + # Validate in a specific project schema_validate(note_type="person", project="my-research") """ @@ -285,6 +323,20 @@ async def schema_validate( f"warnings={result.warning_count} errors={result.error_count}" ) + # --- All-types mode --- + # Trigger: called with neither identifier nor note_type (#1013) + # Why: an empty report here means "no schemas defined", not + # "no notes of type 'unknown'" — the guards below would mislead + # Outcome: per-type summary, or guidance for defining a first schema + if note_type is None and identifier is None: + if not result.type_summaries: + if output_format == "json": + return {"error": "No schemas defined in this project"} + return _no_schemas_defined_guidance("schema_validate") + if output_format == "json": + return result.model_dump(mode="json", exclude_none=True) + return _format_validation_report(result) + # --- No notes guard --- # Trigger: no entities of this type exist in the project # Why: can't validate notes that don't exist yet diff --git a/src/basic_memory/schemas/schema.py b/src/basic_memory/schemas/schema.py index 0194d437a..4207499a3 100644 --- a/src/basic_memory/schemas/schema.py +++ b/src/basic_memory/schemas/schema.py @@ -42,6 +42,17 @@ class NoteValidationResponse(BaseModel): errors: list[str] = Field(default_factory=list) +class TypeValidationSummary(BaseModel): + """Per-type rollup used when validating all schema-covered types at once.""" + + note_type: str + total_notes: int = 0 + total_entities: int = 0 + valid_count: int = 0 + warning_count: int = 0 + error_count: int = 0 + + class ValidationReport(BaseModel): """Full validation report for one or more notes.""" @@ -52,6 +63,10 @@ class ValidationReport(BaseModel): warning_count: int = 0 error_count: int = 0 results: list[NoteValidationResponse] = Field(default_factory=list) + type_summaries: list[TypeValidationSummary] = Field( + default_factory=list, + description="Per-type breakdown, populated when validating all schema-covered types", + ) # --- Inference Response Models --- diff --git a/tests/api/v2/test_schema_router.py b/tests/api/v2/test_schema_router.py index d3cf1c395..36b39c4e3 100644 --- a/tests/api/v2/test_schema_router.py +++ b/tests/api/v2/test_schema_router.py @@ -366,6 +366,139 @@ async def test_validate_total_entities_without_schema( assert data["results"] == [] +# --- All-Types Validation Tests (#1013) --- + + +@pytest.mark.asyncio +async def test_validate_all_types_with_schemas( + client: AsyncClient, + test_project: Project, + v2_project_url: str, + entity_service, + search_service, +): + """Validate with no params covers every note type that has a schema defined.""" + person_schema, _ = await entity_service.create_or_update_entity( + EntitySchema( + title="Person Schema", + directory="schemas", + note_type="schema", + entity_metadata={ + "entity": "person", + "schema": {"name": "string", "role": "string"}, + }, + content=dedent("""\ + ## Observations + - [note] Schema definition for person entities + """), + ) + ) + await search_service.index_entity(person_schema) + + # Second schema whose target type has no notes yet + meeting_schema, _ = await entity_service.create_or_update_entity( + EntitySchema( + title="Meeting Schema", + directory="schemas", + note_type="schema", + entity_metadata={ + "entity": "meeting", + "schema": {"date": "string"}, + }, + content=dedent("""\ + ## Observations + - [note] Schema definition for meeting entities + """), + ) + ) + await search_service.index_entity(meeting_schema) + + await create_person_entities(entity_service, search_service) + + response = await client.post(f"{v2_project_url}/schema/validate") + + assert response.status_code == 200 + data = response.json() + assert data["note_type"] is None + assert data["total_entities"] == 3 + assert data["total_notes"] == 3 + assert data["valid_count"] == 3 + assert len(data["results"]) == 3 + + summaries = {s["note_type"]: s for s in data["type_summaries"]} + assert set(summaries) == {"person", "meeting"} + assert summaries["person"]["total_entities"] == 3 + assert summaries["person"]["total_notes"] == 3 + assert summaries["person"]["valid_count"] == 3 + assert summaries["meeting"]["total_entities"] == 0 + assert summaries["meeting"]["total_notes"] == 0 + + +@pytest.mark.asyncio +async def test_validate_all_types_no_schemas( + client: AsyncClient, + test_project: Project, + v2_project_url: str, + entity_service, + search_service, +): + """Validate with no params returns an empty report when no schemas are defined.""" + # Notes exist, but nothing covers them + await create_person_entities(entity_service, search_service) + + response = await client.post(f"{v2_project_url}/schema/validate") + + assert response.status_code == 200 + data = response.json() + assert data["note_type"] is None + assert data["total_notes"] == 0 + assert data["total_entities"] == 0 + assert data["results"] == [] + assert data["type_summaries"] == [] + + +@pytest.mark.asyncio +async def test_validate_all_types_normalizes_target_type( + client: AsyncClient, + test_project: Project, + v2_project_url: str, + entity_service, + search_service, +): + """Schema declaring entity 'Person' still covers snake_case 'person' notes.""" + schema_entity, _ = await entity_service.create_or_update_entity( + EntitySchema( + title="Person Schema", + directory="schemas", + note_type="schema", + entity_metadata={ + "entity": "Person", + "schema": {"name": "string", "role": "string"}, + }, + content=dedent("""\ + ## Observations + - [note] Schema with capitalized target type + """), + ) + ) + await search_service.index_entity(schema_entity) + + await create_person_entities(entity_service, search_service) + + response = await client.post(f"{v2_project_url}/schema/validate") + + assert response.status_code == 200 + data = response.json() + assert data["total_entities"] == 3 + assert data["total_notes"] == 3 + + summaries = {s["note_type"]: s for s in data["type_summaries"]} + # The summary carries the type label as the schema author wrote it + assert set(summaries) == {"Person"} + assert summaries["Person"]["total_entities"] == 3 + assert summaries["Person"]["valid_count"] == 3 + + # --- Frontmatter Validation Tests --- diff --git a/tests/mcp/test_tool_schema.py b/tests/mcp/test_tool_schema.py index e9a8e9947..6c36016cb 100644 --- a/tests/mcp/test_tool_schema.py +++ b/tests/mcp/test_tool_schema.py @@ -81,6 +81,24 @@ async def run_index() -> None: """ +MEETING_SCHEMA = """\ +--- +title: Meeting +type: schema +entity: meeting +version: 1 +schema: + date: string, meeting date +settings: + validation: warn +--- + +# Meeting + +Schema for meeting entities. +""" + + # --- Success-path tests (full ASGI stack) --- @@ -356,6 +374,99 @@ async def test_schema_diff(app, test_project, index_project): assert "**hobby**" in result +# --- All-types validation (#1013) --- + + +@pytest.mark.asyncio +async def test_schema_validate_all_types(app, test_project, index_project): + """With no arguments, validate every note type that has a schema defined.""" + project_path = Path(test_project.path) + + _write_schema_file(project_path, "schemas/Person.md", PERSON_SCHEMA) + # Meeting schema exists but no meeting notes do + _write_schema_file(project_path, "schemas/Meeting.md", MEETING_SCHEMA) + _write_schema_file( + project_path, + "people/Alice.md", + PERSON_NOTE.format(name="Alice", permalink="alice"), + ) + + await index_project() + + result = await schema_validate(project=test_project.name) + + assert isinstance(result, str) + assert "Schema Validation: all" in result + assert "## By Type" in result + assert "- **person**: 1/1 valid" in result + assert "- **meeting**: no notes" in result + assert "**Alice**" in result + + +@pytest.mark.asyncio +async def test_schema_validate_all_types_json(app, test_project, index_project): + """JSON output in all-types mode includes the per-type breakdown.""" + project_path = Path(test_project.path) + + _write_schema_file(project_path, "schemas/Person.md", PERSON_SCHEMA) + _write_schema_file(project_path, "schemas/Meeting.md", MEETING_SCHEMA) + _write_schema_file( + project_path, + "people/Alice.md", + PERSON_NOTE.format(name="Alice", permalink="alice"), + ) + + await index_project() + + result = await schema_validate(project=test_project.name, output_format="json") + + assert isinstance(result, dict) + assert result["total_notes"] == 1 + assert result["valid_count"] == 1 + + summaries = {s["note_type"]: s for s in result["type_summaries"]} + assert set(summaries) == {"person", "meeting"} + assert summaries["person"]["valid_count"] == 1 + assert summaries["meeting"]["total_entities"] == 0 + + +@pytest.mark.asyncio +async def test_schema_validate_all_types_no_schemas_returns_guidance( + app, test_project, index_project +): + """With no arguments and no schemas defined, return guidance — not 'unknown'. + + Regression test for issue #1013: schema_validate() without note_type or + identifier reported "No Notes Found of Type 'unknown'" instead of + explaining that no schemas exist yet. + """ + project_path = Path(test_project.path) + + # Notes exist, but no schema notes are defined + _write_schema_file( + project_path, + "people/Alice.md", + PERSON_NOTE.format(name="Alice", permalink="alice"), + ) + + await index_project() + + result = await schema_validate(project=test_project.name) + + assert isinstance(result, str) + assert "No Schemas Defined" in result + assert "unknown" not in result + assert "schema_infer" in result + + +@pytest.mark.asyncio +async def test_schema_validate_all_types_no_schemas_json(app, test_project, index_project): + """JSON output with no schemas defined returns a clear error dict.""" + result = await schema_validate(project=test_project.name, output_format="json") + + assert result == {"error": "No schemas defined in this project"} + + # --- write_note metadata → schema workflow --- From 3e2d90595f6682106d3ca03d0b8716431d0e1f3d Mon Sep 17 00:00:00 2001 From: phernandez Date: Wed, 15 Jul 2026 00:03:13 -0500 Subject: [PATCH 2/2] fix(api): discover directly referenced schemas Signed-off-by: phernandez --- .../api/v2/routers/schema_router.py | 53 ++++++---- tests/api/v2/test_schema_router.py | 97 +++++++++++++++++++ 2 files changed, 133 insertions(+), 17 deletions(-) diff --git a/src/basic_memory/api/v2/routers/schema_router.py b/src/basic_memory/api/v2/routers/schema_router.py index 46850dea8..d590516ca 100644 --- a/src/basic_memory/api/v2/routers/schema_router.py +++ b/src/basic_memory/api/v2/routers/schema_router.py @@ -400,34 +400,53 @@ async def _schema_covered_note_types( session: AsyncSession, entity_repository: EntityRepositoryV2ExternalDep, ) -> dict[str, list[str]]: - """Map each schema-declared target type to the stored note_type values it covers. - - Schema notes declare their target via the frontmatter `entity` field. Stored - note types are snake_case while schema authors may write "Person" or "person", - so both sides are compared through generate_permalink normalization — the same - matching rule _find_schema_entities uses for implicit type lookup. Targets with - no matching notes map to an empty list so they still appear in the report. + """Map each schema-covered target type to the stored note_type values it covers. + + Coverage comes from both standalone schema notes and notes that carry inline + schemas or explicit schema references. Stored note types are snake_case while + schema authors may write "Person" or "person", so both sides are compared + through generate_permalink normalization — the same matching rule + _find_schema_entities uses for implicit type lookup. Standalone targets with no + matching notes map to an empty list so they still appear in the report. """ schema_query = entity_repository.select().where(Entity.note_type == "schema") schema_result = await entity_repository.execute_query(session, schema_query) - # normalized target -> raw value as the schema author wrote it (first seen wins) - targets: dict[str, str] = {} + # normalized target -> (display label, stored note types); first label wins + targets: dict[str, tuple[str, set[str]]] = {} for schema_entity in schema_result.scalars().all(): target = (schema_entity.entity_metadata or {}).get("entity") if isinstance(target, str) and target: - targets.setdefault(generate_permalink(target), target) + targets.setdefault(generate_permalink(target), (target, set())) - # Column-only select: skip eager-load options, they only apply to full-entity queries - type_query = entity_repository.select(Entity.note_type).distinct() - type_result = await entity_repository.execute_query( - session, type_query, use_query_options=False + # Column-only select: skip eager-load options, which apply only to full entities. + # Reading metadata here also discovers inline schemas and explicit references; + # those notes are covered even when their type differs from a schema's entity. + note_query = entity_repository.select(Entity.note_type, Entity.entity_metadata).where( + Entity.note_type != "schema" + ) + note_result = await entity_repository.execute_query( + session, note_query, use_query_options=False ) - stored_types = [t for t in type_result.scalars().all() if t] + for stored_type, metadata in note_result.all(): + if not stored_type: + continue + + normalized_type = generate_permalink(stored_type) + if normalized_type in targets: + targets[normalized_type][1].add(stored_type) + + schema_value = (metadata or {}).get("schema") + has_direct_schema = isinstance(schema_value, dict) or ( + isinstance(schema_value, str) and bool(schema_value) + ) + if has_direct_schema: + _, stored_types = targets.setdefault(normalized_type, (stored_type, set())) + stored_types.add(stored_type) return { - raw: [t for t in stored_types if generate_permalink(t) == normalized] - for normalized, raw in sorted(targets.items()) + display_label: sorted(stored_types) + for _, (display_label, stored_types) in sorted(targets.items()) } diff --git a/tests/api/v2/test_schema_router.py b/tests/api/v2/test_schema_router.py index 36b39c4e3..50931dd00 100644 --- a/tests/api/v2/test_schema_router.py +++ b/tests/api/v2/test_schema_router.py @@ -499,6 +499,103 @@ async def test_validate_all_types_normalizes_target_type( assert summaries["Person"]["valid_count"] == 3 +@pytest.mark.asyncio +async def test_validate_all_types_discovers_inline_schema( + client: AsyncClient, + test_project: Project, + v2_project_url: str, + entity_service, + search_service, +): + """No-argument validation includes types covered only by inline schemas.""" + entity, _ = await entity_service.create_or_update_entity( + EntitySchema( + title="Inline Task", + directory="tasks", + note_type="task", + entity_metadata={"schema": {"status": "string"}}, + content=dedent("""\ + ## Observations + - [status] active + """), + ) + ) + await search_service.index_entity(entity) + + response = await client.post(f"{v2_project_url}/schema/validate") + + assert response.status_code == 200 + data = response.json() + assert data["total_entities"] == 1 + assert data["total_notes"] == 1 + assert data["valid_count"] == 1 + assert data["type_summaries"] == [ + { + "note_type": "task", + "total_notes": 1, + "total_entities": 1, + "valid_count": 1, + "warning_count": 0, + "error_count": 0, + } + ] + + +@pytest.mark.asyncio +async def test_validate_all_types_discovers_cross_type_explicit_schema_reference( + client: AsyncClient, + test_project: Project, + v2_project_url: str, + entity_service, + search_service, +): + """Explicit schema references cover a type even when entity targets differ.""" + schema_entity, _ = await entity_service.create_or_update_entity( + EntitySchema( + title="Reusable Work Schema", + directory="schemas", + note_type="schema", + entity_metadata={ + "entity": "work_item", + "schema": {"status": "string"}, + }, + content=dedent("""\ + ## Observations + - [note] Shared schema for work records + """), + ) + ) + await search_service.index_entity(schema_entity) + + task_entity, _ = await entity_service.create_or_update_entity( + EntitySchema( + title="Referenced Task", + directory="tasks", + note_type="task", + entity_metadata={"schema": "reusable-work-schema"}, + content=dedent("""\ + ## Observations + - [status] active + """), + ) + ) + await search_service.index_entity(task_entity) + + response = await client.post(f"{v2_project_url}/schema/validate") + + assert response.status_code == 200 + data = response.json() + assert data["total_entities"] == 1 + assert data["total_notes"] == 1 + assert data["valid_count"] == 1 + + summaries = {summary["note_type"]: summary for summary in data["type_summaries"]} + assert set(summaries) == {"task", "work_item"} + assert summaries["task"]["total_entities"] == 1 + assert summaries["task"]["valid_count"] == 1 + assert summaries["work_item"]["total_entities"] == 0 + + # --- Frontmatter Validation Tests ---