Compare commits

...

1 Commits

Author SHA1 Message Date
claude[bot] d8cb7313db feat(cli): add orphans command to show entities without relations
Adds a new `basic-memory orphans` CLI command that identifies entities
with no incoming or outgoing connections in the knowledge graph, helping
users discover isolated notes that haven't been linked into their knowledge base.

Changes:
- EntityRepository.find_without_relations(): efficient SQL EXISTS subquery
- GET /v2/projects/{id}/knowledge/orphans API endpoint
- OrphanEntitiesResponse schema (entities list + total count)
- KnowledgeClient.get_orphans() typed API client method
- `basic-memory orphans` CLI command with table and --json output
- Tests for repository method, API endpoint, and CLI command

Closes #762

Co-authored-by: Paul Hernandez <phernandez@users.noreply.github.com>
Signed-off-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
2026-04-23 02:13:55 +00:00
11 changed files with 524 additions and 1 deletions
@@ -38,6 +38,7 @@ from basic_memory.schemas.v2 import (
MoveEntityRequestV2,
MoveDirectoryRequestV2,
DeleteDirectoryRequestV2,
OrphanEntitiesResponse,
)
from basic_memory.schemas.response import DirectoryMoveResult, DirectoryDeleteResult
@@ -110,6 +111,43 @@ 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 relations in the knowledge graph.
Orphan entities have no incoming or outgoing connections — they are isolated
nodes that may indicate newly created notes not yet linked to the graph,
or entities whose relations have been removed.
"""
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
@@ -4,6 +4,7 @@ from . import status, db, doctor, import_memory_json, mcp, import_claude_convers
from . import (
import_claude_projects,
import_chatgpt,
orphans,
tool,
project,
format,
@@ -20,6 +21,7 @@ __all__ = [
"import_claude_conversations",
"import_claude_projects",
"import_chatgpt",
"orphans",
"tool",
"project",
"format",
+95
View File
@@ -0,0 +1,95 @@
"""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 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.
Use --json for machine-readable output.
Use --local to force local routing when cloud mode is enabled.
Use --cloud to force cloud routing when cloud mode is disabled.
"""
from basic_memory.cli.commands.command_utils import run_with_cleanup
try:
validate_routing_flags(local, cloud)
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))
else:
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, typer.Exit):
raise
except Exception as e:
logger.error(f"Error fetching orphan entities: {e}")
if json_output:
print(json.dumps({"error": str(e)}, indent=2))
else:
console.print(f"[red]Error: {e}[/red]")
raise typer.Exit(code=1) # pragma: no cover
+1
View File
@@ -24,6 +24,7 @@ if not _version_only_invocation(sys.argv[1:]):
import_claude_projects,
import_memory_json,
mcp,
orphans,
project,
schema,
status,
+25
View File
@@ -275,6 +275,31 @@ class KnowledgeClient:
)
return DirectoryDeleteResult.model_validate(response.json())
# --- Orphan detection ---
async def get_orphans(self) -> list[dict]:
"""Get entities that have no incoming or outgoing relations.
Returns:
List of entity dicts with external_id, title, note_type, file_path
Raises:
ToolError: If the request fails
"""
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:
@@ -5,7 +5,7 @@ from typing import List, Optional, Sequence, Union, Any
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 selectinload
@@ -436,6 +436,32 @@ class EntityRepository(Repository[Entity]):
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.
An orphan entity has no entries as from_id in any relation and no resolved
entries as to_id. These are isolated nodes in the knowledge graph.
Returns:
Sequence of entities with no outgoing or incoming relations
"""
# Trigger: entity appears as a source in any relation (outgoing link)
# Why: even unresolved outgoing links mean the entity references something
# Outcome: entities with any outgoing relation are excluded
has_outgoing = exists().where(Relation.from_id == Entity.id)
# Trigger: entity appears as a resolved target in any relation (incoming link)
# Why: to_id is null for unresolved links; only resolved links form graph edges
# Outcome: entities referenced by resolved links are excluded
has_incoming = exists().where(
Relation.to_id == Entity.id,
Relation.to_id.is_not(None),
)
query = self.select().where(~has_outgoing).where(~has_incoming)
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.
+2
View File
@@ -14,6 +14,7 @@ from basic_memory.schemas.v2.graph import (
GraphEdge,
GraphNode,
GraphResponse,
OrphanEntitiesResponse,
)
from basic_memory.schemas.v2.resource import (
CreateResourceRequest,
@@ -33,6 +34,7 @@ __all__ = [
"GraphEdge",
"GraphNode",
"GraphResponse",
"OrphanEntitiesResponse",
"CreateResourceRequest",
"UpdateResourceRequest",
"ResourceResponse",
+9
View File
@@ -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")
+108
View File
@@ -0,0 +1,108 @@
"""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
data = response.json()
assert data["entities"] == []
assert data["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."""
r1 = await client.post(
f"{v2_project_url}/knowledge/entities",
json={"title": "Orphan One", "directory": "orphan", "content": "No links here"},
)
assert r1.status_code == 200
r2 = await client.post(
f"{v2_project_url}/knowledge/entities",
json={"title": "Orphan Two", "directory": "orphan", "content": "Also no links"},
)
assert r2.status_code == 200
response = await client.get(f"{v2_project_url}/knowledge/orphans")
assert response.status_code == 200
data = response.json()
assert "entities" in data
assert "total" in data
titles = {e["title"] for e in data["entities"]}
assert "Orphan One" in titles
assert "Orphan Two" in titles
assert data["total"] == len(data["entities"])
assert data["total"] == 2
@pytest.mark.asyncio
async def test_get_orphan_entities_excludes_entity_with_outgoing_relation(
client: AsyncClient, v2_project_url
):
"""An entity with an outgoing wiki-link relation is excluded from orphans."""
# Source entity references another via wikilink in content
r_source = await client.post(
f"{v2_project_url}/knowledge/entities",
json={
"title": "Source Note",
"directory": "linked",
"content": "- links_to [[Target Note]]",
},
)
assert r_source.status_code == 200
# Target entity (no outgoing links)
await client.post(
f"{v2_project_url}/knowledge/entities",
json={"title": "Target Note", "directory": "linked", "content": "Referenced entity"},
)
# Unlinked entity - should appear in orphans
await client.post(
f"{v2_project_url}/knowledge/entities",
json={"title": "Standalone Note", "directory": "linked", "content": "No links at all"},
)
response = await client.get(f"{v2_project_url}/knowledge/orphans")
assert response.status_code == 200
data = response.json()
titles = {e["title"] for e in data["entities"]}
# Source has outgoing relation — not an orphan
assert "Source Note" not in titles
# Standalone has no links — is an orphan
assert "Standalone Note" in titles
@pytest.mark.asyncio
async def test_get_orphan_entities_response_shape(client: AsyncClient, v2_project_url):
"""Each entity in the response has the expected fields."""
await client.post(
f"{v2_project_url}/knowledge/entities",
json={"title": "Shape Test", "directory": "shape", "content": "Testing response shape"},
)
response = await client.get(f"{v2_project_url}/knowledge/orphans")
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
entity = next(e for e in data["entities"] if e["title"] == "Shape Test")
assert "external_id" in entity
assert "title" in entity
assert "file_path" in entity
assert "note_type" in entity
assert entity["title"] == "Shape Test"
assert entity["file_path"].endswith(".md")
+112
View File
@@ -0,0 +1,112 @@
"""Tests for the 'basic-memory orphans' CLI command."""
import json
from contextlib import asynccontextmanager
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from typer.testing import CliRunner
from basic_memory.cli.main import app as cli_app
# Register the orphans command on the shared app.
import basic_memory.cli.commands.orphans as orphans_cmd # noqa: F401
runner = CliRunner()
_MOCK_PROJECT_ITEM = MagicMock()
_MOCK_PROJECT_ITEM.name = "test-project"
_MOCK_PROJECT_ITEM.external_id = "11111111-1111-1111-1111-111111111111"
_ORPHAN_ENTITIES = [
{
"external_id": "aaaa-1111",
"title": "Isolated Note",
"file_path": "notes/isolated.md",
"note_type": "note",
},
{
"external_id": "bbbb-2222",
"title": "Dangling Spec",
"file_path": "specs/dangling.md",
"note_type": "spec",
},
]
def _mock_config_manager():
mock_cm = MagicMock()
mock_cm.default_project = "test-project"
return mock_cm
@asynccontextmanager
async def _fake_get_client(project_name=None):
yield MagicMock()
@patch("basic_memory.cli.commands.orphans.ConfigManager")
@patch("basic_memory.cli.commands.orphans.get_active_project", new_callable=AsyncMock)
@patch("basic_memory.cli.commands.orphans.get_client")
@patch("basic_memory.cli.commands.orphans.KnowledgeClient")
def test_orphans_json_output(mock_knowledge_cls, mock_get_client, mock_get_active, mock_config_cls):
"""bm orphans --json outputs a JSON array of entity objects."""
mock_config_cls.return_value = _mock_config_manager()
mock_get_active.return_value = _MOCK_PROJECT_ITEM
mock_get_client.side_effect = _fake_get_client
mock_knowledge_instance = AsyncMock()
mock_knowledge_instance.get_orphans.return_value = _ORPHAN_ENTITIES
mock_knowledge_cls.return_value = mock_knowledge_instance
result = runner.invoke(cli_app, ["orphans", "--json"])
assert result.exit_code == 0, f"CLI failed: {result.output}"
start = result.output.index("[")
data = json.loads(result.output[start:])
assert len(data) == 2
titles = {e["title"] for e in data}
assert "Isolated Note" in titles
assert "Dangling Spec" in titles
@patch("basic_memory.cli.commands.orphans.ConfigManager")
@patch("basic_memory.cli.commands.orphans.get_active_project", new_callable=AsyncMock)
@patch("basic_memory.cli.commands.orphans.get_client")
@patch("basic_memory.cli.commands.orphans.KnowledgeClient")
def test_orphans_table_output(mock_knowledge_cls, mock_get_client, mock_get_active, mock_config_cls):
"""bm orphans (default) renders a Rich table with titles and paths."""
mock_config_cls.return_value = _mock_config_manager()
mock_get_active.return_value = _MOCK_PROJECT_ITEM
mock_get_client.side_effect = _fake_get_client
mock_knowledge_instance = AsyncMock()
mock_knowledge_instance.get_orphans.return_value = _ORPHAN_ENTITIES
mock_knowledge_cls.return_value = mock_knowledge_instance
result = runner.invoke(cli_app, ["orphans"])
assert result.exit_code == 0, f"CLI failed: {result.output}"
assert "Isolated Note" in result.output
assert "Dangling Spec" in result.output
assert "notes/isolated.md" in result.output
@patch("basic_memory.cli.commands.orphans.ConfigManager")
@patch("basic_memory.cli.commands.orphans.get_active_project", new_callable=AsyncMock)
@patch("basic_memory.cli.commands.orphans.get_client")
@patch("basic_memory.cli.commands.orphans.KnowledgeClient")
def test_orphans_no_results(mock_knowledge_cls, mock_get_client, mock_get_active, mock_config_cls):
"""bm orphans prints a success message when no orphans are found."""
mock_config_cls.return_value = _mock_config_manager()
mock_get_active.return_value = _MOCK_PROJECT_ITEM
mock_get_client.side_effect = _fake_get_client
mock_knowledge_instance = AsyncMock()
mock_knowledge_instance.get_orphans.return_value = []
mock_knowledge_cls.return_value = mock_knowledge_instance
result = runner.invoke(cli_app, ["orphans"])
assert result.exit_code == 0, f"CLI failed: {result.output}"
assert "No orphan entities" in result.output
+105
View File
@@ -1114,3 +1114,108 @@ async def test_get_file_path_to_permalink_map(entity_repository: EntityRepositor
assert len(mapping) == 2
assert mapping["test/entity1.md"] == "test/entity1"
assert mapping["test/entity2.md"] == "test/entity2"
@pytest.mark.asyncio
async def test_find_without_relations_returns_isolated_entities(
entity_repository: EntityRepository, session_maker, test_project: Project
):
"""Entities with no outgoing or incoming relations are returned as orphans."""
async with db.scoped_session(session_maker) as session:
orphan = Entity(
project_id=test_project.id,
title="Orphan",
note_type="test",
permalink="orphan/orphan",
file_path="orphan/orphan.md",
content_type="text/markdown",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
source = Entity(
project_id=test_project.id,
title="Source",
note_type="test",
permalink="source/source",
file_path="source/source.md",
content_type="text/markdown",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
target = Entity(
project_id=test_project.id,
title="Target",
note_type="test",
permalink="target/target",
file_path="target/target.md",
content_type="text/markdown",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
session.add_all([orphan, source, target])
await session.flush()
relation = Relation(
project_id=test_project.id,
from_id=source.id,
to_id=target.id,
to_name=target.title,
relation_type="links_to",
)
session.add(relation)
result = await entity_repository.find_without_relations()
titles = {e.title for e in result}
assert "Orphan" in titles
assert "Source" not in titles
assert "Target" not in titles
@pytest.mark.asyncio
async def test_find_without_relations_empty_project(entity_repository: EntityRepository):
"""An empty project returns no orphans."""
result = await entity_repository.find_without_relations()
assert result == []
@pytest.mark.asyncio
async def test_find_without_relations_all_connected(
entity_repository: EntityRepository, session_maker, test_project: Project
):
"""When all entities are connected, no orphans are returned."""
async with db.scoped_session(session_maker) as session:
a = Entity(
project_id=test_project.id,
title="A",
note_type="test",
permalink="conn/a",
file_path="conn/a.md",
content_type="text/markdown",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
b = Entity(
project_id=test_project.id,
title="B",
note_type="test",
permalink="conn/b",
file_path="conn/b.md",
content_type="text/markdown",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
session.add_all([a, b])
await session.flush()
session.add(
Relation(
project_id=test_project.id,
from_id=a.id,
to_id=b.id,
to_name=b.title,
relation_type="connects",
)
)
result = await entity_repository.find_without_relations()
assert result == []