style: normalize spacing in project context tests

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

CHANGELOG.md

@@ -2,139 +2,13 @@
 
 ## Unreleased
 
-## v0.19.0 (2026-03-07)
-
-### Highlights
-
-- **Semantic vector search** for SQLite and Postgres with FastEmbed embeddings
-- **Schema system** for validating and inferring knowledge base structure
-- **Per-project cloud routing** with API key authentication
-- **Upgraded to FastMCP 3.0** with tool annotations
-- **CLI overhaul** with JSON output, workspace awareness, and project dashboard
-
 ### Features
 
-- **#550**: Add semantic vector search for SQLite and Postgres
-  - FastEmbed-based embeddings with automatic backfill
-  - Hybrid search combining full-text and vector similarity
-  - Score-based fusion replacing RRF for better ranking
-  - `min_similarity` override for tuning search precision
-  - Semantic dependencies are now default, with optional extras fallback
-
-- **#549**: Schema system for Basic Memory
-  - `schema_infer` — infer schema from existing notes
-  - `schema_validate` — validate notes against a schema definition
-  - `schema_diff` — compare schemas across projects
-  - Frontmatter validation support (#597)
-  - Read schema definitions from file instead of stale DB metadata (#635)
-
-- **#555**: Per-project local/cloud routing with API key auth
-  - Individual projects route through cloud while others stay local
-  - `basic-memory cloud set-key` and `basic-memory project set-cloud/set-local`
-  - Stdio MCP honors per-project cloud routing (#590)
-
-- **#598**: Upgrade FastMCP 2.12.3 to 3.0.1 with tool annotations
-
-- **#585**: Add JSON output mode for MCP tools (default text)
-  - `--json` output for CLI commands for scripting and CI
-
-- **#576**: Add workspace selection flow for MCP and CLI
-  - Workspace-aware cloud project listing
-  - CLI refactoring for workspace support
-
-- **#544**: Project-prefixed permalinks and memory URL routing
-
-- **#632**: Add overwrite guard to `write_note` tool
-
-- **#614**: `edit_note` append/prepend auto-creates note if not found
-
-- **#609**: Richer content context in search results
-  - Return matched chunk text in search results (#601)
-  - Improved content hit rate
-
-- **#602**: Add `created_by` and `last_updated_by` user tracking to Entity
-
-- **#600**: Rename `entity_type` to `note_type` across codebase
-
-- **#574**: Add `display_name` and `is_private` to ProjectItem
-
-- **#569**: Expose `external_id` in EntityResponse and link resolver
-
-- **#567**: Isolate default SQLite DB by config dir
-
-- **#560**: Enable `default_project_mode` by default
-
-- **#559**: Add `basic-memory watch` CLI command
-
-- **#546**: Add cloud discovery touchpoints to CLI and MCP
-
-- **#572**: CLI analytics via Umami event collector
-
-- Replace project info with htop-inspired dashboard
-
-- Merge `search_by_metadata` into `search_notes` with optional query
-
 - Add `--strip-frontmatter` to `basic-memory tool read-note`
-
-- Add `destination_folder` parameter to `move_note` tool
-
-### Bug Fixes
-
-- **#644**: Fix default project resolution in cloud mode
-  - ChatGPT search/fetch tools broken in cloud mode
-  - `resolve_project_parameter` falls back to projects API
-
-- **#638**: Restore API backward compatibility for v0.18.x clients
-
-- **#637**: Create backup before config migration overwrites old format
-
-- **#636**: `list_workspaces` bypasses factory pattern on cloud MCP server
-
-- **#631**: `build_context` related_results schema validation failure
-
-- **#613**: Reduce excessive log volume by demoting per-request noise to DEBUG
-
-- **#612**: Handle quoted picoschema enum strings in YAML frontmatter
-
-- **#607**: Guard against closed streams in promo and missing vector tables
-
-- **#606**: Accept null for `expected_replacements` in `edit_note`
-
-- **#595**: `recent_activity` dedup and pagination across MCP tools
-
-- **#593**: Backend-specific distance-to-similarity conversion
-
-- **#582**: Use LinkResolver fallback in `build_context` for flexible identifier matching
-
-- **#577**: Replace RRF with score-based fusion in hybrid search
-
-- **#575**: Remove hardcoded "main" default from `default_project`
-
-- **#534**: Speed up `bm --version` startup
-
-- Fix semantic embeddings not generated on fresh DB or upgrade
-
-- Clarify `search_notes` parameter naming and fix `note_types` case sensitivity
-
-- Parse `tag:` prefix at MCP tool level to avoid hybrid search failure
-
-- Cap sqlite-vec knn k parameter at 4096 limit
-
-- Parameterize SQL queries in search repository type filters
-
-- Coerce list frontmatter values to strings for title and type fields
-
-- Avoid `Post(**metadata)` crash when frontmatter contains 'content' or 'handler' keys
-
-- Upgrade cryptography and python-multipart for security advisories
-
-### Internal
-
-- **#594**: Add `ty` as supplemental type checker
-- Batched vector sync orchestration across repositories
-- FastEmbed parallel guardrails and provider caching
-- Improved cloud CLI status and error messages
-- CI coverage and Postgres test fixes
+  - Default behavior is unchanged: `content` still includes raw markdown with frontmatter.
+  - With `--strip-frontmatter`, both text and JSON modes return body-only markdown content.
+  - JSON output now includes an additive `frontmatter` field with parsed YAML metadata (or `null`
+    when no valid opening frontmatter block exists).
 
 ## v0.18.5 (2026-02-13)
 

justfile

@@ -69,6 +69,24 @@ testmon *args:
 test-smoke:
     BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov -m smoke test-int/mcp/test_smoke_integration.py
 
+# Run graph intelligence API contract tests only
+test-graph-intel-api:
+    BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov tests/api/v2/test_graph_intelligence_router.py
+
+# Run graph intelligence MCP tests only
+test-graph-intel-mcp:
+    BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov tests/mcp/clients/test_graph_clients.py tests/mcp/test_tool_graph_intelligence.py tests/mcp/test_tool_contracts.py
+
+# Run graph intelligence CLI passthrough tests only
+test-graph-intel-cli:
+    BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov tests/cli/test_cli_tool_graph_intelligence_json_output.py
+
+# Run the full graph intelligence fast iteration slice
+test-graph-intel:
+    just test-graph-intel-api
+    just test-graph-intel-mcp
+    just test-graph-intel-cli
+
 # Fast local loop: lint, format, typecheck, impacted tests
 fast-check:
     just fix

server.json

@@ -6,12 +6,12 @@
     "url": "https://github.com/basicmachines-co/basic-memory.git",
     "source": "github"
   },
-  "version": "0.19.0",
+  "version": "0.18.5",
   "packages": [
     {
       "registryType": "pypi",
       "identifier": "basic-memory",
-      "version": "0.19.0",
+      "version": "0.18.5",
       "runtimeHint": "uvx",
       "runtimeArguments": [
         {"type": "positional", "value": "basic-memory"},

src/basic_memory/__init__.py

@@ -1,7 +1,7 @@
 """basic-memory - Local-first knowledge management combining Zettelkasten with knowledge graphs"""
 
 # Package version - updated by release automation
-__version__ = "0.19.0"
+__version__ = "0.18.5"
 
 # API version for FastAPI - independent of package version
 __api_version__ = "v0"

src/basic_memory/api/app.py

@@ -19,6 +19,8 @@ from basic_memory.api.v2.routers import (
     prompt_router as v2_prompt,
     importer_router as v2_importer,
     schema_router as v2_schema,
+    graph_router as v2_graph,
+    fcm_router as v2_fcm,
 )
 from basic_memory.api.v2.routers.project_router import (
     add_project,
@@ -86,6 +88,8 @@ app.include_router(v2_directory, prefix="/v2/projects/{project_id}")
 app.include_router(v2_prompt, prefix="/v2/projects/{project_id}")
 app.include_router(v2_importer, prefix="/v2/projects/{project_id}")
 app.include_router(v2_schema, prefix="/v2/projects/{project_id}")
+app.include_router(v2_graph, prefix="/v2/projects/{project_id}")
+app.include_router(v2_fcm, prefix="/v2/projects/{project_id}")
 app.include_router(v2_project, prefix="/v2")
 
 # Legacy web app proxy paths (compat with /proxy/projects/projects)

src/basic_memory/api/v2/__init__.py

@@ -21,6 +21,8 @@ from basic_memory.api.v2.routers import (
     directory_router,
     prompt_router,
     importer_router,
+    graph_router,
+    fcm_router,
 )
 
 __all__ = [
@@ -32,4 +34,6 @@ __all__ = [
     "directory_router",
     "prompt_router",
     "importer_router",
+    "graph_router",
+    "fcm_router",
 ]

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

@@ -9,6 +9,8 @@ from basic_memory.api.v2.routers.directory_router import router as directory_rou
 from basic_memory.api.v2.routers.prompt_router import router as prompt_router
 from basic_memory.api.v2.routers.importer_router import router as importer_router
 from basic_memory.api.v2.routers.schema_router import router as schema_router
+from basic_memory.api.v2.routers.graph_router import router as graph_router
+from basic_memory.api.v2.routers.fcm_router import router as fcm_router
 
 __all__ = [
     "knowledge_router",
@@ -20,4 +22,6 @@ __all__ = [
     "prompt_router",
     "importer_router",
     "schema_router",
+    "graph_router",
+    "fcm_router",
 ]

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

@@ -0,0 +1,61 @@
+"""V2 router for FCM simulation and interop endpoints."""
+
+from fastapi import APIRouter
+
+from basic_memory.deps import FCMServiceV2ExternalDep, ProjectExternalIdPathDep
+from basic_memory.schemas.graph_intelligence import (
+    FCMExportRequest,
+    FCMExportResponse,
+    FCMImportRequest,
+    FCMImportResponse,
+    FCMRankActionsRequest,
+    FCMRankActionsResponse,
+    FCMSimulateRequest,
+    FCMSimulateResponse,
+)
+
+router = APIRouter(prefix="/fcm", tags=["fcm-v2"])
+
+
+@router.post("/simulate", response_model=FCMSimulateResponse)
+async def fcm_simulate(
+    request: FCMSimulateRequest,
+    fcm_service: FCMServiceV2ExternalDep,
+    project_id: ProjectExternalIdPathDep,
+) -> FCMSimulateResponse:
+    """Run an FCM scenario simulation."""
+    _ = project_id
+    return await fcm_service.simulate(request)
+
+
+@router.post("/rank-actions", response_model=FCMRankActionsResponse)
+async def fcm_rank_actions(
+    request: FCMRankActionsRequest,
+    fcm_service: FCMServiceV2ExternalDep,
+    project_id: ProjectExternalIdPathDep,
+) -> FCMRankActionsResponse:
+    """Rank action candidates toward a goal."""
+    _ = project_id
+    return await fcm_service.rank_actions(request)
+
+
+@router.post("/import", response_model=FCMImportResponse)
+async def fcm_import(
+    request: FCMImportRequest,
+    fcm_service: FCMServiceV2ExternalDep,
+    project_id: ProjectExternalIdPathDep,
+) -> FCMImportResponse:
+    """Import an FCM model using a supported interchange format."""
+    _ = project_id
+    return await fcm_service.import_model(request)
+
+
+@router.post("/export", response_model=FCMExportResponse)
+async def fcm_export(
+    request: FCMExportRequest,
+    fcm_service: FCMServiceV2ExternalDep,
+    project_id: ProjectExternalIdPathDep,
+) -> FCMExportResponse:
+    """Export an FCM model using a supported interchange format."""
+    _ = project_id
+    return await fcm_service.export_model(request)

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

@@ -0,0 +1,71 @@
+"""V2 router for graph intelligence endpoints."""
+
+from fastapi import APIRouter, Query
+
+from basic_memory.deps import (
+    GraphIntelligenceServiceV2ExternalDep,
+    ProjectExternalIdPathDep,
+    TaskSchedulerDep,
+)
+from basic_memory.schemas.graph_intelligence import (
+    GraphHealthResponse,
+    GraphImpactRequest,
+    GraphImpactResponse,
+    GraphLineageRequest,
+    GraphLineageResponse,
+    GraphReindexRequest,
+    GraphReindexResponse,
+)
+
+router = APIRouter(prefix="/graph", tags=["graph-v2"])
+
+
+@router.post("/lineage", response_model=GraphLineageResponse)
+async def graph_lineage(
+    request: GraphLineageRequest,
+    graph_service: GraphIntelligenceServiceV2ExternalDep,
+    project_id: ProjectExternalIdPathDep,
+) -> GraphLineageResponse:
+    """Build lineage paths from a start node toward an optional goal."""
+    _ = project_id
+    return await graph_service.lineage(request)
+
+
+@router.post("/impact", response_model=GraphImpactResponse)
+async def graph_impact(
+    request: GraphImpactRequest,
+    graph_service: GraphIntelligenceServiceV2ExternalDep,
+    project_id: ProjectExternalIdPathDep,
+) -> GraphImpactResponse:
+    """Compute impact radius from a target node."""
+    _ = project_id
+    return await graph_service.impact(request)
+
+
+@router.get("/health", response_model=GraphHealthResponse)
+async def graph_health(
+    graph_service: GraphIntelligenceServiceV2ExternalDep,
+    project_id: ProjectExternalIdPathDep,
+    scope: str | None = Query(default=None),
+    timeframe: str | None = Query(default=None),
+) -> GraphHealthResponse:
+    """Report graph quality metrics and issue candidates."""
+    _ = project_id
+    return await graph_service.health(scope=scope, timeframe=timeframe)
+
+
+@router.post("/reindex", response_model=GraphReindexResponse)
+async def graph_reindex(
+    request: GraphReindexRequest,
+    graph_service: GraphIntelligenceServiceV2ExternalDep,
+    task_scheduler: TaskSchedulerDep,
+    project_id: ProjectExternalIdPathDep,
+) -> GraphReindexResponse:
+    """Queue a graph reindex operation for the current project."""
+    task_scheduler.schedule(
+        "reindex_graph_project",
+        project_id=project_id,
+        mode=request.mode,
+        reason=request.reason,
+    )
+    return await graph_service.start_reindex_job()

src/basic_memory/cli/commands/tool.py

@@ -16,6 +16,13 @@ from basic_memory.cli.commands.command_utils import run_with_cleanup
 from basic_memory.cli.commands.routing import force_routing, validate_routing_flags
 from basic_memory.mcp.tools import build_context as mcp_build_context
 from basic_memory.mcp.tools import edit_note as mcp_edit_note
+from basic_memory.mcp.tools import fcm_export_model as mcp_fcm_export_model
+from basic_memory.mcp.tools import fcm_import_model as mcp_fcm_import_model
+from basic_memory.mcp.tools import fcm_rank_actions as mcp_fcm_rank_actions
+from basic_memory.mcp.tools import fcm_simulate as mcp_fcm_simulate
+from basic_memory.mcp.tools import graph_health as mcp_graph_health
+from basic_memory.mcp.tools import graph_impact as mcp_graph_impact
+from basic_memory.mcp.tools import graph_lineage as mcp_graph_lineage
 from basic_memory.mcp.tools import list_memory_projects as mcp_list_projects
 from basic_memory.mcp.tools import list_workspaces as mcp_list_workspaces
 from basic_memory.mcp.tools import read_note as mcp_read_note
@@ -40,6 +47,17 @@ def _print_json(result: Any) -> None:
     print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
 
 
+def _parse_json_option(raw_value: Optional[str], option_name: str) -> Any:
+    """Parse a JSON CLI option with deterministic error handling."""
+    if raw_value is None:
+        return None
+    try:
+        return json.loads(raw_value)
+    except json.JSONDecodeError as exc:
+        typer.echo(f"Invalid JSON for {option_name}: {exc}", err=True)
+        raise typer.Exit(1)
+
+
 # --- Commands ---
 
 
@@ -366,6 +384,372 @@ def recent_activity(
         raise
 
 
+@tool_app.command("graph-lineage")
+def graph_lineage(
+    start: Annotated[str, typer.Argument(help="Start node identifier or memory:// reference")],
+    goal: Annotated[
+        Optional[str],
+        typer.Option("--goal", help="Optional goal node identifier for targeted lineage"),
+    ] = None,
+    max_hops: int = typer.Option(4, "--max-hops", help="Maximum traversal hops (1-6)"),
+    relation_filters: Annotated[
+        Optional[List[str]],
+        typer.Option("--relation-filter", help="Relation filters (repeatable)"),
+    ] = None,
+    project: Annotated[
+        Optional[str],
+        typer.Option(help="The project to use. If not provided, the default project will be used."),
+    ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
+    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"),
+):
+    """Get graph lineage paths from a start node."""
+    try:
+        validate_routing_flags(local, cloud)
+        with force_routing(local=local, cloud=cloud):
+            result = run_with_cleanup(
+                mcp_graph_lineage(
+                    start=start,
+                    goal=goal,
+                    max_hops=max_hops,
+                    relation_filters=relation_filters or [],
+                    project=project,
+                    workspace=workspace,
+                    output_format="json",
+                )
+            )
+        _print_json(result)
+    except ValueError as e:
+        typer.echo(f"Error: {e}", err=True)
+        raise typer.Exit(1)
+    except Exception as e:  # pragma: no cover
+        if not isinstance(e, typer.Exit):
+            typer.echo(f"Error during graph_lineage: {e}", err=True)
+            raise typer.Exit(1)
+        raise
+
+
+@tool_app.command("graph-impact")
+def graph_impact(
+    target: Annotated[str, typer.Argument(help="Target node identifier or memory:// reference")],
+    horizon: int = typer.Option(2, "--horizon", help="Impact horizon in hops (1-4)"),
+    relation_filters: Annotated[
+        Optional[List[str]],
+        typer.Option("--relation-filter", help="Relation filters (repeatable)"),
+    ] = None,
+    include_reasons: bool = typer.Option(
+        True,
+        "--include-reasons/--no-include-reasons",
+        help="Include reason strings in impact output",
+    ),
+    project: Annotated[
+        Optional[str],
+        typer.Option(help="The project to use. If not provided, the default project will be used."),
+    ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
+    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"),
+):
+    """Get impact radius for a target node."""
+    try:
+        validate_routing_flags(local, cloud)
+        with force_routing(local=local, cloud=cloud):
+            result = run_with_cleanup(
+                mcp_graph_impact(
+                    target=target,
+                    horizon=horizon,
+                    relation_filters=relation_filters or [],
+                    include_reasons=include_reasons,
+                    project=project,
+                    workspace=workspace,
+                    output_format="json",
+                )
+            )
+        _print_json(result)
+    except ValueError as e:
+        typer.echo(f"Error: {e}", err=True)
+        raise typer.Exit(1)
+    except Exception as e:  # pragma: no cover
+        if not isinstance(e, typer.Exit):
+            typer.echo(f"Error during graph_impact: {e}", err=True)
+            raise typer.Exit(1)
+        raise
+
+
+@tool_app.command("graph-health")
+def graph_health(
+    scope: Annotated[Optional[str], typer.Option("--scope", help="Optional scope prefix")] = None,
+    timeframe: Annotated[
+        Optional[str], typer.Option("--timeframe", help="Optional timeframe filter")
+    ] = None,
+    project: Annotated[
+        Optional[str],
+        typer.Option(help="The project to use. If not provided, the default project will be used."),
+    ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
+    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"),
+):
+    """Get graph health metrics and issue candidates."""
+    try:
+        validate_routing_flags(local, cloud)
+        with force_routing(local=local, cloud=cloud):
+            result = run_with_cleanup(
+                mcp_graph_health(
+                    scope=scope,
+                    timeframe=timeframe,
+                    project=project,
+                    workspace=workspace,
+                    output_format="json",
+                )
+            )
+        _print_json(result)
+    except ValueError as e:
+        typer.echo(f"Error: {e}", err=True)
+        raise typer.Exit(1)
+    except Exception as e:  # pragma: no cover
+        if not isinstance(e, typer.Exit):
+            typer.echo(f"Error during graph_health: {e}", err=True)
+            raise typer.Exit(1)
+        raise
+
+
+@tool_app.command("fcm-simulate")
+def fcm_simulate(
+    actions_json: Annotated[
+        str,
+        typer.Option(
+            "--actions-json",
+            help='JSON array of actions, e.g. [{"node_id":"n1","delta":0.2}]',
+        ),
+    ],
+    scenario_json: Annotated[
+        Optional[str],
+        typer.Option("--scenario-json", help="Optional JSON scenario object"),
+    ] = None,
+    clamp_rules_json: Annotated[
+        Optional[str],
+        typer.Option("--clamp-rules-json", help="Optional JSON array of clamp rules"),
+    ] = None,
+    project: Annotated[
+        Optional[str],
+        typer.Option(help="The project to use. If not provided, the default project will be used."),
+    ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
+    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"),
+):
+    """Run an FCM simulation."""
+    actions = _parse_json_option(actions_json, "--actions-json")
+    scenario = _parse_json_option(scenario_json, "--scenario-json")
+    clamp_rules = _parse_json_option(clamp_rules_json, "--clamp-rules-json")
+    if not isinstance(actions, list):
+        typer.echo("Invalid JSON for --actions-json: expected a JSON array", err=True)
+        raise typer.Exit(1)
+    if scenario is not None and not isinstance(scenario, dict):
+        typer.echo("Invalid JSON for --scenario-json: expected a JSON object", err=True)
+        raise typer.Exit(1)
+    if clamp_rules is not None and not isinstance(clamp_rules, list):
+        typer.echo("Invalid JSON for --clamp-rules-json: expected a JSON array", err=True)
+        raise typer.Exit(1)
+
+    try:
+        validate_routing_flags(local, cloud)
+        with force_routing(local=local, cloud=cloud):
+            result = run_with_cleanup(
+                mcp_fcm_simulate(
+                    actions=actions,
+                    scenario=scenario,
+                    clamp_rules=clamp_rules,
+                    project=project,
+                    workspace=workspace,
+                    output_format="json",
+                )
+            )
+        _print_json(result)
+    except ValueError as e:
+        typer.echo(f"Error: {e}", err=True)
+        raise typer.Exit(1)
+    except Exception as e:  # pragma: no cover
+        if not isinstance(e, typer.Exit):
+            typer.echo(f"Error during fcm_simulate: {e}", err=True)
+            raise typer.Exit(1)
+        raise
+
+
+@tool_app.command("fcm-rank-actions")
+def fcm_rank_actions(
+    goal: Annotated[str, typer.Argument(help="Goal node identifier")],
+    constraints_json: Annotated[
+        Optional[str],
+        typer.Option("--constraints-json", help="Optional JSON object of ranking constraints"),
+    ] = None,
+    top_k: int = typer.Option(10, "--top-k", help="Number of recommendations to return"),
+    project: Annotated[
+        Optional[str],
+        typer.Option(help="The project to use. If not provided, the default project will be used."),
+    ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
+    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"),
+):
+    """Rank intervention actions for an FCM goal."""
+    constraints = _parse_json_option(constraints_json, "--constraints-json")
+    if constraints is not None and not isinstance(constraints, dict):
+        typer.echo("Invalid JSON for --constraints-json: expected a JSON object", err=True)
+        raise typer.Exit(1)
+
+    try:
+        validate_routing_flags(local, cloud)
+        with force_routing(local=local, cloud=cloud):
+            result = run_with_cleanup(
+                mcp_fcm_rank_actions(
+                    goal=goal,
+                    constraints=constraints,
+                    top_k=top_k,
+                    project=project,
+                    workspace=workspace,
+                    output_format="json",
+                )
+            )
+        _print_json(result)
+    except ValueError as e:
+        typer.echo(f"Error: {e}", err=True)
+        raise typer.Exit(1)
+    except Exception as e:  # pragma: no cover
+        if not isinstance(e, typer.Exit):
+            typer.echo(f"Error during fcm_rank_actions: {e}", err=True)
+            raise typer.Exit(1)
+        raise
+
+
+@tool_app.command("fcm-import-model")
+def fcm_import_model(
+    source: Annotated[str, typer.Argument(help="Source path or URI for import payload")],
+    format: Annotated[
+        str,
+        typer.Option("--format", help="Import format (currently csv_bundle_v1)"),
+    ] = "csv_bundle_v1",
+    merge_mode: Annotated[
+        str,
+        typer.Option("--merge-mode", help="Merge strategy: replace or upsert"),
+    ] = "upsert",
+    project: Annotated[
+        Optional[str],
+        typer.Option(help="The project to use. If not provided, the default project will be used."),
+    ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
+    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"),
+):
+    """Import an FCM model."""
+    try:
+        validate_routing_flags(local, cloud)
+        with force_routing(local=local, cloud=cloud):
+            result = run_with_cleanup(
+                mcp_fcm_import_model(
+                    source=source,
+                    format=format,  # pyright: ignore[reportArgumentType]
+                    merge_mode=merge_mode,  # pyright: ignore[reportArgumentType]
+                    project=project,
+                    workspace=workspace,
+                    output_format="json",
+                )
+            )
+        _print_json(result)
+    except ValueError as e:
+        typer.echo(f"Error: {e}", err=True)
+        raise typer.Exit(1)
+    except Exception as e:  # pragma: no cover
+        if not isinstance(e, typer.Exit):
+            typer.echo(f"Error during fcm_import_model: {e}", err=True)
+            raise typer.Exit(1)
+        raise
+
+
+@tool_app.command("fcm-export-model")
+def fcm_export_model(
+    format: Annotated[
+        str,
+        typer.Option("--format", help="Export format (currently csv_bundle_v1)"),
+    ] = "csv_bundle_v1",
+    selection_json: Annotated[
+        Optional[str],
+        typer.Option("--selection-json", help="Optional JSON object selection payload"),
+    ] = None,
+    project: Annotated[
+        Optional[str],
+        typer.Option(help="The project to use. If not provided, the default project will be used."),
+    ] = None,
+    workspace: Annotated[
+        Optional[str],
+        typer.Option(help="Cloud workspace tenant ID or unique name to route this request."),
+    ] = None,
+    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"),
+):
+    """Export an FCM model."""
+    selection = _parse_json_option(selection_json, "--selection-json")
+    if selection is not None and not isinstance(selection, dict):
+        typer.echo("Invalid JSON for --selection-json: expected a JSON object", err=True)
+        raise typer.Exit(1)
+
+    try:
+        validate_routing_flags(local, cloud)
+        with force_routing(local=local, cloud=cloud):
+            result = run_with_cleanup(
+                mcp_fcm_export_model(
+                    format=format,  # pyright: ignore[reportArgumentType]
+                    selection=selection,
+                    project=project,
+                    workspace=workspace,
+                    output_format="json",
+                )
+            )
+        _print_json(result)
+    except ValueError as e:
+        typer.echo(f"Error: {e}", err=True)
+        raise typer.Exit(1)
+    except Exception as e:  # pragma: no cover
+        if not isinstance(e, typer.Exit):
+            typer.echo(f"Error during fcm_export_model: {e}", err=True)
+            raise typer.Exit(1)
+        raise
+
+
 @tool_app.command("search-notes")
 def search_notes(
     query: Annotated[

src/basic_memory/cli/promo.py

@@ -12,7 +12,7 @@ from basic_memory.config import ConfigManager
 
 OSS_DISCOUNT_CODE = "BMFOSS"
 CLOUD_LEARN_MORE_URL = (
-    "https://basicmemory.com?utm_source=bm-foss&utm_medium=promo&utm_campaign=cloud-upsell"
+    "https://basicmemory.com?utm_source=bm-cli&utm_medium=promo&utm_campaign=cloud-upsell"
 )
 
 

src/basic_memory/db.py

@@ -43,37 +43,40 @@ if sys.platform == "win32":  # pragma: no cover
 _engine: Optional[AsyncEngine] = None
 _session_maker: Optional[async_sessionmaker[AsyncSession]] = None
 
+# Alembic revision that enables one-time automatic embedding backfill.
+SEMANTIC_EMBEDDING_BACKFILL_REVISION = "i2c3d4e5f6g7"
 
-async def _needs_semantic_embedding_backfill(
-    app_config: BasicMemoryConfig,
+
+async def _load_applied_alembic_revisions(
     session_maker: async_sessionmaker[AsyncSession],
-) -> bool:
-    """Check if entities exist but vector embeddings are empty.
+) -> set[str]:
+    """Load applied Alembic revisions from alembic_version.
 
-    This is the reliable way to detect that embeddings need to be generated,
-    regardless of how migrations were applied (fresh DB, upgrade, reset, etc.).
+    Returns an empty set when the version table does not exist yet
+    (fresh database before first migration).
     """
-    if not app_config.semantic_search_enabled:
-        return False
-
     try:
         async with scoped_session(session_maker) as session:
-            entity_count = (
-                await session.execute(text("SELECT COUNT(*) FROM entity"))
-            ).scalar() or 0
-            if entity_count == 0:
-                return False
-
-            # Check if vector chunks table exists and is empty
-            embedding_count = (
-                await session.execute(text("SELECT COUNT(*) FROM search_vector_chunks"))
-            ).scalar() or 0
-
-            return embedding_count == 0
+            result = await session.execute(text("SELECT version_num FROM alembic_version"))
+            return {str(row[0]) for row in result.fetchall() if row[0]}
     except Exception as exc:
-        # Table might not exist yet (pre-migration)
-        logger.debug(f"Could not check embedding status: {exc}")
-        return False
+        error_message = str(exc).lower()
+        if "alembic_version" in error_message and (
+            "no such table" in error_message or "does not exist" in error_message
+        ):
+            return set()
+        raise
+
+
+def _should_run_semantic_embedding_backfill(
+    revisions_before_upgrade: set[str],
+    revisions_after_upgrade: set[str],
+) -> bool:
+    """Check if this migration run newly applied the backfill-trigger revision."""
+    return (
+        SEMANTIC_EMBEDDING_BACKFILL_REVISION in revisions_after_upgrade
+        and SEMANTIC_EMBEDDING_BACKFILL_REVISION not in revisions_before_upgrade
+    )
 
 
 async def _run_semantic_embedding_backfill(
@@ -477,9 +480,26 @@ async def run_migrations(
     Note: Alembic tracks which migrations have been applied via the alembic_version table,
     so it's safe to call this multiple times - it will only run pending migrations.
     """
-    logger.info("Running database migrations...")
+    logger.debug("Running database migrations...")
     temp_engine: AsyncEngine | None = None
     try:
+        revisions_before_upgrade: set[str] = set()
+        # Trigger: run_migrations() can be invoked before module-level session maker is set.
+        # Why: we still need reliable before/after revision detection for one-time backfill.
+        # Outcome: create a short-lived session maker when needed, then dispose it immediately.
+        if _session_maker is None:
+            precheck_engine, temp_session_maker = _create_engine_and_session(
+                app_config.database_path,
+                database_type,
+                app_config,
+            )
+            try:
+                revisions_before_upgrade = await _load_applied_alembic_revisions(temp_session_maker)
+            finally:
+                await precheck_engine.dispose()
+        else:
+            revisions_before_upgrade = await _load_applied_alembic_revisions(_session_maker)
+
         # Get the absolute path to the alembic directory relative to this file
         alembic_dir = Path(__file__).parent / "alembic"
         config = Config()
@@ -499,7 +519,7 @@ async def run_migrations(
         config.set_main_option("sqlalchemy.url", db_url)
 
         command.upgrade(config, "head")
-        logger.info("Migrations completed successfully")
+        logger.debug("Migrations completed successfully")
 
         # Get session maker - ensure we don't trigger recursive migration calls
         if _session_maker is None:
@@ -521,14 +541,12 @@ async def run_migrations(
         else:
             await SQLiteSearchRepository(session_maker, 1).init_search_index()
 
-        # Check if backfill is needed — actual backfill runs in background
-        # from the MCP server lifespan to avoid blocking startup.
-        if await _needs_semantic_embedding_backfill(app_config, session_maker):
-            logger.info(
-                "Semantic embeddings missing — backfill will run in background after startup"
-            )
-        else:
-            logger.info("Semantic embeddings: up to date")
+        revisions_after_upgrade = await _load_applied_alembic_revisions(session_maker)
+        if _should_run_semantic_embedding_backfill(
+            revisions_before_upgrade,
+            revisions_after_upgrade,
+        ):
+            await _run_semantic_embedding_backfill(app_config, session_maker)
     except Exception as e:  # pragma: no cover
         logger.error(f"Error running migrations: {e}")
         raise

src/basic_memory/deps/__init__.py

@@ -131,6 +131,10 @@ from basic_memory.deps.services import (
     DirectoryServiceV2Dep,
     get_directory_service_v2_external,
     DirectoryServiceV2ExternalDep,
+    get_graph_intelligence_service_v2_external,
+    GraphIntelligenceServiceV2ExternalDep,
+    get_fcm_service_v2_external,
+    FCMServiceV2ExternalDep,
 )
 
 from basic_memory.deps.importers import (
@@ -269,6 +273,10 @@ __all__ = [
     "DirectoryServiceV2Dep",
     "get_directory_service_v2_external",
     "DirectoryServiceV2ExternalDep",
+    "get_graph_intelligence_service_v2_external",
+    "GraphIntelligenceServiceV2ExternalDep",
+    "get_fcm_service_v2_external",
+    "FCMServiceV2ExternalDep",
     # Importers
     "get_chatgpt_importer",
     "ChatGPTImporterDep",

src/basic_memory/deps/services.py

@@ -39,6 +39,8 @@ from basic_memory.deps.repositories import (
 from basic_memory.markdown import EntityParser
 from basic_memory.markdown.markdown_processor import MarkdownProcessor
 from basic_memory.services import EntityService, ProjectService
+from basic_memory.services.fcm_service import FCMService
+from basic_memory.services.graph_intelligence_service import GraphIntelligenceService
 from basic_memory.services.context_service import ContextService
 from basic_memory.services.directory_service import DirectoryService
 from basic_memory.services.file_service import FileService
@@ -358,6 +360,30 @@ async def get_context_service_v2_external(
 ContextServiceV2ExternalDep = Annotated[ContextService, Depends(get_context_service_v2_external)]
 
 
+# --- Graph Intelligence Service ---
+
+
+async def get_graph_intelligence_service_v2_external() -> GraphIntelligenceService:
+    """Create GraphIntelligenceService for v2 API (uses external_id routing)."""
+    return GraphIntelligenceService()
+
+
+GraphIntelligenceServiceV2ExternalDep = Annotated[
+    GraphIntelligenceService, Depends(get_graph_intelligence_service_v2_external)
+]
+
+
+# --- FCM Service ---
+
+
+async def get_fcm_service_v2_external() -> FCMService:
+    """Create FCMService for v2 API (uses external_id routing)."""
+    return FCMService()
+
+
+FCMServiceV2ExternalDep = Annotated[FCMService, Depends(get_fcm_service_v2_external)]
+
+
 # --- Sync Service ---
 
 
@@ -535,6 +561,21 @@ async def get_task_scheduler(
     async def _reindex_project(**_: Any) -> None:
         await search_service.reindex_all()
 
+    async def _sync_graph_entity(entity_id: int, **extra_payload: Any) -> None:
+        # Trigger: graph-entity sync task is scheduled from graph lifecycle hooks.
+        # Why: keep scheduler contract stable while graph index provider work lands in later phases.
+        # Outcome: no-op in phase 1; task name remains valid for API and tool contracts.
+        del entity_id, extra_payload
+
+    async def _sync_graph_project(force_full: bool = False, **_: Any) -> None:
+        await _sync_project(force_full=force_full)
+
+    async def _reindex_graph_project(**_: Any) -> None:
+        # Trigger: graph reindex requested.
+        # Why: phase 1 has no dedicated graph index worker yet.
+        # Outcome: run project sync path so writes stay coherent while graph provider ships.
+        await _sync_project(force_full=True)
+
     scheduler = LocalTaskScheduler(
         {
             "reindex_entity": _reindex_entity,
@@ -542,6 +583,9 @@ async def get_task_scheduler(
             "sync_entity_vectors": _sync_entity_vectors,
             "sync_project": _sync_project,
             "reindex_project": _reindex_project,
+            "sync_graph_entity": _sync_graph_entity,
+            "sync_graph_project": _sync_graph_project,
+            "reindex_graph_project": _reindex_graph_project,
         },
         test_mode=app_config.is_test_env,
     )

src/basic_memory/mcp/clients/__init__.py

@@ -18,6 +18,8 @@ from basic_memory.mcp.clients.directory import DirectoryClient
 from basic_memory.mcp.clients.resource import ResourceClient
 from basic_memory.mcp.clients.project import ProjectClient
 from basic_memory.mcp.clients.schema import SchemaClient
+from basic_memory.mcp.clients.graph import GraphClient
+from basic_memory.mcp.clients.fcm import FCMClient
 
 __all__ = [
     "KnowledgeClient",
@@ -27,4 +29,6 @@ __all__ = [
     "ResourceClient",
     "ProjectClient",
     "SchemaClient",
+    "GraphClient",
+    "FCMClient",
 ]

src/basic_memory/mcp/clients/fcm.py

@@ -0,0 +1,56 @@
+"""Typed client for FCM API operations."""
+
+from httpx import AsyncClient
+
+from basic_memory.mcp.tools.utils import call_post
+from basic_memory.schemas.graph_intelligence import (
+    FCMExportRequest,
+    FCMExportResponse,
+    FCMImportRequest,
+    FCMImportResponse,
+    FCMRankActionsRequest,
+    FCMRankActionsResponse,
+    FCMSimulateRequest,
+    FCMSimulateResponse,
+)
+
+
+class FCMClient:
+    """Typed client for FCM operations."""
+
+    def __init__(self, http_client: AsyncClient, project_id: str):
+        self.http_client = http_client
+        self.project_id = project_id
+        self._base_path = f"/v2/projects/{project_id}/fcm"
+
+    async def simulate(self, request: FCMSimulateRequest) -> FCMSimulateResponse:
+        response = await call_post(
+            self.http_client,
+            f"{self._base_path}/simulate",
+            json=request.model_dump(mode="json"),
+        )
+        return FCMSimulateResponse.model_validate(response.json())
+
+    async def rank_actions(self, request: FCMRankActionsRequest) -> FCMRankActionsResponse:
+        response = await call_post(
+            self.http_client,
+            f"{self._base_path}/rank-actions",
+            json=request.model_dump(mode="json"),
+        )
+        return FCMRankActionsResponse.model_validate(response.json())
+
+    async def import_model(self, request: FCMImportRequest) -> FCMImportResponse:
+        response = await call_post(
+            self.http_client,
+            f"{self._base_path}/import",
+            json=request.model_dump(mode="json"),
+        )
+        return FCMImportResponse.model_validate(response.json())
+
+    async def export_model(self, request: FCMExportRequest) -> FCMExportResponse:
+        response = await call_post(
+            self.http_client,
+            f"{self._base_path}/export",
+            json=request.model_dump(mode="json"),
+        )
+        return FCMExportResponse.model_validate(response.json())

src/basic_memory/mcp/clients/graph.py

@@ -0,0 +1,62 @@
+"""Typed client for graph intelligence API operations."""
+
+from httpx import AsyncClient
+
+from basic_memory.mcp.tools.utils import call_get, call_post
+from basic_memory.schemas.graph_intelligence import (
+    GraphHealthResponse,
+    GraphImpactRequest,
+    GraphImpactResponse,
+    GraphLineageRequest,
+    GraphLineageResponse,
+    GraphReindexRequest,
+    GraphReindexResponse,
+)
+
+
+class GraphClient:
+    """Typed client for graph intelligence operations."""
+
+    def __init__(self, http_client: AsyncClient, project_id: str):
+        self.http_client = http_client
+        self.project_id = project_id
+        self._base_path = f"/v2/projects/{project_id}/graph"
+
+    async def lineage(self, request: GraphLineageRequest) -> GraphLineageResponse:
+        response = await call_post(
+            self.http_client,
+            f"{self._base_path}/lineage",
+            json=request.model_dump(mode="json"),
+        )
+        return GraphLineageResponse.model_validate(response.json())
+
+    async def impact(self, request: GraphImpactRequest) -> GraphImpactResponse:
+        response = await call_post(
+            self.http_client,
+            f"{self._base_path}/impact",
+            json=request.model_dump(mode="json"),
+        )
+        return GraphImpactResponse.model_validate(response.json())
+
+    async def health(
+        self, scope: str | None = None, timeframe: str | None = None
+    ) -> GraphHealthResponse:
+        params: dict[str, str] = {}
+        if scope is not None:
+            params["scope"] = scope
+        if timeframe is not None:
+            params["timeframe"] = timeframe
+        response = await call_get(
+            self.http_client,
+            f"{self._base_path}/health",
+            params=params,
+        )
+        return GraphHealthResponse.model_validate(response.json())
+
+    async def reindex(self, request: GraphReindexRequest) -> GraphReindexResponse:
+        response = await call_post(
+            self.http_client,
+            f"{self._base_path}/reindex",
+            json=request.model_dump(mode="json"),
+        )
+        return GraphReindexResponse.model_validate(response.json())

src/basic_memory/mcp/server.py

@@ -2,71 +2,18 @@
 Basic Memory FastMCP server.
 """
 
-import asyncio
 import time
 from contextlib import asynccontextmanager
 
 from fastmcp import FastMCP
 from loguru import logger
-from sqlalchemy import text
-from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
 
 from basic_memory import db
 from basic_memory.cli.auth import CLIAuth
-from basic_memory.config import BasicMemoryConfig
-from basic_memory.db import (
-    scoped_session,
-    _needs_semantic_embedding_backfill,
-    _run_semantic_embedding_backfill,
-)
 from basic_memory.mcp.container import McpContainer, set_container
 from basic_memory.services.initialization import initialize_app
 
 
-async def _log_embedding_status(session_maker: async_sessionmaker[AsyncSession]) -> None:
-    """Log a clear summary of semantic embedding status at startup."""
-    try:
-        async with scoped_session(session_maker) as session:
-            entity_count = (
-                await session.execute(text("SELECT COUNT(*) FROM entity"))
-            ).scalar() or 0
-            chunk_count = (
-                await session.execute(text("SELECT COUNT(*) FROM search_vector_chunks"))
-            ).scalar() or 0
-            embedding_count = (
-                await session.execute(text("SELECT COUNT(*) FROM search_vector_embeddings_rowids"))
-            ).scalar() or 0
-
-        if entity_count == 0:
-            logger.info("Semantic embeddings: no entities yet")
-        elif embedding_count == 0:
-            logger.warning(
-                f"Semantic embeddings: EMPTY — {entity_count} entities have no embeddings. "
-                "Backfill running in background..."
-            )
-        else:
-            logger.info(
-                f"Semantic embeddings: {embedding_count} embeddings "
-                f"across {chunk_count} chunks for {entity_count} entities"
-            )
-    except Exception as exc:
-        logger.debug(f"Could not check embedding status at startup: {exc}")
-
-
-async def _background_embedding_backfill(
-    config: BasicMemoryConfig,
-    session_maker: async_sessionmaker[AsyncSession],
-) -> None:
-    """Run semantic embedding backfill in the background without blocking startup."""
-    try:
-        if await _needs_semantic_embedding_backfill(config, session_maker):
-            logger.info("Background embedding backfill starting...")
-            await _run_semantic_embedding_backfill(config, session_maker)
-            await _log_embedding_status(session_maker)
-    except Exception as exc:
-        logger.error(f"Background embedding backfill failed: {exc}")
-
-
 @asynccontextmanager
 async def lifespan(app: FastMCP):
     """Lifecycle manager for the MCP server.
@@ -123,16 +70,6 @@ async def lifespan(app: FastMCP):
     # Initialize app (runs migrations, reconciles projects)
     await initialize_app(container.config)
 
-    # Log embedding status so it's easy to spot in the logs
-    backfill_task: asyncio.Task | None = None  # type: ignore[type-arg]
-    if config.semantic_search_enabled and db._session_maker is not None:
-        await _log_embedding_status(db._session_maker)
-        # Launch backfill in background so MCP server is ready immediately
-        backfill_task = asyncio.create_task(
-            _background_embedding_backfill(config, db._session_maker),
-            name="embedding-backfill",
-        )
-
     # Create and start sync coordinator (lifecycle centralized in coordinator)
     sync_coordinator = container.create_sync_coordinator()
     await sync_coordinator.start()
@@ -142,15 +79,6 @@ async def lifespan(app: FastMCP):
     finally:
         # Shutdown - coordinator handles clean task cancellation
         logger.debug("Shutting down Basic Memory MCP server")
-
-        # Cancel embedding backfill if still running
-        if backfill_task is not None and not backfill_task.done():
-            backfill_task.cancel()
-            try:
-                await backfill_task
-            except asyncio.CancelledError:
-                logger.info("Background embedding backfill cancelled during shutdown")
-
         await sync_coordinator.stop()
 
         # Only shutdown DB if we created it (not if test fixture provided it)

src/basic_memory/mcp/tools/__init__.py

@@ -24,6 +24,16 @@ from basic_memory.mcp.tools.list_directory import list_directory
 from basic_memory.mcp.tools.edit_note import edit_note
 from basic_memory.mcp.tools.move_note import move_note
 from basic_memory.mcp.tools.workspaces import list_workspaces
+from basic_memory.mcp.tools.graph_intelligence import (
+    graph_lineage,
+    graph_impact,
+    graph_health,
+    graph_reindex,
+    fcm_simulate,
+    fcm_rank_actions,
+    fcm_import_model,
+    fcm_export_model,
+)
 from basic_memory.mcp.tools.project_management import (
     list_memory_projects,
     create_memory_project,
@@ -44,7 +54,15 @@ __all__ = [
     "delete_note",
     "delete_project",
     "edit_note",
+    "fcm_export_model",
+    "fcm_import_model",
+    "fcm_rank_actions",
+    "fcm_simulate",
     "fetch",
+    "graph_health",
+    "graph_impact",
+    "graph_lineage",
+    "graph_reindex",
     "list_directory",
     "list_memory_projects",
     "list_workspaces",

src/basic_memory/mcp/tools/graph_intelligence.py

@@ -0,0 +1,271 @@
+"""MCP tools for graph intelligence and FCM contracts."""
+
+from typing import Any, Literal
+
+from fastmcp import Context
+
+from basic_memory.mcp.project_context import get_project_client
+from basic_memory.mcp.server import mcp
+from basic_memory.schemas.graph_intelligence import (
+    FCMExportRequest,
+    FCMImportRequest,
+    FCMRankActionsRequest,
+    FCMSimulateRequest,
+    GraphImpactRequest,
+    GraphLineageRequest,
+    GraphReindexRequest,
+)
+
+
+def _format_lineage_text(result: dict[str, Any]) -> str:
+    root = result["root"]["title"]
+    path_count = len(result.get("paths", []))
+    return f"# Graph Lineage\n\nRoot: {root}\nPaths: {path_count}"
+
+
+def _format_impact_text(result: dict[str, Any]) -> str:
+    target = result["target"]["title"]
+    affected = len(result.get("affected", []))
+    return f"# Graph Impact\n\nTarget: {target}\nAffected: {affected}"
+
+
+def _format_health_text(result: dict[str, Any]) -> str:
+    metrics = result["metrics"]
+    return (
+        "# Graph Health\n\n"
+        f"- orphan_rate: {metrics['orphan_rate']}\n"
+        f"- stale_central_nodes: {metrics['stale_central_nodes']}\n"
+        f"- overloaded_hubs: {metrics['overloaded_hubs']}\n"
+        f"- contradiction_candidates: {metrics['contradiction_candidates']}"
+    )
+
+
+def _format_fcm_simulate_text(result: dict[str, Any]) -> str:
+    deltas = len(result.get("deltas", []))
+    converged = result["stability"]["converged"]
+    return f"# FCM Simulation\n\nDeltas: {deltas}\nConverged: {converged}"
+
+
+def _format_fcm_rank_text(result: dict[str, Any]) -> str:
+    goal = result["goal"]["label"]
+    count = len(result.get("recommendations", []))
+    return f"# FCM Action Ranking\n\nGoal: {goal}\nRecommendations: {count}"
+
+
+@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
+async def graph_lineage(
+    start: str,
+    goal: str | None = None,
+    max_hops: int = 4,
+    relation_filters: list[str] | None = None,
+    project: str | None = None,
+    workspace: str | None = None,
+    output_format: Literal["json", "text"] = "json",
+    context: Context | None = None,
+) -> dict[str, Any] | str:
+    """Get lineage paths from a start node toward an optional goal."""
+    from basic_memory.mcp.clients import GraphClient
+
+    request = GraphLineageRequest(
+        start=start,
+        goal=goal,
+        max_hops=max_hops,
+        relation_filters=relation_filters or [],
+    )
+    async with get_project_client(project, workspace, context) as (client, active_project):
+        graph_client = GraphClient(client, active_project.external_id)
+        result = await graph_client.lineage(request)
+    payload = result.model_dump(mode="json")
+    if output_format == "text":
+        return _format_lineage_text(payload)
+    return payload
+
+
+@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
+async def graph_impact(
+    target: str,
+    horizon: int,
+    relation_filters: list[str] | None = None,
+    include_reasons: bool = True,
+    project: str | None = None,
+    workspace: str | None = None,
+    output_format: Literal["json", "text"] = "json",
+    context: Context | None = None,
+) -> dict[str, Any] | str:
+    """Get impact radius from a target node."""
+    from basic_memory.mcp.clients import GraphClient
+
+    request = GraphImpactRequest(
+        target=target,
+        horizon=horizon,
+        relation_filters=relation_filters or [],
+        include_reasons=include_reasons,
+    )
+    async with get_project_client(project, workspace, context) as (client, active_project):
+        graph_client = GraphClient(client, active_project.external_id)
+        result = await graph_client.impact(request)
+    payload = result.model_dump(mode="json")
+    if output_format == "text":
+        return _format_impact_text(payload)
+    return payload
+
+
+@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
+async def graph_health(
+    scope: str | None = None,
+    timeframe: str | None = None,
+    project: str | None = None,
+    workspace: str | None = None,
+    output_format: Literal["json", "text"] = "json",
+    context: Context | None = None,
+) -> dict[str, Any] | str:
+    """Get graph health metrics and issues."""
+    from basic_memory.mcp.clients import GraphClient
+
+    async with get_project_client(project, workspace, context) as (client, active_project):
+        graph_client = GraphClient(client, active_project.external_id)
+        result = await graph_client.health(scope=scope, timeframe=timeframe)
+    payload = result.model_dump(mode="json")
+    if output_format == "text":
+        return _format_health_text(payload)
+    return payload
+
+
+@mcp.tool(annotations={"readOnlyHint": False, "openWorldHint": False})
+async def fcm_simulate(
+    actions: list[dict[str, Any]],
+    scenario: dict[str, Any] | None = None,
+    clamp_rules: list[dict[str, Any]] | None = None,
+    project: str | None = None,
+    workspace: str | None = None,
+    output_format: Literal["json", "text"] = "json",
+    context: Context | None = None,
+) -> dict[str, Any] | str:
+    """Run an FCM simulation with optional scenario controls."""
+    from basic_memory.mcp.clients import FCMClient
+
+    request = FCMSimulateRequest.model_validate(
+        {
+            "actions": actions,
+            "scenario": scenario or {},
+            "clamp_rules": clamp_rules or [],
+        }
+    )
+    async with get_project_client(project, workspace, context) as (client, active_project):
+        fcm_client = FCMClient(client, active_project.external_id)
+        result = await fcm_client.simulate(request)
+    payload = result.model_dump(mode="json")
+    if output_format == "text":
+        return _format_fcm_simulate_text(payload)
+    return payload
+
+
+@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
+async def fcm_rank_actions(
+    goal: str,
+    constraints: dict[str, Any] | None = None,
+    top_k: int = 10,
+    project: str | None = None,
+    workspace: str | None = None,
+    output_format: Literal["json", "text"] = "json",
+    context: Context | None = None,
+) -> dict[str, Any] | str:
+    """Rank intervention actions for an FCM goal node."""
+    from basic_memory.mcp.clients import FCMClient
+
+    request = FCMRankActionsRequest.model_validate(
+        {
+            "goal": goal,
+            "constraints": constraints or {},
+            "top_k": top_k,
+        }
+    )
+    async with get_project_client(project, workspace, context) as (client, active_project):
+        fcm_client = FCMClient(client, active_project.external_id)
+        result = await fcm_client.rank_actions(request)
+    payload = result.model_dump(mode="json")
+    if output_format == "text":
+        return _format_fcm_rank_text(payload)
+    return payload
+
+
+@mcp.tool(annotations={"readOnlyHint": False, "openWorldHint": False})
+async def fcm_import_model(
+    source: str,
+    format: Literal["csv_bundle_v1"] = "csv_bundle_v1",
+    merge_mode: Literal["replace", "upsert"] = "upsert",
+    project: str | None = None,
+    workspace: str | None = None,
+    output_format: Literal["json", "text"] = "json",
+    context: Context | None = None,
+) -> dict[str, Any] | str:
+    """Import an FCM model from an external source."""
+    from basic_memory.mcp.clients import FCMClient
+
+    request = FCMImportRequest(source=source, format=format, merge_mode=merge_mode)
+    async with get_project_client(project, workspace, context) as (client, active_project):
+        fcm_client = FCMClient(client, active_project.external_id)
+        result = await fcm_client.import_model(request)
+    payload = result.model_dump(mode="json")
+    if output_format == "text":
+        return (
+            "# FCM Import\n\n"
+            f"Import ID: {payload['import_id']}\n"
+            f"Nodes Loaded: {payload['nodes_loaded']}\n"
+            f"Edges Loaded: {payload['edges_loaded']}"
+        )
+    return payload
+
+
+@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
+async def fcm_export_model(
+    format: Literal["csv_bundle_v1"] = "csv_bundle_v1",
+    selection: dict[str, Any] | None = None,
+    project: str | None = None,
+    workspace: str | None = None,
+    output_format: Literal["json", "text"] = "json",
+    context: Context | None = None,
+) -> dict[str, Any] | str:
+    """Export an FCM model selection."""
+    from basic_memory.mcp.clients import FCMClient
+
+    request = FCMExportRequest.model_validate(
+        {
+            "format": format,
+            "selection": selection or {},
+        }
+    )
+    async with get_project_client(project, workspace, context) as (client, active_project):
+        fcm_client = FCMClient(client, active_project.external_id)
+        result = await fcm_client.export_model(request)
+    payload = result.model_dump(mode="json")
+    if output_format == "text":
+        return (
+            "# FCM Export\n\n"
+            f"Export ID: {payload['export_id']}\n"
+            f"Node Count: {payload['node_count']}\n"
+            f"Edge Count: {payload['edge_count']}"
+        )
+    return payload
+
+
+@mcp.tool(annotations={"readOnlyHint": False, "openWorldHint": False})
+async def graph_reindex(
+    mode: Literal["full", "incremental"] = "incremental",
+    reason: str | None = None,
+    project: str | None = None,
+    workspace: str | None = None,
+    output_format: Literal["json", "text"] = "json",
+    context: Context | None = None,
+) -> dict[str, Any] | str:
+    """Queue a graph reindex for the active project."""
+    from basic_memory.mcp.clients import GraphClient
+
+    request = GraphReindexRequest(mode=mode, reason=reason)
+    async with get_project_client(project, workspace, context) as (client, active_project):
+        graph_client = GraphClient(client, active_project.external_id)
+        result = await graph_client.reindex(request)
+    payload = result.model_dump(mode="json")
+    if output_format == "text":
+        return f"# Graph Reindex\n\nJob ID: {payload['job_id']}\nStatus: {payload['status']}"
+    return payload

src/basic_memory/mcp/tools/schema.py

@@ -160,7 +160,7 @@ def _no_notes_guidance(note_type: str, tool_name: str) -> str:
         f"## Next Steps\n\n"
         f"1. **Create notes of this type** — use `write_note` with "
         f'`note_type="{note_type}"` to create notes\n'
-        f"2. **Check existing types** — use `search_notes` with `note_types` "
+        f"2. **Check existing types** — use `search_notes` with `entity_types` "
         f"filter to see what types exist\n"
         f"3. **Browse content** — use `list_directory` or `recent_activity` to "
         f"see what's in the project\n"
@@ -397,7 +397,7 @@ async def schema_infer(
                     f"share a consistent structure.\n\n"
                     f"## Suggestions\n"
                     f"1. **Use a more specific type** — try `search_notes` with "
-                    f"`note_types` filter to see what types exist\n"
+                    f"`entity_types` filter to see what types exist\n"
                     f"2. **Lower the threshold** — "
                     f'`schema_infer("{note_type}", threshold=0.1)` to include '
                     f"rarer fields\n"

src/basic_memory/mcp/tools/search.py

@@ -2,7 +2,7 @@
 
 import re
 from textwrap import dedent
-from typing import Annotated, List, Optional, Dict, Any, Literal
+from typing import List, Optional, Dict, Any, Literal
 
 from loguru import logger
 from fastmcp import Context
@@ -165,7 +165,7 @@ def _format_search_error_response(
                - Remove restrictive terms: Focus on the most important keywords
 
             5. **Use filtering to narrow scope**:
-               - By note type in frontmatter: `search_notes("{project}","{query}", note_types=["note"])`
+               - By content type: `search_notes("{project}","{query}", note_types=["note"])`
                - By recent content: `search_notes("{project}","{query}", after_date="1 week")`
                - By entity type: `search_notes("{project}","{query}", entity_types=["observation"])`
 
@@ -305,17 +305,8 @@ async def search_notes(
     page_size: int = 10,
     search_type: str | None = None,
     output_format: Literal["text", "json"] = "text",
-    note_types: Annotated[
-        List[str] | None,
-        "Filter by the 'type' field in note frontmatter (e.g. 'note', 'chapter', 'person'). "
-        "Case-insensitive.",
-    ] = None,
-    entity_types: Annotated[
-        List[str] | None,
-        "Filter by knowledge graph item type: 'entity' (whole notes), 'observation', or "
-        "'relation'. Defaults to 'entity'. Do NOT pass schema/frontmatter types like "
-        "'Chapter' here — use note_types instead.",
-    ] = None,
+    note_types: List[str] | None = None,
+    entity_types: List[str] | None = None,
     after_date: Optional[str] = None,
     metadata_filters: Optional[Dict[str, Any]] = None,
     tags: Optional[List[str]] = None,
@@ -359,7 +350,6 @@ async def search_notes(
     ### Search Type Examples
     - `search_notes("my-project", "Meeting", search_type="title")` - Search only in titles
     - `search_notes("work-docs", "docs/meeting-*", search_type="permalink")` - Pattern match permalinks
-      Note: Permalink patterns match the full path (e.g., "project/folder/chapter-13*", not just "chapter-13*").
     - `search_notes("research", "keyword")` - Default search (hybrid when semantic is enabled,
       text when disabled)
 
@@ -446,7 +436,7 @@ async def search_notes(
         # Exact phrase search
         results = await search_notes("\"weekly standup meeting\"")
 
-        # Search with note type filter - type property in frontmatter
+        # Search with note type filter
         results = await search_notes(
             "meeting notes",
             note_types=["note"],
@@ -487,8 +477,7 @@ async def search_notes(
         results = await search_notes("project planning", project="my-project")
     """
     # Avoid mutable-default-argument footguns. Treat None as "no filter".
-    # Lowercase note_types so "Chapter" matches the stored "chapter".
-    note_types = [t.lower() for t in note_types] if note_types else []
+    note_types = note_types or []
     entity_types = entity_types or []
 
     # Parse tag:<value> shorthand at tool level so it works with all search modes.

src/basic_memory/mcp/tools/ui_sdk.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Annotated, Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional
 
 from fastmcp import Context
 from mcp.types import ContentBlock, TextContent
@@ -28,17 +28,8 @@ async def search_notes_ui(
     page: int = 1,
     page_size: int = 10,
     search_type: Optional[str] = None,
-    note_types: Annotated[
-        List[str] | None,
-        "Filter by the 'type' field in note frontmatter (e.g. 'note', 'chapter', 'person'). "
-        "Case-insensitive.",
-    ] = None,
-    entity_types: Annotated[
-        List[str] | None,
-        "Filter by knowledge graph item type: 'entity' (whole notes), 'observation', or "
-        "'relation'. Defaults to 'entity'. Do NOT pass schema/frontmatter types like "
-        "'Chapter' here — use note_types instead.",
-    ] = None,
+    note_types: List[str] | None = None,
+    entity_types: List[str] | None = None,
     after_date: Optional[str] = None,
     metadata_filters: Optional[Dict[str, Any]] = None,
     tags: Optional[List[str]] = None,

src/basic_memory/schemas/graph_intelligence.py

@@ -0,0 +1,318 @@
+"""Schemas for Local+ graph intelligence and FCM contracts."""
+
+from datetime import datetime
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+
+# --- Graph contracts ---
+
+
+class GraphLineageRequest(BaseModel):
+    """Request contract for graph lineage queries."""
+
+    start: str
+    goal: str | None = None
+    max_hops: int = Field(default=4, ge=1, le=6)
+    relation_filters: list[str] = Field(default_factory=list)
+
+
+class GraphNodeRef(BaseModel):
+    """Minimal graph node descriptor."""
+
+    id: str
+    title: str
+    permalink: str | None = None
+
+
+class GraphPathEdge(BaseModel):
+    """Edge descriptor for lineage paths."""
+
+    relation: str
+    direction: Literal["outgoing", "incoming"]
+
+
+class GraphLineagePath(BaseModel):
+    """Single lineage path with scores and provenance."""
+
+    path_id: str
+    nodes: list[GraphNodeRef] = Field(default_factory=list)
+    edges: list[GraphPathEdge] = Field(default_factory=list)
+    deterministic_path_score: float
+    confidence: float
+    evidence_refs: list[str] = Field(default_factory=list)
+
+
+class GraphLineageResponse(BaseModel):
+    """Response contract for graph lineage queries."""
+
+    root: GraphNodeRef
+    paths: list[GraphLineagePath] = Field(default_factory=list)
+    generated_at: datetime
+
+
+class GraphImpactRequest(BaseModel):
+    """Request contract for impact-radius queries."""
+
+    target: str
+    horizon: int = Field(ge=1, le=4)
+    relation_filters: list[str] = Field(default_factory=list)
+    include_reasons: bool = True
+
+
+class GraphImpactTarget(BaseModel):
+    """Impact response target descriptor."""
+
+    id: str
+    title: str
+
+
+class GraphImpactItem(BaseModel):
+    """Affected node entry for impact responses."""
+
+    id: str
+    title: str
+    distance: int
+    impact_score: float
+    confidence: float
+    reasons: list[str] = Field(default_factory=list)
+    evidence_refs: list[str] = Field(default_factory=list)
+
+
+class GraphImpactSummary(BaseModel):
+    """Summary counters for impact responses."""
+
+    total_considered: int
+    total_returned: int
+
+
+class GraphImpactResponse(BaseModel):
+    """Response contract for impact-radius queries."""
+
+    target: GraphImpactTarget
+    affected: list[GraphImpactItem] = Field(default_factory=list)
+    summary: GraphImpactSummary
+
+
+class GraphHealthMetrics(BaseModel):
+    """Top-level graph health metrics."""
+
+    orphan_rate: float
+    stale_central_nodes: int
+    overloaded_hubs: int
+    contradiction_candidates: int
+
+
+class GraphHealthIssue(BaseModel):
+    """Actionable graph-health issue entry."""
+
+    issue_type: Literal[
+        "orphan",
+        "stale_central",
+        "overloaded_hub",
+        "contradiction_candidate",
+    ]
+    entity_id: str
+    severity: Literal["low", "medium", "high"]
+    reason: str
+    suggested_action: str
+    confidence: float | None = None
+
+
+class GraphHealthResponse(BaseModel):
+    """Response contract for health checks."""
+
+    metrics: GraphHealthMetrics
+    issues: list[GraphHealthIssue] = Field(default_factory=list)
+    computed_at: datetime
+
+
+class GraphReindexRequest(BaseModel):
+    """Request contract for graph reindex scheduling."""
+
+    mode: Literal["full", "incremental"] = "incremental"
+    reason: str | None = None
+
+
+class GraphReindexResponse(BaseModel):
+    """Response contract for graph reindex scheduling."""
+
+    job_id: str
+    status: Literal["queued", "running", "completed", "failed"]
+    scheduled_at: datetime
+
+
+# --- FCM contracts ---
+
+
+class FCMAction(BaseModel):
+    """Action delta for simulation input."""
+
+    node_id: str
+    delta: float
+
+
+class FCMScenario(BaseModel):
+    """Simulation runtime configuration."""
+
+    steps: int = 12
+    activation: Literal["tanh", "sigmoid", "bounded_linear"] = "tanh"
+    decay: float = 0.05
+
+
+class FCMClampRule(BaseModel):
+    """Clamp bounds for selected nodes."""
+
+    node_id: str
+    min: float
+    max: float
+
+
+class FCMSimulateRequest(BaseModel):
+    """Request contract for FCM simulation."""
+
+    actions: list[FCMAction]
+    scenario: FCMScenario = Field(default_factory=FCMScenario)
+    clamp_rules: list[FCMClampRule] = Field(default_factory=list)
+
+
+class FCMNodeState(BaseModel):
+    """Node state in baseline/projected vectors."""
+
+    node_id: str
+    state: float
+
+
+class FCMNodeDelta(BaseModel):
+    """Node delta entry in simulation output."""
+
+    node_id: str
+    delta: float
+
+
+class FCMStability(BaseModel):
+    """Simulation stability metadata."""
+
+    converged: bool
+    iterations_used: int
+    residual: float
+
+
+class FCMInfluencer(BaseModel):
+    """Top influencer entry for explanation payload."""
+
+    source: str
+    weight: float
+
+
+class FCMExplanation(BaseModel):
+    """Per-node explanation payload."""
+
+    node_id: str
+    top_influencers: list[FCMInfluencer] = Field(default_factory=list)
+
+
+class FCMSimulateResponse(BaseModel):
+    """Response contract for FCM simulation."""
+
+    baseline: list[FCMNodeState] = Field(default_factory=list)
+    projected: list[FCMNodeState] = Field(default_factory=list)
+    deltas: list[FCMNodeDelta] = Field(default_factory=list)
+    stability: FCMStability
+    confidence: float
+    explanations: list[FCMExplanation] = Field(default_factory=list)
+    evidence_refs: list[str] = Field(default_factory=list)
+
+
+class FCMRankConstraints(BaseModel):
+    """Constraint set for action ranking."""
+
+    max_negative_impact: float | None = None
+    required_tags: list[str] = Field(default_factory=list)
+    disallowed_nodes: list[str] = Field(default_factory=list)
+
+
+class FCMRankActionsRequest(BaseModel):
+    """Request contract for FCM action ranking."""
+
+    goal: str
+    constraints: FCMRankConstraints = Field(default_factory=FCMRankConstraints)
+    top_k: int = Field(default=10, ge=1, le=25)
+
+
+class FCMGoalRef(BaseModel):
+    """Goal descriptor for ranking output."""
+
+    node_id: str
+    label: str
+
+
+class FCMRecommendation(BaseModel):
+    """Ranked intervention candidate."""
+
+    action_node_id: str
+    expected_goal_delta: float
+    risk_penalty: float
+    net_score: float
+    confidence: float
+    rationale: list[str] = Field(default_factory=list)
+    evidence_refs: list[str] = Field(default_factory=list)
+
+
+class FCMRankActionsResponse(BaseModel):
+    """Response contract for action ranking."""
+
+    goal: FCMGoalRef
+    recommendations: list[FCMRecommendation] = Field(default_factory=list)
+
+
+class FCMImportRequest(BaseModel):
+    """Request contract for model import."""
+
+    source: str
+    format: Literal["csv_bundle_v1"] = "csv_bundle_v1"
+    merge_mode: Literal["replace", "upsert"] = "upsert"
+
+
+class FCMImportResponse(BaseModel):
+    """Response contract for model import."""
+
+    import_id: str
+    nodes_loaded: int
+    edges_loaded: int
+    warnings: list[str] = Field(default_factory=list)
+    errors: list[str] = Field(default_factory=list)
+
+
+class FCMExportSelection(BaseModel):
+    """Scope selection for model export."""
+
+    scope: Literal["all", "tag", "subgraph"] = "all"
+    tag: str | None = None
+    seed_nodes: list[str] = Field(default_factory=list)
+
+
+class FCMExportRequest(BaseModel):
+    """Request contract for model export."""
+
+    format: Literal["csv_bundle_v1"] = "csv_bundle_v1"
+    selection: FCMExportSelection = Field(default_factory=FCMExportSelection)
+
+
+class FCMExportFile(BaseModel):
+    """Single file descriptor in an export response."""
+
+    name: str
+    path: str
+
+
+class FCMExportResponse(BaseModel):
+    """Response contract for model export."""
+
+    export_id: str
+    format: Literal["csv_bundle_v1"]
+    files: list[FCMExportFile] = Field(default_factory=list)
+    node_count: int
+    edge_count: int
+    metadata: dict[str, Any] | None = None

src/basic_memory/services/fcm_service.py

@@ -0,0 +1,96 @@
+"""Service layer for FCM contract endpoints."""
+
+from uuid import uuid4
+
+from basic_memory.schemas.graph_intelligence import (
+    FCMExportFile,
+    FCMExportRequest,
+    FCMExportResponse,
+    FCMGoalRef,
+    FCMImportRequest,
+    FCMImportResponse,
+    FCMNodeDelta,
+    FCMNodeState,
+    FCMRankActionsRequest,
+    FCMRankActionsResponse,
+    FCMRecommendation,
+    FCMSimulateRequest,
+    FCMSimulateResponse,
+    FCMStability,
+)
+
+
+class FCMService:
+    """FCM contract service.
+
+    Phase 1 keeps deterministic behavior so API and tool surfaces stabilize
+    before introducing advanced simulation engines.
+    """
+
+    async def simulate(self, request: FCMSimulateRequest) -> FCMSimulateResponse:
+        """Return deterministic baseline/projected state vectors."""
+        baseline = [FCMNodeState(node_id=action.node_id, state=0.0) for action in request.actions]
+        projected = [
+            FCMNodeState(node_id=action.node_id, state=action.delta) for action in request.actions
+        ]
+        deltas = [
+            FCMNodeDelta(node_id=action.node_id, delta=action.delta) for action in request.actions
+        ]
+        return FCMSimulateResponse(
+            baseline=baseline,
+            projected=projected,
+            deltas=deltas,
+            stability=FCMStability(
+                converged=True,
+                iterations_used=min(request.scenario.steps, 5),
+                residual=0.0,
+            ),
+            confidence=0.5,
+            explanations=[],
+            evidence_refs=[],
+        )
+
+    async def rank_actions(self, request: FCMRankActionsRequest) -> FCMRankActionsResponse:
+        """Return deterministic ranked actions for a target goal."""
+        recommendations = [
+            FCMRecommendation(
+                action_node_id=f"{request.goal}:action:{idx + 1}",
+                expected_goal_delta=0.25 - (idx * 0.01),
+                risk_penalty=0.05 + (idx * 0.005),
+                net_score=0.20 - (idx * 0.015),
+                confidence=0.5,
+                rationale=["Contract skeleton recommendation"],
+                evidence_refs=[],
+            )
+            for idx in range(min(request.top_k, 3))
+        ]
+        return FCMRankActionsResponse(
+            goal=FCMGoalRef(node_id=request.goal, label=request.goal),
+            recommendations=recommendations,
+        )
+
+    async def import_model(self, request: FCMImportRequest) -> FCMImportResponse:
+        """Return deterministic import metadata."""
+        _ = request
+        return FCMImportResponse(
+            import_id=str(uuid4()),
+            nodes_loaded=0,
+            edges_loaded=0,
+            warnings=[],
+            errors=[],
+        )
+
+    async def export_model(self, request: FCMExportRequest) -> FCMExportResponse:
+        """Return deterministic export metadata and file descriptors."""
+        scope = request.selection.scope
+        return FCMExportResponse(
+            export_id=str(uuid4()),
+            format=request.format,
+            files=[
+                FCMExportFile(name="nodes.csv", path=f"/tmp/{scope}-nodes.csv"),
+                FCMExportFile(name="edges.csv", path=f"/tmp/{scope}-edges.csv"),
+            ],
+            node_count=0,
+            edge_count=0,
+            metadata={"scope": scope},
+        )

src/basic_memory/services/graph_intelligence_service.py

@@ -0,0 +1,122 @@
+"""Service layer for graph intelligence contract endpoints."""
+
+from datetime import datetime, timezone
+from uuid import uuid4
+
+from basic_memory.schemas.graph_intelligence import (
+    GraphHealthMetrics,
+    GraphHealthResponse,
+    GraphImpactItem,
+    GraphImpactRequest,
+    GraphImpactResponse,
+    GraphImpactSummary,
+    GraphImpactTarget,
+    GraphLineagePath,
+    GraphLineageRequest,
+    GraphLineageResponse,
+    GraphNodeRef,
+    GraphPathEdge,
+    GraphReindexResponse,
+)
+
+
+def _normalize_memory_ref(value: str) -> str:
+    """Normalize user input into a memory:// reference string."""
+    if value.startswith("memory://"):
+        return value
+    return f"memory://{value}"
+
+
+def _normalize_node_id(value: str) -> str:
+    """Return a stable node id for contract skeleton outputs."""
+    return value.removeprefix("memory://")
+
+
+class GraphIntelligenceService:
+    """Graph intelligence contract service.
+
+    Phase 1 behavior is intentionally deterministic and lightweight so routing,
+    clients, and contract tests can ship before deeper traversal engines.
+    """
+
+    async def lineage(self, request: GraphLineageRequest) -> GraphLineageResponse:
+        """Return a deterministic lineage payload for the requested root/goal."""
+        root_ref = _normalize_memory_ref(request.start)
+        root = GraphNodeRef(
+            id=_normalize_node_id(root_ref),
+            title=_normalize_node_id(root_ref),
+            permalink=_normalize_node_id(root_ref),
+        )
+
+        nodes = [root]
+        edges: list[GraphPathEdge] = []
+        if request.goal:
+            goal_ref = _normalize_memory_ref(request.goal)
+            nodes.append(
+                GraphNodeRef(
+                    id=_normalize_node_id(goal_ref),
+                    title=_normalize_node_id(goal_ref),
+                    permalink=_normalize_node_id(goal_ref),
+                )
+            )
+            edges.append(GraphPathEdge(relation="related_to", direction="outgoing"))
+
+        path = GraphLineagePath(
+            path_id=f"path-{uuid4()}",
+            nodes=nodes,
+            edges=edges,
+            deterministic_path_score=1.0 if request.goal else 0.5,
+            confidence=0.5,
+            evidence_refs=[root_ref],
+        )
+
+        return GraphLineageResponse(
+            root=root,
+            paths=[path],
+            generated_at=datetime.now(timezone.utc),
+        )
+
+    async def impact(self, request: GraphImpactRequest) -> GraphImpactResponse:
+        """Return a deterministic impact preview payload."""
+        target_id = _normalize_node_id(_normalize_memory_ref(request.target))
+        affected = [
+            GraphImpactItem(
+                id=f"{target_id}:neighbor:1",
+                title=f"{target_id} dependent",
+                distance=min(request.horizon, 1),
+                impact_score=0.55,
+                confidence=0.5,
+                reasons=["Connected via typed relation in contract skeleton"],
+                evidence_refs=[_normalize_memory_ref(request.target)],
+            )
+        ]
+        if not request.include_reasons:
+            affected[0].reasons = []
+
+        return GraphImpactResponse(
+            target=GraphImpactTarget(id=target_id, title=target_id),
+            affected=affected,
+            summary=GraphImpactSummary(total_considered=1, total_returned=1),
+        )
+
+    async def health(self, scope: str | None, timeframe: str | None) -> GraphHealthResponse:
+        """Return deterministic baseline health metrics."""
+        _ = (scope, timeframe)
+        return GraphHealthResponse(
+            metrics=GraphHealthMetrics(
+                orphan_rate=0.0,
+                stale_central_nodes=0,
+                overloaded_hubs=0,
+                contradiction_candidates=0,
+            ),
+            issues=[],
+            computed_at=datetime.now(timezone.utc),
+        )
+
+    async def start_reindex_job(self) -> GraphReindexResponse:
+        """Create reindex job metadata for queued responses."""
+        return GraphReindexResponse(
+            job_id=str(uuid4()),
+            status="queued",
+            scheduled_at=datetime.now(timezone.utc),
+        )

src/basic_memory/sync/sync_service.py

@@ -293,16 +293,12 @@ class SyncService:
         for path in report.deleted:
             await self.handle_delete(path)
 
-        # then new and modified — collect entity IDs for batch vector embedding
-        synced_entity_ids: list[int] = []
-
+        # then new and modified
         for path in report.new:
             entity, _ = await self.sync_file(path, new=True)
 
-            if entity is not None:
-                synced_entity_ids.append(entity.id)
             # Track if file was skipped
-            elif await self._should_skip_file(path):
+            if entity is None and await self._should_skip_file(path):
                 failure_info = self._file_failures[path]
                 report.skipped_files.append(
                     SkippedFile(
@@ -316,10 +312,8 @@ class SyncService:
         for path in report.modified:
             entity, _ = await self.sync_file(path, new=False)
 
-            if entity is not None:
-                synced_entity_ids.append(entity.id)
             # Track if file was skipped
-            elif await self._should_skip_file(path):
+            if entity is None and await self._should_skip_file(path):
                 failure_info = self._file_failures[path]
                 report.skipped_files.append(
                     SkippedFile(
@@ -337,26 +331,6 @@ class SyncService:
         else:
             logger.info("Skipping relation resolution - no file changes detected")
 
-        # Batch-generate vector embeddings for all synced entities
-        if synced_entity_ids and self.app_config.semantic_search_enabled:
-            try:
-                logger.info(
-                    f"Generating semantic embeddings for {len(synced_entity_ids)} entities..."
-                )
-                batch_result = await self.search_service.sync_entity_vectors_batch(
-                    synced_entity_ids
-                )
-                logger.info(
-                    f"Semantic embeddings complete: "
-                    f"synced={batch_result.entities_synced}, "
-                    f"failed={batch_result.entities_failed}"
-                )
-            except SemanticDependenciesMissingError:
-                logger.warning(
-                    "Semantic search dependencies missing — vector embeddings skipped. "
-                    "Run 'bm reindex --embeddings' after resolving the dependency issue."
-                )
-
         # Update scan watermark after successful sync
         # Use the timestamp from sync start (not end) to ensure we catch files
         # created during the sync on the next iteration

tests/api/v2/test_graph_intelligence_router.py

@@ -0,0 +1,120 @@
+"""Tests for v2 graph intelligence and FCM routers."""
+
+import pytest
+from httpx import AsyncClient
+
+
+@pytest.mark.asyncio
+async def test_graph_lineage_contract(client: AsyncClient, v2_project_url: str):
+    response = await client.post(
+        f"{v2_project_url}/graph/lineage",
+        json={"start": "memory://specs/search"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert set(["root", "paths", "generated_at"]).issubset(data.keys())
+    assert data["root"]["id"] == "specs/search"
+    assert isinstance(data["paths"], list)
+
+
+@pytest.mark.asyncio
+async def test_graph_impact_contract(client: AsyncClient, v2_project_url: str):
+    response = await client.post(
+        f"{v2_project_url}/graph/impact",
+        json={"target": "memory://specs/search", "horizon": 2},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert set(["target", "affected", "summary"]).issubset(data.keys())
+    assert data["summary"]["total_considered"] >= data["summary"]["total_returned"]
+
+
+@pytest.mark.asyncio
+async def test_graph_health_contract(client: AsyncClient, v2_project_url: str):
+    response = await client.get(
+        f"{v2_project_url}/graph/health",
+        params={"scope": "specs", "timeframe": "30d"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert set(["metrics", "issues", "computed_at"]).issubset(data.keys())
+    assert "orphan_rate" in data["metrics"]
+
+
+@pytest.mark.asyncio
+async def test_graph_reindex_schedules_task(
+    client: AsyncClient,
+    v2_project_url: str,
+    task_scheduler_spy: list[dict[str, object]],
+):
+    response = await client.post(
+        f"{v2_project_url}/graph/reindex",
+        json={"mode": "full", "reason": "contract test"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["status"] == "queued"
+    assert data["job_id"]
+
+    assert task_scheduler_spy
+    last = task_scheduler_spy[-1]
+    assert last["task_name"] == "reindex_graph_project"
+    assert last["payload"]["mode"] == "full"
+    assert last["payload"]["reason"] == "contract test"
+
+
+@pytest.mark.asyncio
+async def test_fcm_simulate_contract(client: AsyncClient, v2_project_url: str):
+    response = await client.post(
+        f"{v2_project_url}/fcm/simulate",
+        json={"actions": [{"node_id": "test-node", "delta": 0.2}]},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert set(["baseline", "projected", "deltas", "stability", "confidence"]).issubset(data.keys())
+    assert data["stability"]["converged"] is True
+
+
+@pytest.mark.asyncio
+async def test_fcm_rank_actions_contract(client: AsyncClient, v2_project_url: str):
+    response = await client.post(
+        f"{v2_project_url}/fcm/rank-actions",
+        json={"goal": "reduce-regressions", "top_k": 2},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert set(["goal", "recommendations"]).issubset(data.keys())
+    assert len(data["recommendations"]) <= 2
+
+
+@pytest.mark.asyncio
+async def test_fcm_import_contract(client: AsyncClient, v2_project_url: str):
+    response = await client.post(
+        f"{v2_project_url}/fcm/import",
+        json={"source": "/tmp/model.csv", "format": "csv_bundle_v1"},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert set(["import_id", "nodes_loaded", "edges_loaded", "warnings", "errors"]).issubset(
+        data.keys()
+    )
+
+
+@pytest.mark.asyncio
+async def test_fcm_export_contract(client: AsyncClient, v2_project_url: str):
+    response = await client.post(
+        f"{v2_project_url}/fcm/export",
+        json={"format": "csv_bundle_v1", "selection": {"scope": "all"}},
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert set(["export_id", "format", "files", "node_count", "edge_count"]).issubset(data.keys())
+    assert len(data["files"]) == 2

tests/cli/test_cli_tool_graph_intelligence_json_output.py

@@ -0,0 +1,184 @@
+"""Tests for graph/FCM CLI tool JSON passthrough commands."""
+
+import json
+from unittest.mock import AsyncMock, patch
+
+from typer.testing import CliRunner
+
+from basic_memory.cli.main import app as cli_app
+
+runner = CliRunner()
+
+
+@patch(
+    "basic_memory.cli.commands.tool.mcp_graph_lineage",
+    new_callable=AsyncMock,
+    return_value={
+        "root": {"id": "specs/search"},
+        "paths": [],
+        "generated_at": "2026-03-05T00:00:00Z",
+    },
+)
+def test_graph_lineage_json_output(mock_tool):
+    result = runner.invoke(cli_app, ["tool", "graph-lineage", "memory://specs/search"])
+    assert result.exit_code == 0, f"CLI failed: {result.output}"
+    data = json.loads(result.output)
+    assert data["root"]["id"] == "specs/search"
+    assert mock_tool.call_args.kwargs["output_format"] == "json"
+
+
+@patch(
+    "basic_memory.cli.commands.tool.mcp_graph_impact",
+    new_callable=AsyncMock,
+    return_value={
+        "target": {"id": "specs/search", "title": "specs/search"},
+        "affected": [],
+        "summary": {"total_considered": 0, "total_returned": 0},
+    },
+)
+def test_graph_impact_passthrough(mock_tool):
+    result = runner.invoke(
+        cli_app,
+        [
+            "tool",
+            "graph-impact",
+            "memory://specs/search",
+            "--horizon",
+            "3",
+            "--relation-filter",
+            "depends_on",
+        ],
+    )
+    assert result.exit_code == 0, f"CLI failed: {result.output}"
+    assert mock_tool.call_args.kwargs["horizon"] == 3
+    assert mock_tool.call_args.kwargs["relation_filters"] == ["depends_on"]
+    assert mock_tool.call_args.kwargs["output_format"] == "json"
+
+
+@patch(
+    "basic_memory.cli.commands.tool.mcp_graph_health",
+    new_callable=AsyncMock,
+    return_value={
+        "metrics": {
+            "orphan_rate": 0.0,
+            "stale_central_nodes": 0,
+            "overloaded_hubs": 0,
+            "contradiction_candidates": 0,
+        },
+        "issues": [],
+        "computed_at": "2026-03-05T00:00:00Z",
+    },
+)
+def test_graph_health_json_output(mock_tool):
+    result = runner.invoke(
+        cli_app,
+        ["tool", "graph-health", "--scope", "specs", "--timeframe", "30d"],
+    )
+    assert result.exit_code == 0, f"CLI failed: {result.output}"
+    data = json.loads(result.output)
+    assert "metrics" in data
+    assert mock_tool.call_args.kwargs["scope"] == "specs"
+    assert mock_tool.call_args.kwargs["timeframe"] == "30d"
+
+
+@patch(
+    "basic_memory.cli.commands.tool.mcp_fcm_simulate",
+    new_callable=AsyncMock,
+    return_value={
+        "baseline": [],
+        "projected": [],
+        "deltas": [],
+        "stability": {"converged": True, "iterations_used": 1, "residual": 0.0},
+        "confidence": 0.5,
+    },
+)
+def test_fcm_simulate_json_output(mock_tool):
+    result = runner.invoke(
+        cli_app,
+        [
+            "tool",
+            "fcm-simulate",
+            "--actions-json",
+            '[{"node_id":"n1","delta":0.2}]',
+            "--scenario-json",
+            '{"steps":8}',
+        ],
+    )
+    assert result.exit_code == 0, f"CLI failed: {result.output}"
+    assert mock_tool.call_args.kwargs["actions"] == [{"node_id": "n1", "delta": 0.2}]
+    assert mock_tool.call_args.kwargs["scenario"] == {"steps": 8}
+
+
+def test_fcm_simulate_invalid_actions_json():
+    result = runner.invoke(
+        cli_app,
+        ["tool", "fcm-simulate", "--actions-json", '{"node_id":"n1","delta":0.2}'],
+    )
+    assert result.exit_code == 1
+    assert "expected a JSON array" in result.output
+
+
+@patch(
+    "basic_memory.cli.commands.tool.mcp_fcm_rank_actions",
+    new_callable=AsyncMock,
+    return_value={"goal": {"node_id": "g1", "label": "g1"}, "recommendations": []},
+)
+def test_fcm_rank_actions_passthrough(mock_tool):
+    result = runner.invoke(
+        cli_app,
+        ["tool", "fcm-rank-actions", "g1", "--constraints-json", '{"required_tags":["risk"]}'],
+    )
+    assert result.exit_code == 0, f"CLI failed: {result.output}"
+    assert mock_tool.call_args.kwargs["constraints"] == {"required_tags": ["risk"]}
+    assert mock_tool.call_args.kwargs["output_format"] == "json"
+
+
+@patch(
+    "basic_memory.cli.commands.tool.mcp_fcm_import_model",
+    new_callable=AsyncMock,
+    return_value={
+        "import_id": "imp-1",
+        "nodes_loaded": 0,
+        "edges_loaded": 0,
+        "warnings": [],
+        "errors": [],
+    },
+)
+def test_fcm_import_model_json_output(mock_tool):
+    result = runner.invoke(
+        cli_app,
+        ["tool", "fcm-import-model", "/tmp/model.csv", "--format", "csv_bundle_v1"],
+    )
+    assert result.exit_code == 0, f"CLI failed: {result.output}"
+    data = json.loads(result.output)
+    assert data["import_id"] == "imp-1"
+    assert mock_tool.call_args.kwargs["output_format"] == "json"
+
+
+@patch(
+    "basic_memory.cli.commands.tool.mcp_fcm_export_model",
+    new_callable=AsyncMock,
+    return_value={
+        "export_id": "exp-1",
+        "format": "csv_bundle_v1",
+        "files": [],
+        "node_count": 0,
+        "edge_count": 0,
+    },
+)
+def test_fcm_export_model_json_output(mock_tool):
+    result = runner.invoke(
+        cli_app,
+        [
+            "tool",
+            "fcm-export-model",
+            "--format",
+            "csv_bundle_v1",
+            "--selection-json",
+            '{"scope":"all"}',
+        ],
+    )
+    assert result.exit_code == 0, f"CLI failed: {result.output}"
+    data = json.loads(result.output)
+    assert data["export_id"] == "exp-1"
+    assert mock_tool.call_args.kwargs["selection"] == {"scope": "all"}

tests/mcp/clients/test_graph_clients.py

@@ -0,0 +1,216 @@
+"""Tests for graph and FCM typed clients."""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from basic_memory.mcp.clients import FCMClient, GraphClient
+
+
+class TestGraphClient:
+    def test_init(self):
+        mock_http = MagicMock()
+        client = GraphClient(mock_http, "project-123")
+        assert client.http_client is mock_http
+        assert client.project_id == "project-123"
+        assert client._base_path == "/v2/projects/project-123/graph"
+
+    @pytest.mark.asyncio
+    async def test_lineage(self, monkeypatch):
+        from basic_memory.mcp.clients import graph as graph_mod
+        from basic_memory.schemas.graph_intelligence import GraphLineageRequest
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "root": {"id": "specs/search", "title": "specs/search", "permalink": "specs/search"},
+            "paths": [],
+            "generated_at": "2026-03-05T00:00:00+00:00",
+        }
+
+        async def mock_call_post(client, url, **kwargs):
+            assert "/v2/projects/proj-123/graph/lineage" in url
+            return mock_response
+
+        monkeypatch.setattr(graph_mod, "call_post", mock_call_post)
+
+        client = GraphClient(MagicMock(), "proj-123")
+        result = await client.lineage(GraphLineageRequest(start="memory://specs/search"))
+        assert result.root.id == "specs/search"
+
+    @pytest.mark.asyncio
+    async def test_impact(self, monkeypatch):
+        from basic_memory.mcp.clients import graph as graph_mod
+        from basic_memory.schemas.graph_intelligence import GraphImpactRequest
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "target": {"id": "specs/search", "title": "specs/search"},
+            "affected": [],
+            "summary": {"total_considered": 0, "total_returned": 0},
+        }
+
+        async def mock_call_post(client, url, **kwargs):
+            assert "/v2/projects/proj-123/graph/impact" in url
+            return mock_response
+
+        monkeypatch.setattr(graph_mod, "call_post", mock_call_post)
+
+        client = GraphClient(MagicMock(), "proj-123")
+        result = await client.impact(GraphImpactRequest(target="memory://specs/search", horizon=2))
+        assert result.summary.total_returned == 0
+
+    @pytest.mark.asyncio
+    async def test_health(self, monkeypatch):
+        from basic_memory.mcp.clients import graph as graph_mod
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "metrics": {
+                "orphan_rate": 0.0,
+                "stale_central_nodes": 0,
+                "overloaded_hubs": 0,
+                "contradiction_candidates": 0,
+            },
+            "issues": [],
+            "computed_at": "2026-03-05T00:00:00+00:00",
+        }
+
+        async def mock_call_get(client, url, **kwargs):
+            assert "/v2/projects/proj-123/graph/health" in url
+            assert kwargs["params"]["scope"] == "specs"
+            return mock_response
+
+        monkeypatch.setattr(graph_mod, "call_get", mock_call_get)
+
+        client = GraphClient(MagicMock(), "proj-123")
+        result = await client.health(scope="specs", timeframe="30d")
+        assert result.metrics.orphan_rate == 0.0
+
+    @pytest.mark.asyncio
+    async def test_reindex(self, monkeypatch):
+        from basic_memory.mcp.clients import graph as graph_mod
+        from basic_memory.schemas.graph_intelligence import GraphReindexRequest
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "job_id": "job-123",
+            "status": "queued",
+            "scheduled_at": "2026-03-05T00:00:00+00:00",
+        }
+
+        async def mock_call_post(client, url, **kwargs):
+            assert "/v2/projects/proj-123/graph/reindex" in url
+            return mock_response
+
+        monkeypatch.setattr(graph_mod, "call_post", mock_call_post)
+
+        client = GraphClient(MagicMock(), "proj-123")
+        result = await client.reindex(GraphReindexRequest(mode="full"))
+        assert result.status == "queued"
+
+
+class TestFCMClient:
+    def test_init(self):
+        mock_http = MagicMock()
+        client = FCMClient(mock_http, "project-123")
+        assert client.http_client is mock_http
+        assert client.project_id == "project-123"
+        assert client._base_path == "/v2/projects/project-123/fcm"
+
+    @pytest.mark.asyncio
+    async def test_simulate(self, monkeypatch):
+        from basic_memory.mcp.clients import fcm as fcm_mod
+        from basic_memory.schemas.graph_intelligence import FCMSimulateRequest
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "baseline": [{"node_id": "n1", "state": 0.0}],
+            "projected": [{"node_id": "n1", "state": 0.2}],
+            "deltas": [{"node_id": "n1", "delta": 0.2}],
+            "stability": {"converged": True, "iterations_used": 3, "residual": 0.0},
+            "confidence": 0.5,
+            "explanations": [],
+            "evidence_refs": [],
+        }
+
+        async def mock_call_post(client, url, **kwargs):
+            assert "/v2/projects/proj-123/fcm/simulate" in url
+            return mock_response
+
+        monkeypatch.setattr(fcm_mod, "call_post", mock_call_post)
+
+        request = FCMSimulateRequest(actions=[{"node_id": "n1", "delta": 0.2}])
+        result = await FCMClient(MagicMock(), "proj-123").simulate(request)
+        assert result.stability.converged is True
+
+    @pytest.mark.asyncio
+    async def test_rank_actions(self, monkeypatch):
+        from basic_memory.mcp.clients import fcm as fcm_mod
+        from basic_memory.schemas.graph_intelligence import FCMRankActionsRequest
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "goal": {"node_id": "g1", "label": "g1"},
+            "recommendations": [],
+        }
+
+        async def mock_call_post(client, url, **kwargs):
+            assert "/v2/projects/proj-123/fcm/rank-actions" in url
+            return mock_response
+
+        monkeypatch.setattr(fcm_mod, "call_post", mock_call_post)
+
+        request = FCMRankActionsRequest(goal="g1")
+        result = await FCMClient(MagicMock(), "proj-123").rank_actions(request)
+        assert result.goal.node_id == "g1"
+
+    @pytest.mark.asyncio
+    async def test_import_model(self, monkeypatch):
+        from basic_memory.mcp.clients import fcm as fcm_mod
+        from basic_memory.schemas.graph_intelligence import FCMImportRequest
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "import_id": "imp-1",
+            "nodes_loaded": 0,
+            "edges_loaded": 0,
+            "warnings": [],
+            "errors": [],
+        }
+
+        async def mock_call_post(client, url, **kwargs):
+            assert "/v2/projects/proj-123/fcm/import" in url
+            return mock_response
+
+        monkeypatch.setattr(fcm_mod, "call_post", mock_call_post)
+
+        request = FCMImportRequest(source="/tmp/model.csv")
+        result = await FCMClient(MagicMock(), "proj-123").import_model(request)
+        assert result.import_id == "imp-1"
+
+    @pytest.mark.asyncio
+    async def test_export_model(self, monkeypatch):
+        from basic_memory.mcp.clients import fcm as fcm_mod
+        from basic_memory.schemas.graph_intelligence import FCMExportRequest
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "export_id": "exp-1",
+            "format": "csv_bundle_v1",
+            "files": [
+                {"name": "nodes.csv", "path": "/tmp/nodes.csv"},
+                {"name": "edges.csv", "path": "/tmp/edges.csv"},
+            ],
+            "node_count": 0,
+            "edge_count": 0,
+        }
+
+        async def mock_call_post(client, url, **kwargs):
+            assert "/v2/projects/proj-123/fcm/export" in url
+            return mock_response
+
+        monkeypatch.setattr(fcm_mod, "call_post", mock_call_post)
+
+        request = FCMExportRequest()
+        result = await FCMClient(MagicMock(), "proj-123").export_model(request)
+        assert result.format == "csv_bundle_v1"

tests/mcp/test_tool_contracts.py

@@ -35,7 +35,31 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
         "expected_replacements",
         "output_format",
     ],
+    "fcm_export_model": ["format", "selection", "project", "workspace", "output_format"],
+    "fcm_import_model": ["source", "format", "merge_mode", "project", "workspace", "output_format"],
+    "fcm_rank_actions": ["goal", "constraints", "top_k", "project", "workspace", "output_format"],
+    "fcm_simulate": ["actions", "scenario", "clamp_rules", "project", "workspace", "output_format"],
     "fetch": ["id"],
+    "graph_health": ["scope", "timeframe", "project", "workspace", "output_format"],
+    "graph_impact": [
+        "target",
+        "horizon",
+        "relation_filters",
+        "include_reasons",
+        "project",
+        "workspace",
+        "output_format",
+    ],
+    "graph_lineage": [
+        "start",
+        "goal",
+        "max_hops",
+        "relation_filters",
+        "project",
+        "workspace",
+        "output_format",
+    ],
+    "graph_reindex": ["mode", "reason", "project", "workspace", "output_format"],
     "list_directory": ["dir_name", "depth", "file_name_glob", "project", "workspace"],
     "list_memory_projects": ["output_format", "workspace"],
     "list_workspaces": ["output_format"],
@@ -113,7 +137,15 @@ TOOL_FUNCTIONS: dict[str, object] = {
     "delete_note": tools.delete_note,
     "delete_project": tools.delete_project,
     "edit_note": tools.edit_note,
+    "fcm_export_model": tools.fcm_export_model,
+    "fcm_import_model": tools.fcm_import_model,
+    "fcm_rank_actions": tools.fcm_rank_actions,
+    "fcm_simulate": tools.fcm_simulate,
     "fetch": tools.fetch,
+    "graph_health": tools.graph_health,
+    "graph_impact": tools.graph_impact,
+    "graph_lineage": tools.graph_lineage,
+    "graph_reindex": tools.graph_reindex,
     "list_directory": tools.list_directory,
     "list_memory_projects": tools.list_memory_projects,
     "list_workspaces": tools.list_workspaces,

tests/mcp/test_tool_graph_intelligence.py

@@ -0,0 +1,114 @@
+"""Tests for graph intelligence MCP tools."""
+
+import pytest
+
+from basic_memory.mcp.tools import (
+    fcm_export_model,
+    fcm_import_model,
+    fcm_rank_actions,
+    fcm_simulate,
+    graph_health,
+    graph_impact,
+    graph_lineage,
+    graph_reindex,
+)
+
+
+@pytest.mark.asyncio
+async def test_graph_lineage_json_and_text_modes(app, test_project):
+    json_result = await graph_lineage(
+        start="memory://specs/search",
+        project=test_project.name,
+        output_format="json",
+    )
+    assert isinstance(json_result, dict)
+    assert set(["root", "paths", "generated_at"]).issubset(json_result.keys())
+
+    text_result = await graph_lineage(
+        start="memory://specs/search",
+        project=test_project.name,
+        output_format="text",
+    )
+    assert isinstance(text_result, str)
+    assert "Graph Lineage" in text_result
+
+
+@pytest.mark.asyncio
+async def test_graph_impact_and_health(app, test_project):
+    impact = await graph_impact(
+        target="memory://specs/search",
+        horizon=2,
+        project=test_project.name,
+        output_format="json",
+    )
+    assert isinstance(impact, dict)
+    assert set(["target", "affected", "summary"]).issubset(impact.keys())
+
+    health = await graph_health(
+        scope="specs",
+        timeframe="30d",
+        project=test_project.name,
+        output_format="json",
+    )
+    assert isinstance(health, dict)
+    assert set(["metrics", "issues", "computed_at"]).issubset(health.keys())
+
+
+@pytest.mark.asyncio
+async def test_graph_reindex(app, test_project):
+    result = await graph_reindex(project=test_project.name, output_format="json")
+    assert isinstance(result, dict)
+    assert result["status"] == "queued"
+
+
+@pytest.mark.asyncio
+async def test_fcm_simulate_and_rank_actions(app, test_project):
+    simulation = await fcm_simulate(
+        actions=[{"node_id": "n1", "delta": 0.2}],
+        project=test_project.name,
+        output_format="json",
+    )
+    assert isinstance(simulation, dict)
+    assert set(["baseline", "projected", "deltas", "stability", "confidence"]).issubset(
+        simulation.keys()
+    )
+
+    ranking = await fcm_rank_actions(
+        goal="reduce-regressions",
+        top_k=2,
+        project=test_project.name,
+        output_format="json",
+    )
+    assert isinstance(ranking, dict)
+    assert set(["goal", "recommendations"]).issubset(ranking.keys())
+    assert len(ranking["recommendations"]) <= 2
+
+
+@pytest.mark.asyncio
+async def test_fcm_import_export_json_and_text(app, test_project):
+    imported = await fcm_import_model(
+        source="/tmp/model.csv",
+        format="csv_bundle_v1",
+        project=test_project.name,
+        output_format="json",
+    )
+    assert isinstance(imported, dict)
+    assert "import_id" in imported
+
+    exported_json = await fcm_export_model(
+        format="csv_bundle_v1",
+        selection={"scope": "all"},
+        project=test_project.name,
+        output_format="json",
+    )
+    assert isinstance(exported_json, dict)
+    assert set(["export_id", "files", "node_count", "edge_count"]).issubset(exported_json.keys())
+
+    exported_text = await fcm_export_model(
+        format="csv_bundle_v1",
+        selection={"scope": "all"},
+        project=test_project.name,
+        output_format="text",
+    )
+    assert isinstance(exported_text, str)
+    assert "FCM Export" in exported_text

tests/mcp/test_tool_search.py

@@ -1146,54 +1146,6 @@ async def test_search_notes_explicit_entity_types_overrides_default(monkeypatch)
     assert captured_payload["entity_types"] == ["observation"]
 
 
-# --- Tests for note_types case-insensitivity ------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_search_notes_note_types_lowercased(monkeypatch):
-    """note_types values are lowercased so 'Chapter' matches stored 'chapter'."""
-    import importlib
-
-    search_mod = importlib.import_module("basic_memory.mcp.tools.search")
-    clients_mod = importlib.import_module("basic_memory.mcp.clients")
-
-    class StubProject:
-        name = "test-project"
-        external_id = "test-external-id"
-
-    @asynccontextmanager
-    async def fake_get_project_client(*args, **kwargs):
-        yield (object(), StubProject())
-
-    async def fake_resolve_project_and_path(
-        client, identifier, project=None, context=None, headers=None
-    ):
-        return StubProject(), identifier, False
-
-    captured_payload: dict = {}
-
-    class MockSearchClient:
-        def __init__(self, *args, **kwargs):
-            pass
-
-        async def search(self, payload, page, page_size):
-            captured_payload.update(payload)
-            return SearchResponse(results=[], current_page=page, page_size=page_size)
-
-    monkeypatch.setattr(search_mod, "get_project_client", fake_get_project_client)
-    monkeypatch.setattr(search_mod, "resolve_project_and_path", fake_resolve_project_and_path)
-    monkeypatch.setattr(clients_mod, "SearchClient", MockSearchClient)
-
-    await search_mod.search_notes(
-        project="test-project",
-        query="test",
-        note_types=["Chapter", "Person"],
-    )
-
-    # note_types should be lowercased
-    assert captured_payload["note_types"] == ["chapter", "person"]
-
-
 # --- Tests for tag: prefix parsing (issue #30) ---------------------------------
 
 

tests/services/test_initialization.py

@@ -200,10 +200,10 @@ async def test_initialize_app_no_precedence_warning_when_not_conflicting(
 
 
 @pytest.mark.asyncio
-async def test_run_migrations_triggers_embedding_backfill_when_entities_exist_but_no_embeddings(
+async def test_run_migrations_triggers_embedding_backfill_on_new_revision(
     monkeypatch, app_config: BasicMemoryConfig
 ):
-    """run_migrations checks for missing embeddings (actual backfill runs in background from MCP)."""
+    """When the trigger revision is newly applied, run automatic embedding backfill once."""
 
     class StubSearchRepository:
         def __init__(self, *args, **kwargs):
@@ -224,24 +224,29 @@ async def test_run_migrations_triggers_embedding_backfill_when_entities_exist_bu
         monkeypatch.setattr("basic_memory.db.SQLiteSearchRepository", StubSearchRepository)
         monkeypatch.setattr("basic_memory.db.PostgresSearchRepository", StubSearchRepository)
 
-        needs_backfill_mock = AsyncMock(return_value=True)
-        monkeypatch.setattr(
-            "basic_memory.db._needs_semantic_embedding_backfill", needs_backfill_mock
+        load_revisions_mock = AsyncMock(
+            side_effect=[
+                set(),
+                {db.SEMANTIC_EMBEDDING_BACKFILL_REVISION},
+            ]
         )
+        backfill_mock = AsyncMock()
+        monkeypatch.setattr("basic_memory.db._load_applied_alembic_revisions", load_revisions_mock)
+        monkeypatch.setattr("basic_memory.db._run_semantic_embedding_backfill", backfill_mock)
 
         await db.run_migrations(app_config)
 
-        # Verifies the check runs — backfill itself is launched by MCP lifespan
-        needs_backfill_mock.assert_awaited_once_with(app_config, session_marker)
+        assert load_revisions_mock.await_count == 2
+        backfill_mock.assert_awaited_once_with(app_config, session_marker)
     finally:
         db._session_maker = original_session_maker  # pyright: ignore [reportPrivateUsage]
 
 
 @pytest.mark.asyncio
-async def test_run_migrations_skips_embedding_backfill_when_embeddings_already_exist(
+async def test_run_migrations_skips_embedding_backfill_when_revision_already_applied(
     monkeypatch, app_config: BasicMemoryConfig
 ):
-    """When embeddings already exist, no backfill is needed."""
+    """If the trigger revision was already present before upgrade, skip backfill."""
 
     class StubSearchRepository:
         def __init__(self, *args, **kwargs):
@@ -262,14 +267,20 @@ async def test_run_migrations_skips_embedding_backfill_when_embeddings_already_e
         monkeypatch.setattr("basic_memory.db.SQLiteSearchRepository", StubSearchRepository)
         monkeypatch.setattr("basic_memory.db.PostgresSearchRepository", StubSearchRepository)
 
-        needs_backfill_mock = AsyncMock(return_value=False)
-        monkeypatch.setattr(
-            "basic_memory.db._needs_semantic_embedding_backfill", needs_backfill_mock
+        load_revisions_mock = AsyncMock(
+            side_effect=[
+                {db.SEMANTIC_EMBEDDING_BACKFILL_REVISION},
+                {db.SEMANTIC_EMBEDDING_BACKFILL_REVISION},
+            ]
         )
+        backfill_mock = AsyncMock()
+        monkeypatch.setattr("basic_memory.db._load_applied_alembic_revisions", load_revisions_mock)
+        monkeypatch.setattr("basic_memory.db._run_semantic_embedding_backfill", backfill_mock)
 
         await db.run_migrations(app_config)
 
-        needs_backfill_mock.assert_awaited_once_with(app_config, session_marker)
+        assert load_revisions_mock.await_count == 2
+        assert backfill_mock.await_count == 0
     finally:
         db._session_maker = original_session_maker  # pyright: ignore [reportPrivateUsage]
 
@@ -367,58 +378,3 @@ async def test_semantic_embedding_backfill_skips_when_semantic_disabled(
     app_config.semantic_search_enabled = False
     await db._run_semantic_embedding_backfill(app_config, session_maker)  # pyright: ignore [reportPrivateUsage]
     assert called is False
-
-
-@pytest.mark.asyncio
-async def test_needs_semantic_embedding_backfill_true_when_entities_exist_no_embeddings(
-    app_config: BasicMemoryConfig,
-    session_maker,
-    test_project,
-):
-    """Should return True when entities exist but vector chunks table is empty."""
-    from basic_memory.repository.entity_repository import EntityRepository
-
-    entity_repository = EntityRepository(session_maker, project_id=test_project.id)
-    await entity_repository.create(
-        {
-            "title": "Test Entity",
-            "note_type": "note",
-            "entity_metadata": {},
-            "content_type": "text/markdown",
-            "file_path": "test/backfill-check.md",
-            "permalink": "test/backfill-check",
-            "project_id": test_project.id,
-            "created_at": datetime.now(),
-            "updated_at": datetime.now(),
-        }
-    )
-
-    # Clear any embeddings left by other tests in the shared DB
-    async with db.scoped_session(session_maker) as session:
-        await session.execute(db.text("DELETE FROM search_vector_chunks"))
-
-    app_config.semantic_search_enabled = True
-    result = await db._needs_semantic_embedding_backfill(app_config, session_maker)  # pyright: ignore [reportPrivateUsage]
-    assert result is True
-
-
-@pytest.mark.asyncio
-async def test_needs_semantic_embedding_backfill_false_when_no_entities(
-    app_config: BasicMemoryConfig,
-    session_maker,
-):
-    """Should return False when no entities exist (nothing to backfill)."""
-    app_config.semantic_search_enabled = True
-    result = await db._needs_semantic_embedding_backfill(app_config, session_maker)  # pyright: ignore [reportPrivateUsage]
-    assert result is False
-
-
-@pytest.mark.asyncio
-async def test_needs_semantic_embedding_backfill_false_when_semantic_disabled(
-    app_config: BasicMemoryConfig,
-    session_maker,
-):
-    """Should return False when semantic search is disabled."""
-    app_config.semantic_search_enabled = False
-    result = await db._needs_semantic_embedding_backfill(app_config, session_maker)  # pyright: ignore [reportPrivateUsage]
-    assert result is False