style: normalize spacing in project context tests

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

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

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/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/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/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),
+        )

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_project_context.py

@@ -31,6 +31,7 @@ async def test_returns_none_when_no_default_and_no_project(config_manager, monke
     config_manager.save_config(cfg)
 
     monkeypatch.delenv("BASIC_MEMORY_MCP_PROJECT", raising=False)
+
     # Prevent API fallback from returning a project via stale dependency overrides
     async def _no_api_fallback():
         return None
@@ -117,6 +118,7 @@ async def test_returns_none_when_no_default(config_manager, monkeypatch):
     config_manager.save_config(cfg)
 
     monkeypatch.delenv("BASIC_MEMORY_MCP_PROJECT", raising=False)
+
     # Prevent API fallback from returning a project via stale dependency overrides
     async def _no_api_fallback():
         return None

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