feat(cli): add orphan entity command

Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/api/v2/routers/knowledge_router.py

@@ -38,6 +38,7 @@
     MoveEntityRequestV2,
     MoveDirectoryRequestV2,
     DeleteDirectoryRequestV2,
+    OrphanEntitiesResponse,
 )
 from basic_memory.schemas.response import DirectoryMoveResult, DirectoryDeleteResult
 
@@ -110,6 +111,38 @@ async def get_graph(
         return GraphResponse(nodes=nodes, edges=edges)
 
 
+## Orphan entities endpoint
+
+
+@router.get("/orphans", response_model=OrphanEntitiesResponse)
+async def get_orphan_entities(
+    project_id: ProjectExternalIdPathDep,
+    entity_repository: EntityRepositoryV2ExternalDep,
+) -> OrphanEntitiesResponse:
+    """Return entities that have no incoming or outgoing relations."""
+    with logfire.span(
+        "api.request.knowledge.get_orphans",
+        entrypoint="api",
+        domain="knowledge",
+        action="get_orphans",
+    ):
+        logger.info("API v2 request: get_orphan_entities")
+
+        entities = await entity_repository.find_without_relations()
+        nodes = [
+            GraphNode(
+                external_id=entity.external_id,
+                title=entity.title,
+                note_type=entity.note_type,
+                file_path=entity.file_path,
+            )
+            for entity in entities
+        ]
+
+        logger.info(f"API v2 response: {len(nodes)} orphan entities")
+        return OrphanEntitiesResponse(entities=nodes, total=len(nodes))
+
+
 ## Resolution endpoint
 
 

src/basic_memory/cli/commands/__init__.py

@@ -1,6 +1,6 @@
 """CLI commands for basic-memory."""
 
-from . import status, db, doctor, import_memory_json, mcp, import_claude_conversations
+from . import status, db, doctor, import_memory_json, mcp, import_claude_conversations, orphans
 from . import (
     import_claude_projects,
     import_chatgpt,
@@ -18,6 +18,7 @@
     "import_memory_json",
     "mcp",
     "import_claude_conversations",
+    "orphans",
     "import_claude_projects",
     "import_chatgpt",
     "tool",

src/basic_memory/cli/commands/orphans.py

@@ -0,0 +1,97 @@
+"""Orphans command - show entities with no relations in the knowledge graph."""
+
+import json
+from typing import Annotated, Optional
+
+import typer
+from loguru import logger
+from mcp.server.fastmcp.exceptions import ToolError
+from rich.console import Console
+from rich.table import Table
+
+from basic_memory.cli.app import app
+from basic_memory.cli.commands.routing import force_routing, validate_routing_flags
+from basic_memory.config import ConfigManager
+from basic_memory.mcp.async_client import get_client
+from basic_memory.mcp.clients.knowledge import KnowledgeClient
+from basic_memory.mcp.project_context import get_active_project
+
+console = Console()
+
+
+async def run_orphans(project: Optional[str] = None) -> tuple[str, list[dict]]:
+    """Fetch entities that have no relations in the knowledge graph."""
+    project = project or ConfigManager().default_project
+
+    async with get_client(project_name=project) as client:
+        project_item = await get_active_project(client, project, None)
+        entities = await KnowledgeClient(client, project_item.external_id).get_orphans()
+        return project_item.name, entities
+
+
+@app.command()
+def orphans(
+    project: Annotated[
+        Optional[str],
+        typer.Option(help="The project name."),
+    ] = None,
+    json_output: bool = typer.Option(False, "--json", help="Output in JSON format"),
+    local: bool = typer.Option(
+        False, "--local", help="Force local API routing (ignore cloud mode)"
+    ),
+    cloud: bool = typer.Option(False, "--cloud", help="Force cloud API routing"),
+):
+    """Show entities that have no relations in the knowledge graph.
+
+    Orphan entities have no incoming or outgoing connections. These may indicate
+    newly created notes not yet linked to other entities, or notes that have had
+    their relations removed.
+    """
+    from basic_memory.cli.commands.command_utils import run_with_cleanup
+
+    try:
+        validate_routing_flags(local, cloud)
+        # Trigger: no explicit routing flag provided.
+        # Why: orphans is a project-local graph inspection command like status.
+        # Outcome: default to local routing unless --cloud was explicitly requested.
+        if not local and not cloud:
+            local = True
+        with force_routing(local=local, cloud=cloud):
+            project_name, entities = run_with_cleanup(run_orphans(project))
+
+        if json_output:
+            print(json.dumps(entities, indent=2, default=str))
+            return
+
+        if not entities:
+            console.print(f"[green]No orphan entities in project '{project_name}'[/green]")
+            return
+
+        table = Table(title=f"{project_name}: Entities Without Relations ({len(entities)} total)")
+        table.add_column("Title", style="cyan")
+        table.add_column("File Path", style="yellow")
+        table.add_column("Type", style="green")
+
+        for entity in entities:
+            table.add_row(
+                entity.get("title", ""),
+                entity.get("file_path", ""),
+                entity.get("note_type") or "",
+            )
+
+        console.print(table)
+    except (ValueError, ToolError) as exc:
+        if json_output:
+            print(json.dumps({"error": str(exc)}, indent=2))
+        else:
+            console.print(f"[red]Error: {exc}[/red]")
+        raise typer.Exit(code=1)
+    except typer.Exit:
+        raise
+    except Exception as exc:
+        logger.error(f"Error fetching orphan entities: {exc}")
+        if json_output:
+            print(json.dumps({"error": str(exc)}, indent=2))
+        else:
+            console.print(f"[red]Error: {exc}[/red]")
+        raise typer.Exit(code=1)  # pragma: no cover

src/basic_memory/cli/main.py

@@ -24,6 +24,7 @@ def _version_only_invocation(argv: list[str]) -> bool:
         import_claude_projects,
         import_memory_json,
         mcp,
+        orphans,
         project,
         schema,
         status,

src/basic_memory/mcp/clients/knowledge.py

@@ -275,6 +275,24 @@ async def delete_directory(self, directory: str) -> DirectoryDeleteResult:
             )
         return DirectoryDeleteResult.model_validate(response.json())
 
+    # --- Orphan detection ---
+
+    async def get_orphans(self) -> list[dict]:
+        """Get entities that have no incoming or outgoing relations."""
+        with logfire.span(
+            "mcp.client.knowledge.get_orphans",
+            client_name="knowledge",
+            operation="get_orphans",
+        ):
+            response = await call_get(
+                self.http_client,
+                f"{self._base_path}/orphans",
+                client_name="knowledge",
+                operation="get_orphans",
+                path_template="/v2/projects/{project_id}/knowledge/orphans",
+            )
+        return response.json()["entities"]
+
     # --- Resolution ---
 
     async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:

src/basic_memory/repository/entity_repository.py

@@ -5,7 +5,7 @@
 
 
 from loguru import logger
-from sqlalchemy import select, func
+from sqlalchemy import exists, func, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 from sqlalchemy.orm import load_only, selectinload
@@ -454,6 +454,25 @@ async def get_all_file_paths(self) -> List[str]:
         result = await self.execute_query(query, use_query_options=False)
         return list(result.scalars().all())
 
+    async def find_without_relations(self) -> Sequence[Entity]:
+        """Find entities that have no incoming or outgoing relations."""
+        # Trigger: entity appears as a source in any relation.
+        # Why: even unresolved outgoing links mean the entity references another node.
+        # Outcome: entities with outgoing relations are excluded from the orphan list.
+        has_outgoing = exists().where(Relation.from_id == Entity.id)
+
+        # Trigger: entity appears as the resolved target in any relation.
+        # Why: only resolved relation targets are graph nodes with an incoming edge.
+        # Outcome: entities referenced by resolved links are excluded from orphans.
+        has_incoming = exists().where(
+            Relation.to_id == Entity.id,
+            Relation.to_id.is_not(None),
+        )
+
+        query = self.select().where(~has_outgoing).where(~has_incoming).order_by(Entity.file_path)
+        result = await self.execute_query(query, use_query_options=False)
+        return list(result.scalars().all())
+
     async def get_distinct_directories(self) -> List[str]:
         """Extract unique directory paths from file_path column.
 

src/basic_memory/schemas/v2/__init__.py

@@ -14,6 +14,7 @@
     GraphEdge,
     GraphNode,
     GraphResponse,
+    OrphanEntitiesResponse,
 )
 from basic_memory.schemas.v2.resource import (
     CreateResourceRequest,
@@ -33,6 +34,7 @@
     "GraphEdge",
     "GraphNode",
     "GraphResponse",
+    "OrphanEntitiesResponse",
     "CreateResourceRequest",
     "UpdateResourceRequest",
     "ResourceResponse",

src/basic_memory/schemas/v2/graph.py

@@ -29,3 +29,12 @@ class GraphResponse(BaseModel):
     edges: list[GraphEdge] = Field(
         default_factory=list, description="All resolved relations as edges"
     )
+
+
+class OrphanEntitiesResponse(BaseModel):
+    """Entities that have no incoming or outgoing relations in the knowledge graph."""
+
+    entities: list[GraphNode] = Field(
+        default_factory=list, description="Entities with no relations"
+    )
+    total: int = Field(..., description="Total count of orphan entities")

tests/api/v2/test_orphan_router.py

@@ -0,0 +1,92 @@
+"""Tests for the /knowledge/orphans API endpoint."""
+
+import pytest
+from httpx import AsyncClient
+
+
+@pytest.mark.asyncio
+async def test_get_orphan_entities_empty_project(client: AsyncClient, v2_project_url):
+    """An empty project returns an empty orphans list."""
+    response = await client.get(f"{v2_project_url}/knowledge/orphans")
+
+    assert response.status_code == 200
+    assert response.json() == {"entities": [], "total": 0}
+
+
+@pytest.mark.asyncio
+async def test_get_orphan_entities_returns_unlinked_entities(client: AsyncClient, v2_project_url):
+    """Entities with no relations appear in the orphans endpoint."""
+    first = await client.post(
+        f"{v2_project_url}/knowledge/entities",
+        json={"title": "Orphan One", "directory": "orphan", "content": "No links here"},
+    )
+    second = await client.post(
+        f"{v2_project_url}/knowledge/entities",
+        json={"title": "Orphan Two", "directory": "orphan", "content": "Also no links"},
+    )
+    assert first.status_code == 200
+    assert second.status_code == 200
+
+    response = await client.get(f"{v2_project_url}/knowledge/orphans")
+
+    assert response.status_code == 200
+    data = response.json()
+    titles = {entity["title"] for entity in data["entities"]}
+    assert titles == {"Orphan One", "Orphan Two"}
+    assert data["total"] == 2
+
+
+@pytest.mark.asyncio
+async def test_get_orphan_entities_excludes_incoming_and_outgoing_relation_nodes(
+    client: AsyncClient, v2_project_url
+):
+    """Entities with either side of a resolved relation are excluded from orphans."""
+    target = await client.post(
+        f"{v2_project_url}/knowledge/entities",
+        json={
+            "title": "Target Note",
+            "directory": "linked",
+            "content": "Referenced entity",
+        },
+    )
+    source = await client.post(
+        f"{v2_project_url}/knowledge/entities",
+        json={
+            "title": "Source Note",
+            "directory": "linked",
+            "content": "- links_to [[Target Note]]",
+        },
+    )
+    standalone = await client.post(
+        f"{v2_project_url}/knowledge/entities",
+        json={"title": "Standalone Note", "directory": "linked", "content": "No links"},
+    )
+    assert source.status_code == 200
+    assert target.status_code == 200
+    assert standalone.status_code == 200
+
+    response = await client.get(f"{v2_project_url}/knowledge/orphans")
+
+    assert response.status_code == 200
+    titles = {entity["title"] for entity in response.json()["entities"]}
+    assert "Source Note" not in titles
+    assert "Target Note" not in titles
+    assert "Standalone Note" in titles
+
+
+@pytest.mark.asyncio
+async def test_get_orphan_entities_response_shape(client: AsyncClient, v2_project_url):
+    """Each orphan entity in the response has the expected graph-node fields."""
+    created = await client.post(
+        f"{v2_project_url}/knowledge/entities",
+        json={"title": "Shape Test", "directory": "shape", "content": "Testing shape"},
+    )
+    assert created.status_code == 200
+
+    response = await client.get(f"{v2_project_url}/knowledge/orphans")
+
+    assert response.status_code == 200
+    data = response.json()
+    entity = next(entity for entity in data["entities"] if entity["title"] == "Shape Test")
+    assert set(entity) == {"external_id", "title", "note_type", "file_path"}
+    assert entity["file_path"].endswith(".md")