diff --git a/src/basic_memory/api/v2/routers/schema_router.py b/src/basic_memory/api/v2/routers/schema_router.py
index 58c66cfc6..373b6dc59 100644
--- a/src/basic_memory/api/v2/routers/schema_router.py
+++ b/src/basic_memory/api/v2/routers/schema_router.py
@@ -16,6 +16,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from basic_memory.deps import (
+    SchemaValidationObserverDep,
     EntityRepositoryV2ExternalDep,
     FileServiceV2ExternalDep,
     LinkResolverV2ExternalDep,
@@ -36,6 +37,10 @@
 from basic_memory.picoschema.resolver import SchemaSearchFn, resolve_schema
 from basic_memory.picoschema.parser import SchemaDefinition
 from basic_memory.picoschema.validator import validate_note
+from basic_memory.services.schema_validation_hooks import (
+    SchemaValidationObserver,
+    ValidatedNoteOutcome,
+)
 from basic_memory.picoschema.inference import infer_schema, NoteData, ObservationData, RelationData
 from basic_memory.picoschema.diff import diff_schema
 from basic_memory.utils import generate_permalink
@@ -152,6 +157,7 @@ async def validate_schema(
     file_service: FileServiceV2ExternalDep,
     link_resolver: LinkResolverV2ExternalDep,
     session: SessionDep,
+    validation_observer: SchemaValidationObserverDep,
     project_id: str = Path(..., description="Project external UUID"),
     note_type: str | None = Query(None, description="Note type to validate"),
     identifier: str | None = Query(None, description="Specific note identifier"),
@@ -168,6 +174,7 @@ async def validate_schema(
     even when file changes haven't been synced to the database yet.
     """
     results: list[NoteValidationResponse] = []
+    outcomes: list[ValidatedNoteOutcome] = []
 
     # --- Single note validation ---
     if identifier:
@@ -175,7 +182,16 @@ async def validate_schema(
         # to match how read_note and other tools resolve identifiers
         entity = await link_resolver.resolve_link(identifier, session=session)
         if not entity:
-            return ValidationReport(note_type=note_type, total_notes=0, total_entities=0)
+            # A request that resolved to nothing still validated nothing, which
+            # is the same report the note-type branch produces for an empty
+            # type. Returning it without telling the observer would make the
+            # once-per-request contract depend on how the request was scoped.
+            return await _observed(
+                validation_observer,
+                project_external_id=project_id,
+                outcomes=outcomes,
+                report=ValidationReport(note_type=note_type, total_notes=0, total_entities=0),
+            )
 
         frontmatter = _entity_frontmatter(entity)
         schema_ref = frontmatter.get("schema")
@@ -198,31 +214,52 @@ async def search_fn(query: str) -> list[dict[str, Any]]:
                 _entity_relations(entity),
                 frontmatter=frontmatter,
             )
-            results.append(_to_note_validation_response(result))
+            response = _to_note_validation_response(result)
+            results.append(response)
+            outcomes.append(
+                ValidatedNoteOutcome(
+                    note_external_id=entity.external_id,
+                    schema_entity=response.schema_entity,
+                    schema_reference=schema_ref if isinstance(schema_ref, str) else None,
+                    passed=response.passed,
+                )
+            )
 
-        return ValidationReport(
-            note_type=note_type or entity.note_type,
-            total_notes=len(results),
-            total_entities=1,
-            valid_count=1 if (results and results[0].passed) else 0,
-            warning_count=sum(len(r.warnings) for r in results),
-            error_count=sum(len(r.errors) for r in results),
-            results=results,
+        return await _observed(
+            validation_observer,
+            project_external_id=project_id,
+            outcomes=outcomes,
+            report=ValidationReport(
+                note_type=note_type or entity.note_type,
+                total_notes=len(results),
+                total_entities=1,
+                valid_count=1 if (results and results[0].passed) else 0,
+                warning_count=sum(len(r.warnings) for r in results),
+                error_count=sum(len(r.errors) for r in results),
+                results=results,
+            ),
         )
 
     # --- Batch validation by note type ---
     if note_type:
         canonical_note_type = normalize_note_type(note_type)
         entities = await _find_by_note_type(session, entity_repository, canonical_note_type)
-        results = await _validate_note_entities(session, entity_repository, file_service, entities)
-        return ValidationReport(
-            note_type=canonical_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,
+        results = await _validate_note_entities(
+            session, entity_repository, file_service, entities, outcomes
+        )
+        return await _observed(
+            validation_observer,
+            project_external_id=project_id,
+            outcomes=outcomes,
+            report=ValidationReport(
+                note_type=canonical_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,
+            ),
         )
 
     # --- All-types validation ---
@@ -237,7 +274,7 @@ async def search_fn(query: str) -> list[dict[str, Any]]:
     for target_type in covered_types:
         entities = await _find_by_note_type(session, entity_repository, target_type)
         type_results = await _validate_note_entities(
-            session, entity_repository, file_service, entities
+            session, entity_repository, file_service, entities, outcomes
         )
         type_summaries.append(
             TypeValidationSummary(
@@ -252,16 +289,41 @@ async def search_fn(query: str) -> list[dict[str, Any]]:
         results.extend(type_results)
         total_entities += len(entities)
 
-    return ValidationReport(
-        note_type=None,
-        total_notes=len(results),
-        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,
+    return await _observed(
+        validation_observer,
+        project_external_id=project_id,
+        outcomes=outcomes,
+        report=ValidationReport(
+            note_type=None,
+            total_notes=len(results),
+            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,
+        ),
+    )
+
+
+async def _observed(
+    observer: SchemaValidationObserver,
+    *,
+    project_external_id: str,
+    outcomes: list[ValidatedNoteOutcome],
+    report: ValidationReport,
+) -> ValidationReport:
+    """Tell the observer what was validated, then return the report unchanged.
+
+    Every exit from `validate_schema` goes through here, so an observer sees a
+    validation exactly once however the request was scoped -- one note, one
+    type, or every schema-covered type.
+    """
+    await observer.on_notes_validated(
+        project_external_id=project_external_id,
+        outcomes=outcomes,
     )
+    return report
 
 
 # --- Inference ---
@@ -379,6 +441,7 @@ async def _validate_note_entities(
     entity_repository: EntityRepositoryV2ExternalDep,
     file_service: FileServiceV2ExternalDep,
     entities: list[Entity],
+    outcomes: list[ValidatedNoteOutcome] | None = None,
 ) -> list[NoteValidationResponse]:
     """Validate a batch of note entities against their resolved schemas.
 
@@ -409,7 +472,17 @@ async def search_fn(query: str) -> list[dict[str, Any]]:
                 _entity_relations(entity),
                 frontmatter=frontmatter,
             )
-            results.append(_to_note_validation_response(result))
+            response = _to_note_validation_response(result)
+            results.append(response)
+            if outcomes is not None:
+                outcomes.append(
+                    ValidatedNoteOutcome(
+                        note_external_id=entity.external_id,
+                        schema_entity=response.schema_entity,
+                        schema_reference=schema_ref if isinstance(schema_ref, str) else None,
+                        passed=response.passed,
+                    )
+                )
 
     return results
 
diff --git a/src/basic_memory/deps/__init__.py b/src/basic_memory/deps/__init__.py
index b79565217..df0146322 100644
--- a/src/basic_memory/deps/__init__.py
+++ b/src/basic_memory/deps/__init__.py
@@ -80,6 +80,7 @@
     NoteContentQueryServiceDep,
     get_note_content_mutation_service,
     NoteContentMutationServiceDep,
+    SchemaValidationObserverDep,
     get_note_content_materialization_provider,
     NoteContentMaterializationProviderDep,
     get_directory_delete_service,
@@ -170,6 +171,7 @@
     "NoteContentQueryServiceDep",
     "get_note_content_mutation_service",
     "NoteContentMutationServiceDep",
+    "SchemaValidationObserverDep",
     "get_note_content_materialization_provider",
     "NoteContentMaterializationProviderDep",
     "get_directory_delete_service",
diff --git a/src/basic_memory/deps/services.py b/src/basic_memory/deps/services.py
index 732f61796..d6fd81256 100644
--- a/src/basic_memory/deps/services.py
+++ b/src/basic_memory/deps/services.py
@@ -37,6 +37,7 @@
 from basic_memory.services.note_content_reads import NoteContentQueryService
 from basic_memory.services.project_readiness import ProjectReadinessService
 from basic_memory.services.note_content_writes import NoteContentMutationService
+from basic_memory.services.schema_validation_hooks import SchemaValidationObserver
 from basic_memory.index.local_dependencies import build_local_markdown_file_indexer
 from basic_memory.index.local_notes import (
     LocalAcceptedNotePreparerFactory,
@@ -369,6 +370,16 @@ async def get_note_content_mutation_service(
 ]
 
 
+async def get_schema_validation_observer() -> SchemaValidationObserver:
+    """Provide the no-op validation observer a deployment can override."""
+    return SchemaValidationObserver()
+
+
+SchemaValidationObserverDep = Annotated[
+    SchemaValidationObserver, Depends(get_schema_validation_observer)
+]
+
+
 # --- Project Indexing ---
 
 
diff --git a/src/basic_memory/services/schema_validation_hooks.py b/src/basic_memory/services/schema_validation_hooks.py
new file mode 100644
index 000000000..d96635bb1
--- /dev/null
+++ b/src/basic_memory/services/schema_validation_hooks.py
@@ -0,0 +1,66 @@
+"""Overridable seam for observing authoritative schema validation.
+
+Validation is the only place that knows whether a note actually satisfies its
+schema: nothing on the write path runs it, and no validation state is stored on
+the entity. A deployment that needs to react to a validation result -- a hosted
+one recording that a user structured a note successfully, say -- would otherwise
+have to re-run validation itself and own a copy of this module's schema
+resolution, which would then drift from the answer the API returns.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class ValidatedNoteOutcome:
+    """One note's validation result, reduced to identity and pass state.
+
+    Deliberately narrower than `NoteValidationResponse`: it carries no field
+    names, values, warnings or error text, so an observer cannot come to depend
+    on note content, and the note is named by its external id rather than by a
+    title. What is left is what an observer can legitimately act on -- which
+    schema, and whether the note satisfied it.
+
+    The two schema fields answer different questions and neither replaces the
+    other. `schema_entity` is the note type the schema covers, read from the
+    schema's own `entity:` frontmatter, so two schema notes that both cover
+    `person` report the same value. `schema_reference` is what the validated
+    note pointed at -- the string in its `schema:` frontmatter -- and is None
+    when the schema was declared inline and there was nothing to point at. An
+    observer that needs to tell two schemas for one entity apart needs the
+    reference; it is the reference as written and matched, not a stable id.
+    """
+
+    note_external_id: str
+    schema_entity: str
+    schema_reference: str | None
+    passed: bool
+
+
+class SchemaValidationObserver:
+    """Observes completed schema validations. A no-op in core."""
+
+    async def on_notes_validated(
+        self,
+        *,
+        project_external_id: str,
+        outcomes: Sequence[ValidatedNoteOutcome],
+    ) -> None:
+        """React to a finished validation, after its report is complete.
+
+        Called once per request with every note the request actually validated,
+        which is not every note it looked at: entities whose frontmatter
+        resolves to no schema are skipped, exactly as they are skipped in the
+        report.
+
+        Unlike `NoteContentMutationService.on_accepted_mutation`, this runs
+        outside any transaction and has nothing to make atomic -- validation
+        reads. An implementation is therefore responsible for its own durability
+        and its own failures: raising here fails the caller's validation
+        request, which is virtually never the right trade for bookkeeping that
+        the user did not ask for.
+        """
+        return None
diff --git a/tests/api/v2/test_schema_router.py b/tests/api/v2/test_schema_router.py
index dbe2e4fb0..32bc532b4 100644
--- a/tests/api/v2/test_schema_router.py
+++ b/tests/api/v2/test_schema_router.py
@@ -7,8 +7,10 @@
 spellings remain part of the same logical population.
 """
 
+from collections.abc import Generator
 from pathlib import Path
 from textwrap import dedent
+from typing import override
 
 import pytest
 from httpx import AsyncClient
@@ -16,7 +18,12 @@
 
 from basic_memory.models import Entity, Project
 from basic_memory.schemas.base import Entity as EntitySchema
+from basic_memory.deps.services import get_schema_validation_observer
 from basic_memory.services.file_service import FileService
+from basic_memory.services.schema_validation_hooks import (
+    SchemaValidationObserver,
+    ValidatedNoteOutcome,
+)
 
 
 # --- Helpers ---
@@ -1250,3 +1257,321 @@ async def test_diff_falls_back_to_db_on_missing_file(
     assert response.status_code == 200
     data = response.json()
     assert data["note_type"] == "diff_missing_type"
+
+
+# --- Validation observer seam ---
+
+
+class RecordingValidationObserver(SchemaValidationObserver):
+    """Captures what a deployment overriding the seam would actually see."""
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, tuple[ValidatedNoteOutcome, ...]]] = []
+
+    @override
+    async def on_notes_validated(
+        self,
+        *,
+        project_external_id: str,
+        outcomes,
+    ) -> None:
+        self.calls.append((project_external_id, tuple(outcomes)))
+
+
+@pytest.fixture
+def validation_observer(app) -> Generator[RecordingValidationObserver, None, None]:
+    observer = RecordingValidationObserver()
+    app.dependency_overrides[get_schema_validation_observer] = lambda: observer
+    yield observer
+    app.dependency_overrides.pop(get_schema_validation_observer, None)
+
+
+@pytest.mark.asyncio
+async def test_validation_observer_sees_a_passing_note_by_external_id(
+    client: AsyncClient,
+    test_project: Project,
+    v2_project_url: str,
+    entity_service,
+    search_service,
+    validation_observer: RecordingValidationObserver,
+):
+    """The seam reports which schema was satisfied, and by which note."""
+    entity, _ = await entity_service.create_or_update_entity(
+        EntitySchema(
+            title="Dave",
+            directory="people",
+            note_type="person",
+            entity_metadata={"schema": {"name": "string", "role": "string"}},
+            content=dedent("""\
+                ## Observations
+                - [name] Dave Wilson
+                - [role] Architect
+            """),
+        )
+    )
+    await search_service.index_entity(entity)
+
+    response = await client.post(
+        f"{v2_project_url}/schema/validate",
+        params={"note_type": "person"},
+    )
+    assert response.status_code == 200
+
+    assert len(validation_observer.calls) == 1
+    project_external_id, outcomes = validation_observer.calls[0]
+    assert project_external_id == test_project.external_id
+    assert len(outcomes) == 1
+    outcome = outcomes[0]
+    assert outcome.note_external_id == entity.external_id
+    assert outcome.schema_entity == "person"
+    assert outcome.passed is True
+
+
+@pytest.mark.asyncio
+async def test_validation_observer_reports_a_failing_note_as_failing(
+    client: AsyncClient,
+    test_project: Project,
+    v2_project_url: str,
+    entity_service,
+    search_service,
+    validation_observer: RecordingValidationObserver,
+):
+    """A note that misses a required field must not look like a pass.
+
+    Awarding on this seam is only safe if `passed` tracks the report.
+    """
+    entity, _ = await entity_service.create_or_update_entity(
+        EntitySchema(
+            title="Erin",
+            directory="people",
+            note_type="person",
+            entity_metadata={
+                "schema": {"name": "string", "role": "string"},
+                # Strict, so the missing field is an error rather than a
+                # warning; `passed` only goes false on errors.
+                "settings": {"validation": "strict"},
+            },
+            content=dedent("""\
+                ## Observations
+                - [name] Erin Only
+            """),
+        )
+    )
+    await search_service.index_entity(entity)
+
+    response = await client.post(
+        f"{v2_project_url}/schema/validate",
+        params={"identifier": "Erin"},
+    )
+    assert response.status_code == 200
+
+    assert len(validation_observer.calls) == 1
+    _, outcomes = validation_observer.calls[0]
+    assert [o.passed for o in outcomes] == [response.json()["results"][0]["passed"]]
+    assert outcomes[0].passed is False
+
+
+@pytest.mark.asyncio
+async def test_validation_observer_is_told_once_when_nothing_was_validated(
+    client: AsyncClient,
+    test_project: Project,
+    v2_project_url: str,
+    validation_observer: RecordingValidationObserver,
+):
+    """An empty validation is still a validation, and reports no outcomes.
+
+    Entities whose frontmatter resolves to no schema are skipped in the report,
+    and they are skipped here too rather than arriving as unvalidated passes.
+    """
+    response = await client.post(
+        f"{v2_project_url}/schema/validate",
+        params={"note_type": "person"},
+    )
+    assert response.status_code == 200
+
+    assert len(validation_observer.calls) == 1
+    _, outcomes = validation_observer.calls[0]
+    assert outcomes == ()
+
+
+@pytest.mark.asyncio
+async def test_validation_observer_sees_every_scope_exactly_once(
+    client: AsyncClient,
+    test_project: Project,
+    v2_project_url: str,
+    entity_service,
+    search_service,
+    validation_observer: RecordingValidationObserver,
+):
+    """All three exits from the endpoint report, and none reports twice.
+
+    The endpoint returns from three separate branches -- one note, one type,
+    every schema-covered type -- and a seam that only covered some of them
+    would award on one scope and silently miss the others.
+    """
+    entity, _ = await entity_service.create_or_update_entity(
+        EntitySchema(
+            title="Dave",
+            directory="people",
+            note_type="person",
+            entity_metadata={"schema": {"name": "string"}},
+            content=dedent("""\
+                ## Observations
+                - [name] Dave Wilson
+            """),
+        )
+    )
+    await search_service.index_entity(entity)
+
+    for params in ({"identifier": "Dave"}, {"note_type": "person"}, {}):
+        response = await client.post(f"{v2_project_url}/schema/validate", params=params)
+        assert response.status_code == 200
+
+    assert len(validation_observer.calls) == 3, "one report per request, whatever its scope"
+    for _, outcomes in validation_observer.calls:
+        assert [o.note_external_id for o in outcomes] == [entity.external_id]
+        assert all(o.passed for o in outcomes)
+
+
+@pytest.mark.asyncio
+async def test_core_ships_a_no_op_observer(
+    client: AsyncClient,
+    test_project: Project,
+    v2_project_url: str,
+    entity_service,
+    search_service,
+):
+    """Without an override, validation behaves exactly as it did before."""
+    entity, _ = await entity_service.create_or_update_entity(
+        EntitySchema(
+            title="Dave",
+            directory="people",
+            note_type="person",
+            entity_metadata={"schema": {"name": "string"}},
+            content="## Observations\n- [name] Dave Wilson\n",
+        )
+    )
+    await search_service.index_entity(entity)
+
+    response = await client.post(
+        f"{v2_project_url}/schema/validate",
+        params={"note_type": "person"},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["results"][0]["passed"] is True
+
+
+@pytest.mark.asyncio
+async def test_validation_observer_is_told_when_the_identifier_resolves_to_nothing(
+    client: AsyncClient,
+    test_project: Project,
+    v2_project_url: str,
+    validation_observer: RecordingValidationObserver,
+):
+    """The fourth exit reports too, or the contract depends on request shape.
+
+    An unresolvable identifier returns the same empty report an empty note type
+    returns. If only one of them notified, "once per request" would quietly mean
+    "once per request, unless you asked by identifier".
+    """
+    response = await client.post(
+        f"{v2_project_url}/schema/validate",
+        params={"identifier": "no-such-note-anywhere"},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["total_notes"] == 0
+    assert len(validation_observer.calls) == 1
+    _, outcomes = validation_observer.calls[0]
+    assert outcomes == ()
+
+
+@pytest.mark.asyncio
+async def test_validation_outcome_names_the_schema_the_note_pointed_at(
+    client: AsyncClient,
+    test_project: Project,
+    v2_project_url: str,
+    entity_service,
+    search_service,
+    validation_observer: RecordingValidationObserver,
+):
+    """Two schema notes can cover one entity, so the entity cannot identify one.
+
+    `schema_entity` comes from the schema's own `entity:` frontmatter, so both
+    of these report `person`. Only the reference the note carried says which
+    schema actually produced the result.
+    """
+    schema_note, _ = await entity_service.create_or_update_entity(
+        EntitySchema(
+            title="strict-person-v2",
+            directory="schemas",
+            note_type="schema",
+            entity_metadata={
+                "entity": "person",
+                "version": 2,
+                "schema": {"name": "string"},
+            },
+            content="Strict person schema.\n",
+        )
+    )
+    await search_service.index_entity(schema_note)
+
+    entity, _ = await entity_service.create_or_update_entity(
+        EntitySchema(
+            title="Fran",
+            directory="people",
+            note_type="person",
+            entity_metadata={"schema": "strict-person-v2"},
+            content=dedent("""\
+                ## Observations
+                - [name] Fran Baker
+            """),
+        )
+    )
+    await search_service.index_entity(entity)
+
+    response = await client.post(
+        f"{v2_project_url}/schema/validate",
+        params={"identifier": "Fran"},
+    )
+    assert response.status_code == 200
+
+    _, outcomes = validation_observer.calls[0]
+    assert len(outcomes) == 1
+    outcome = outcomes[0]
+    assert outcome.schema_entity == "person", "the covered type, shared by every person schema"
+    assert outcome.schema_reference == "strict-person-v2", "the schema this note actually used"
+
+
+@pytest.mark.asyncio
+async def test_an_inline_schema_has_no_reference_to_report(
+    client: AsyncClient,
+    test_project: Project,
+    v2_project_url: str,
+    entity_service,
+    search_service,
+    validation_observer: RecordingValidationObserver,
+):
+    """Nothing was pointed at, so None is the honest answer rather than a guess."""
+    entity, _ = await entity_service.create_or_update_entity(
+        EntitySchema(
+            title="Gus",
+            directory="people",
+            note_type="person",
+            entity_metadata={"schema": {"name": "string"}},
+            content="## Observations\n- [name] Gus Inline\n",
+        )
+    )
+    await search_service.index_entity(entity)
+
+    response = await client.post(
+        f"{v2_project_url}/schema/validate",
+        params={"note_type": "person"},
+    )
+    assert response.status_code == 200
+
+    _, outcomes = validation_observer.calls[0]
+    assert len(outcomes) == 1
+    assert outcomes[0].schema_entity == "person"
+    assert outcomes[0].schema_reference is None