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/api/v2/routers/project_router.py

@@ -48,7 +48,7 @@ async def list_projects(
         A list of all projects with metadata
     """
     projects = await project_service.list_projects()
-    default_project = project_service.default_project
+    default_project = await project_service.get_default_project_name()
 
     project_items = [
         ProjectItem(

src/basic_memory/cli/analytics.py

@@ -25,7 +25,7 @@ import basic_memory
 # Configuration — defaults baked in, overridable via environment
 # ---------------------------------------------------------------------------
 
-_DEFAULT_UMAMI_HOST = "https://cloud.umami.is"
+_DEFAULT_UMAMI_HOST = "https://api-gateway.umami.dev"
 _DEFAULT_UMAMI_SITE_ID = "f6479898-ebaf-4e60-bce2-6dc60a3f6c5c"
 
 
@@ -76,7 +76,9 @@ def track(event_name: str, data: Optional[dict] = None) -> None:
     host = _umami_host()
     site_id = _umami_site_id()
 
+    # Umami v2 /api/send requires "type" at top level alongside "payload"
     payload = {
+        "type": "event",
         "payload": {
             "hostname": "cli.basicmemory.com",
             "language": "en",
@@ -87,7 +89,7 @@ def track(event_name: str, data: Optional[dict] = None) -> None:
                 "version": basic_memory.__version__,
                 **(data or {}),
             },
-        }
+        },
     }
 
     def _send():
@@ -97,11 +99,16 @@ def track(event_name: str, data: Optional[dict] = None) -> None:
                 data=json.dumps(payload).encode("utf-8"),
                 headers={
                     "Content-Type": "application/json",
-                    "User-Agent": f"basic-memory-cli/{basic_memory.__version__}",
+                    # Umami's bot detection rejects non-browser User-Agents
+                    "User-Agent": "Mozilla/5.0 (compatible; BasicMemoryCLI/"
+                    f"{basic_memory.__version__})",
                 },
             )
             urllib.request.urlopen(req, timeout=3)
         except Exception:
             pass  # Never break the CLI for analytics
 
-    threading.Thread(target=_send, daemon=True).start()
+    # Non-daemon so the process waits for the request to complete.
+    # The 3s urllib timeout caps the worst-case exit delay.
+    t = threading.Thread(target=_send)
+    t.start()

src/basic_memory/cli/commands/project.py

@@ -227,9 +227,7 @@ def list_projects(
 
         console.print(table)
         if cloud_error is not None:
-            console.print(
-                f"[yellow]Cloud project discovery failed: {cloud_error}[/yellow]"
-            )
+            console.print(f"[yellow]Cloud project discovery failed: {cloud_error}[/yellow]")
             console.print(
                 "[dim]Showing local projects only. "
                 "Run 'bm cloud login' or 'bm cloud api-key save <key>' if this is a credentials issue.[/dim]"

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/config.py

@@ -173,6 +173,25 @@ class BasicMemoryConfig(BaseSettings):
         description="Batch size for embedding generation.",
         gt=0,
     )
+    semantic_embedding_sync_batch_size: int = Field(
+        default=64,
+        description="Batch size for vector sync orchestration flushes.",
+        gt=0,
+    )
+    semantic_embedding_cache_dir: str | None = Field(
+        default=None,
+        description="Optional cache directory for FastEmbed model artifacts.",
+    )
+    semantic_embedding_threads: int | None = Field(
+        default=None,
+        description="Optional FastEmbed runtime thread count override.",
+        gt=0,
+    )
+    semantic_embedding_parallel: int | None = Field(
+        default=None,
+        description="Optional FastEmbed embed() parallelism override.",
+        gt=0,
+    )
     semantic_vector_k: int = Field(
         default=100,
         description="Vector candidate count for vector and hybrid retrieval.",
@@ -709,9 +728,7 @@ class ConfigManager:
                     # Create backup before overwriting so users can revert if needed
                     backup_path = self.config_file.with_suffix(".json.bak")
                     shutil.copy2(self.config_file, backup_path)
-                    logger.info(
-                        f"Migrating config to current format (backup: {backup_path})"
-                    )
+                    logger.info(f"Migrating config to current format (backup: {backup_path})")
                     save_basic_memory_config(self.config_file, _CONFIG_CACHE)
 
                 return _CONFIG_CACHE

src/basic_memory/db.py

@@ -128,8 +128,13 @@ async def _run_semantic_embedding_backfill(
             project_id=project_id,
             app_config=app_config,
         )
-        for entity_id in entity_ids:
-            await search_repository.sync_entity_vectors(entity_id)
+        batch_result = await search_repository.sync_entity_vectors_batch(entity_ids)
+        if batch_result.entities_failed > 0:
+            logger.warning(
+                "Automatic semantic embedding backfill encountered entity failures: "
+                f"project={project_name}, failed={batch_result.entities_failed}, "
+                f"failed_entity_ids={batch_result.failed_entity_ids}"
+            )
 
     logger.info(
         "Automatic semantic embedding backfill complete: "

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/project_context.py

@@ -40,6 +40,29 @@ def set_workspace_provider(provider: Callable[[], Awaitable[list[WorkspaceInfo]]
     _workspace_provider = provider
 
 
+async def _resolve_default_project_from_api() -> Optional[str]:
+    """Query the projects API for the default project.
+
+    Used as a fallback when ConfigManager has no local config (cloud mode).
+    """
+    from basic_memory.mcp.async_client import get_client
+
+    try:
+        async with get_client() as client:
+            response = await client.get("/v2/projects/")
+            if response.status_code == 200:
+                project_list = ProjectList.model_validate(response.json())
+                if project_list.default_project:
+                    return project_list.default_project
+                # Fallback: find project with is_default=True
+                for p in project_list.projects:
+                    if p.is_default:
+                        return p.name
+    except Exception:
+        pass
+    return None
+
+
 async def resolve_project_parameter(
     project: Optional[str] = None,
     allow_discovery: bool = False,
@@ -66,11 +89,16 @@ async def resolve_project_parameter(
     Returns:
         Resolved project name or None if no resolution possible
     """
-    # Load config for any values not explicitly provided
+    # Load config for any values not explicitly provided.
+    # ConfigManager reads from the local config file, which doesn't exist in cloud mode.
+    # When it returns None, fall back to querying the projects API for the is_default flag.
     if default_project is None:
         config = ConfigManager().config
         default_project = config.default_project
 
+    if default_project is None:
+        default_project = await _resolve_default_project_from_api()
+
     # Create resolver with configuration and resolve
     resolver = ProjectResolver.from_env(
         default_project=default_project,

src/basic_memory/mcp/prompts/continue_conversation.py

@@ -13,7 +13,6 @@ from pydantic import Field
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.tools.recent_activity import recent_activity
 from basic_memory.mcp.tools.search import search_notes
-from basic_memory.schemas.search import SearchResponse
 
 
 @mcp.prompt(
@@ -42,15 +41,12 @@ async def continue_conversation(
     logger.info(f"Continuing session, topic: {topic}, timeframe: {timeframe}")
 
     if topic:
-        # Search for the topic using the search tool directly
-        result = await search_notes(query=topic, after_date=timeframe)
+        # Use json format to get structured data for result counting and branching
+        result = await search_notes(query=topic, after_date=timeframe, output_format="json")
 
-        if isinstance(result, SearchResponse):
-            context_text = _format_continuation_results(result, topic)
-            result_count = len(result.results)
-        elif isinstance(result, dict):
+        if isinstance(result, dict):
             results = result.get("results", [])
-            context_text = str(result)
+            context_text = _format_continuation_results(results, topic)
             result_count = len(results)
         else:
             # Error string
@@ -111,23 +107,24 @@ async def continue_conversation(
     return prompt
 
 
-def _format_continuation_results(result: SearchResponse, topic: str) -> str:
-    """Format search results for conversation continuation context."""
-    if not result.results:
+def _format_continuation_results(results: list[dict], topic: str) -> str:
+    """Format search result dicts for conversation continuation context."""
+    if not results:
         return f"No previous context found for '{topic}'."
 
     lines = [f"## Previous Context for '{topic}'\n"]
 
-    for item in result.results:
-        title = item.title or "Untitled"
-        permalink = item.permalink or ""
+    for item in results:
+        title = item.get("title", "Untitled")
+        permalink = item.get("permalink", "")
 
         lines.append(f"### {title}")
         if permalink:
             lines.append(f"permalink: {permalink}")
             lines.append(f'Read with: `read_note("{permalink}")`')
-        if item.content:
-            content = item.content[:300] + "..." if len(item.content) > 300 else item.content
+        content = item.get("content")
+        if content:
+            content = content[:300] + "..." if len(content) > 300 else content
             lines.append(f"\n{content}")
         lines.append("")
 

src/basic_memory/mcp/prompts/search.py

@@ -11,7 +11,6 @@ from pydantic import Field
 
 from basic_memory.mcp.server import mcp
 from basic_memory.mcp.tools.search import search_notes
-from basic_memory.schemas.search import SearchResponse
 
 
 @mcp.prompt(
@@ -39,18 +38,14 @@ async def search_prompt(
     """
     logger.info(f"Searching knowledge base, query: {query}, timeframe: {timeframe}")
 
-    # Call the search tool directly — it returns SearchResponse, dict, or error string
-    result = await search_notes(query=query, after_date=timeframe)
+    # Use json format to get structured data for result counting and formatting
+    result = await search_notes(query=query, after_date=timeframe, output_format="json")
 
     # Format the tool output into a prompt with guidance
-    if isinstance(result, SearchResponse):
-        result_count = len(result.results)
-        result_text = _format_search_results(result, query)
-    elif isinstance(result, dict):
-        # json output format
+    if isinstance(result, dict):
         results = result.get("results", [])
         result_count = len(results)
-        result_text = str(result)
+        result_text = _format_search_results(results, query)
     else:
         # Error string from search tool
         result_count = 0
@@ -76,28 +71,27 @@ async def search_prompt(
     """)
 
 
-def _format_search_results(result: SearchResponse, query: str) -> str:
-    """Format SearchResponse into readable markdown."""
-    if not result.results:
+def _format_search_results(results: list[dict], query: str) -> str:
+    """Format search result dicts into readable markdown."""
+    if not results:
         return f"No results found for '{query}'."
 
-    lines = [f"Found {len(result.results)} results:\n"]
+    lines = [f"Found {len(results)} results:\n"]
 
-    for item in result.results:
-        title = item.title or "Untitled"
-        permalink = item.permalink or ""
-        score = f" (score: {item.score:.2f})" if item.score else ""
+    for item in results:
+        title = item.get("title", "Untitled")
+        permalink = item.get("permalink", "")
+        score = item.get("score")
+        score_text = f" (score: {score:.2f})" if score else ""
 
-        lines.append(f"- **{title}**{score}")
+        lines.append(f"- **{title}**{score_text}")
         if permalink:
             lines.append(f"  permalink: {permalink}")
-        if item.content:
+        content = item.get("content")
+        if content:
             # Truncate content snippet
-            content = item.content[:200] + "..." if len(item.content) > 200 else item.content
+            content = content[:200] + "..." if len(content) > 200 else content
             lines.append(f"  {content}")
         lines.append("")
 
-    if result.has_more:
-        lines.append("*More results available. Use page=2 to see next page.*")
-
     return "\n".join(lines)

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/chatgpt_tools.py

@@ -7,13 +7,13 @@ a list containing a single `{"type": "text", "text": "{...json...}"}` item.
 
 import json
 from typing import Any, Dict, List, Optional
-from loguru import logger
+
 from fastmcp import Context
+from loguru import logger
 
 from basic_memory.mcp.server import mcp
-from basic_memory.mcp.tools.search import search_notes
 from basic_memory.mcp.tools.read_note import read_note
-from basic_memory.config import ConfigManager
+from basic_memory.mcp.tools.search import search_notes
 from basic_memory.schemas.search import SearchResponse, SearchResult
 
 
@@ -113,16 +113,12 @@ async def search(
     logger.info(f"ChatGPT search request: query='{query}'")
 
     try:
-        # ChatGPT tools don't expose project parameter, so use default project
-        config = ConfigManager().config
-        default_project = config.default_project
-
-        # Call underlying search_notes with sensible defaults for ChatGPT
+        # Let search_notes resolve the default project via get_project_client(),
+        # which works in both local mode (ConfigManager) and cloud mode (database).
         results = await search_notes(
             query=query,
-            project=default_project,  # Use default project for ChatGPT
             page=1,
-            page_size=10,  # Reasonable default for ChatGPT consumption
+            page_size=10,
             output_format="json",
             context=context,
         )
@@ -180,17 +176,13 @@ async def fetch(
     logger.info(f"ChatGPT fetch request: id='{id}'")
 
     try:
-        # ChatGPT tools don't expose project parameter, so use default project
-        config = ConfigManager().config
-        default_project = config.default_project
-
-        # Call underlying read_note function (default output_format="text" returns str)
+        # Let read_note resolve the default project via get_project_client(),
+        # which works in both local mode (ConfigManager) and cloud mode (database).
         content = str(
             await read_note(
                 identifier=id,
-                project=default_project,  # Use default project for ChatGPT
                 page=1,
-                page_size=10,  # Default pagination
+                page_size=10,
                 context=context,
             )
         )

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/search.py

@@ -251,6 +251,46 @@ Error searching for '{query}': {error_message}
 - **Patterns**: `tag:example`, `category:observation`"""
 
 
+def _format_search_markdown(result: SearchResponse, project: str, query: str | None) -> str:
+    """Format SearchResponse as compact markdown text.
+
+    Produces a human-readable markdown representation suitable for LLM
+    consumption when structured data isn't needed.
+    """
+    if not result.results:
+        return f"No results found for '{query or ''}' in project '{project}'."
+
+    parts = []
+
+    # --- Header ---
+    if query:
+        parts.append(f"# Search Results: {query}")
+    else:
+        parts.append("# Search Results")
+    parts.append(f"*project: {project}*")
+    parts.append("")
+
+    # --- Result blocks ---
+    for r in result.results:
+        parts.append(f"### {r.title}")
+        parts.append(f"- permalink: {r.permalink}")
+        parts.append(f"- score: {r.score:.4f}")
+        if r.matched_chunk:
+            parts.append(f"- match: {r.matched_chunk[:200]}")
+        parts.append("")
+
+    # --- Footer with pagination ---
+    parts.append("---")
+    count = len(result.results)
+    parts.append(
+        f"*{count} result{'s' if count != 1 else ''}"
+        f" | page {result.current_page}, page_size {result.page_size}"
+        f"{' | more available' if result.has_more else ''}*"
+    )
+
+    return "\n".join(parts)
+
+
 @mcp.tool(
     description="Search across all content in the knowledge base with advanced syntax support.",
     # TODO: re-enable once MCP client rendering is working
@@ -273,7 +313,7 @@ async def search_notes(
     status: Optional[str] = None,
     min_similarity: Optional[float] = None,
     context: Context | None = None,
-) -> SearchResponse | dict | str:
+) -> dict | str:
     """Search across all content in the knowledge base with comprehensive syntax support.
 
     This tool searches the knowledge base using full-text search, pattern matching,
@@ -373,7 +413,8 @@ async def search_notes(
         context: Optional FastMCP context for performance caching.
 
     Returns:
-        SearchResponse with results and pagination info, or helpful error guidance if search fails
+        Formatted markdown text (output_format="text"), dict (output_format="json"),
+        or helpful error guidance string if search fails
 
     Examples:
         # Basic text search
@@ -519,6 +560,13 @@ async def search_notes(
             if after_date:
                 search_query.after_date = after_date
             if metadata_filters:
+                # Alias common column/model names to their frontmatter key equivalents.
+                # Users often pass "note_type" (the entity model column) when the
+                # frontmatter field is actually "type".
+                _METADATA_KEY_ALIASES = {"note_type": "type"}
+                metadata_filters = {
+                    _METADATA_KEY_ALIASES.get(k, k): v for k, v in metadata_filters.items()
+                }
                 search_query.metadata_filters = metadata_filters
             if tags:
                 search_query.tags = tags
@@ -565,7 +613,7 @@ async def search_notes(
             if output_format == "json":
                 return result.model_dump(mode="json", exclude_none=True)
 
-            return result
+            return _format_search_markdown(result, active_project.name, query)
 
         except Exception as e:
             logger.error(

src/basic_memory/repository/embedding_provider_factory.py

@@ -1,8 +1,34 @@
 """Factory for creating configured semantic embedding providers."""
 
+from threading import Lock
+
 from basic_memory.config import BasicMemoryConfig
 from basic_memory.repository.embedding_provider import EmbeddingProvider
 
+type ProviderCacheKey = tuple[str, str, int | None, int, str | None, int | None, int | None]
+
+_EMBEDDING_PROVIDER_CACHE: dict[ProviderCacheKey, EmbeddingProvider] = {}
+_EMBEDDING_PROVIDER_CACHE_LOCK = Lock()
+
+
+def _provider_cache_key(app_config: BasicMemoryConfig) -> ProviderCacheKey:
+    """Build a stable cache key from provider-relevant semantic embedding config."""
+    return (
+        app_config.semantic_embedding_provider.strip().lower(),
+        app_config.semantic_embedding_model,
+        app_config.semantic_embedding_dimensions,
+        app_config.semantic_embedding_batch_size,
+        app_config.semantic_embedding_cache_dir,
+        app_config.semantic_embedding_threads,
+        app_config.semantic_embedding_parallel,
+    )
+
+
+def reset_embedding_provider_cache() -> None:
+    """Clear process-level embedding provider cache (used by tests)."""
+    with _EMBEDDING_PROVIDER_CACHE_LOCK:
+        _EMBEDDING_PROVIDER_CACHE.clear()
+
 
 def create_embedding_provider(app_config: BasicMemoryConfig) -> EmbeddingProvider:
     """Create an embedding provider based on semantic config.
@@ -10,32 +36,50 @@ def create_embedding_provider(app_config: BasicMemoryConfig) -> EmbeddingProvide
     When semantic_embedding_dimensions is set in config, it overrides
     the provider's default dimensions (384 for FastEmbed, 1536 for OpenAI).
     """
+    cache_key = _provider_cache_key(app_config)
+    with _EMBEDDING_PROVIDER_CACHE_LOCK:
+        if cached_provider := _EMBEDDING_PROVIDER_CACHE.get(cache_key):
+            return cached_provider
+
     provider_name = app_config.semantic_embedding_provider.strip().lower()
     extra_kwargs: dict = {}
     if app_config.semantic_embedding_dimensions is not None:
         extra_kwargs["dimensions"] = app_config.semantic_embedding_dimensions
 
+    provider: EmbeddingProvider
     if provider_name == "fastembed":
         # Deferred import: fastembed (and its onnxruntime dep) may not be installed
         from basic_memory.repository.fastembed_provider import FastEmbedEmbeddingProvider
 
-        return FastEmbedEmbeddingProvider(
+        if app_config.semantic_embedding_cache_dir is not None:
+            extra_kwargs["cache_dir"] = app_config.semantic_embedding_cache_dir
+        if app_config.semantic_embedding_threads is not None:
+            extra_kwargs["threads"] = app_config.semantic_embedding_threads
+        if app_config.semantic_embedding_parallel is not None:
+            extra_kwargs["parallel"] = app_config.semantic_embedding_parallel
+
+        provider = FastEmbedEmbeddingProvider(
             model_name=app_config.semantic_embedding_model,
             batch_size=app_config.semantic_embedding_batch_size,
             **extra_kwargs,
         )
-
-    if provider_name == "openai":
+    elif provider_name == "openai":
         # Deferred import: openai may not be installed
         from basic_memory.repository.openai_provider import OpenAIEmbeddingProvider
 
         model_name = app_config.semantic_embedding_model or "text-embedding-3-small"
         if model_name == "bge-small-en-v1.5":
             model_name = "text-embedding-3-small"
-        return OpenAIEmbeddingProvider(
+        provider = OpenAIEmbeddingProvider(
             model_name=model_name,
             batch_size=app_config.semantic_embedding_batch_size,
             **extra_kwargs,
         )
+    else:
+        raise ValueError(f"Unsupported semantic embedding provider: {provider_name}")
 
-    raise ValueError(f"Unsupported semantic embedding provider: {provider_name}")
+    with _EMBEDDING_PROVIDER_CACHE_LOCK:
+        if cached_provider := _EMBEDDING_PROVIDER_CACHE.get(cache_key):
+            return cached_provider
+        _EMBEDDING_PROVIDER_CACHE[cache_key] = provider
+        return provider

src/basic_memory/repository/fastembed_provider.py

@@ -5,6 +5,8 @@ from __future__ import annotations
 import asyncio
 from typing import TYPE_CHECKING
 
+from loguru import logger
+
 from basic_memory.repository.embedding_provider import EmbeddingProvider
 from basic_memory.repository.semantic_errors import SemanticDependenciesMissingError
 
@@ -19,16 +21,25 @@ class FastEmbedEmbeddingProvider(EmbeddingProvider):
         "bge-small-en-v1.5": "BAAI/bge-small-en-v1.5",
     }
 
+    def _effective_parallel(self) -> int | None:
+        return self.parallel if self.parallel is not None and self.parallel > 1 else None
+
     def __init__(
         self,
         model_name: str = "bge-small-en-v1.5",
         *,
         batch_size: int = 64,
         dimensions: int = 384,
+        cache_dir: str | None = None,
+        threads: int | None = None,
+        parallel: int | None = None,
     ) -> None:
         self.model_name = model_name
         self.dimensions = dimensions
         self.batch_size = batch_size
+        self.cache_dir = cache_dir
+        self.threads = threads
+        self.parallel = parallel
         self._model: TextEmbedding | None = None
         self._model_lock = asyncio.Lock()
 
@@ -52,9 +63,29 @@ class FastEmbedEmbeddingProvider(EmbeddingProvider):
                         "pip install -U basic-memory"
                     ) from exc
                 resolved_model_name = self._MODEL_ALIASES.get(self.model_name, self.model_name)
+                if self.cache_dir is not None and self.threads is not None:
+                    return TextEmbedding(
+                        model_name=resolved_model_name,
+                        cache_dir=self.cache_dir,
+                        threads=self.threads,
+                    )
+                if self.cache_dir is not None:
+                    return TextEmbedding(model_name=resolved_model_name, cache_dir=self.cache_dir)
+                if self.threads is not None:
+                    return TextEmbedding(model_name=resolved_model_name, threads=self.threads)
                 return TextEmbedding(model_name=resolved_model_name)
 
             self._model = await asyncio.to_thread(_create_model)
+            logger.info(
+                "FastEmbed model loaded: model_name={model_name} batch_size={batch_size} "
+                "threads={threads} configured_parallel={configured_parallel} "
+                "effective_parallel={effective_parallel}",
+                model_name=self._MODEL_ALIASES.get(self.model_name, self.model_name),
+                batch_size=self.batch_size,
+                threads=self.threads,
+                configured_parallel=self.parallel,
+                effective_parallel=self._effective_parallel(),
+            )
             return self._model
 
     async def embed_documents(self, texts: list[str]) -> list[list[float]]:
@@ -62,9 +93,23 @@ class FastEmbedEmbeddingProvider(EmbeddingProvider):
             return []
 
         model = await self._load_model()
+        effective_parallel = self._effective_parallel()
+        logger.debug(
+            "FastEmbed embed_documents call: text_count={text_count} batch_size={batch_size} "
+            "threads={threads} configured_parallel={configured_parallel} "
+            "effective_parallel={effective_parallel}",
+            text_count=len(texts),
+            batch_size=self.batch_size,
+            threads=self.threads,
+            configured_parallel=self.parallel,
+            effective_parallel=effective_parallel,
+        )
 
         def _embed_batch() -> list[list[float]]:
-            vectors = list(model.embed(texts, batch_size=self.batch_size))
+            embed_kwargs: dict[str, int] = {"batch_size": self.batch_size}
+            if effective_parallel is not None:
+                embed_kwargs["parallel"] = effective_parallel
+            vectors = list(model.embed(texts, **embed_kwargs))
             normalized: list[list[float]] = []
             for vector in vectors:
                 values = vector.tolist() if hasattr(vector, "tolist") else vector

src/basic_memory/repository/postgres_search_repository.py

@@ -58,6 +58,9 @@ class PostgresSearchRepository(SearchRepositoryBase):
         self._semantic_enabled = self._app_config.semantic_search_enabled
         self._semantic_vector_k = self._app_config.semantic_vector_k
         self._semantic_min_similarity = self._app_config.semantic_min_similarity
+        self._semantic_embedding_sync_batch_size = (
+            self._app_config.semantic_embedding_sync_batch_size
+        )
         self._embedding_provider = embedding_provider
         self._vector_dimensions = 384
         self._vector_tables_initialized = False

src/basic_memory/repository/search_repository.py

@@ -7,7 +7,7 @@ The actual repository implementations are backend-specific:
 """
 
 from datetime import datetime
-from typing import List, Optional, Protocol
+from typing import Any, Callable, List, Optional, Protocol
 
 from sqlalchemy import Result
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
@@ -15,6 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 from basic_memory.config import BasicMemoryConfig, ConfigManager, DatabaseBackend
 from basic_memory.repository.postgres_search_repository import PostgresSearchRepository
 from basic_memory.repository.search_index_row import SearchIndexRow
+from basic_memory.repository.search_repository_base import VectorSyncBatchResult
 from basic_memory.repository.sqlite_search_repository import SQLiteSearchRepository
 from basic_memory.schemas.search import SearchItemType, SearchRetrievalMode
 
@@ -69,6 +70,14 @@ class SearchRepository(Protocol):
         """Sync semantic vector chunks for an entity."""
         ...
 
+    async def sync_entity_vectors_batch(
+        self,
+        entity_ids: list[int],
+        progress_callback: Optional[Callable[[int, int, int], Any]] = None,
+    ) -> VectorSyncBatchResult:
+        """Sync semantic vector chunks for a batch of entities."""
+        ...
+
     async def execute_query(self, query, params: dict) -> Result:
         """Execute a raw SQL query."""
         ...

src/basic_memory/repository/search_repository_base.py

@@ -5,9 +5,9 @@ import json
 import re
 import time
 from abc import ABC, abstractmethod
-from dataclasses import replace
+from dataclasses import dataclass, field, replace
 from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 from loguru import logger
 from sqlalchemy import Executable, Result, text
@@ -35,6 +35,50 @@ HEADER_LINE_PATTERN = re.compile(r"^\s*#{1,6}\s+")
 BULLET_PATTERN = re.compile(r"^[\-\*]\s+")
 
 
+@dataclass
+class VectorSyncBatchResult:
+    """Aggregate result for batched semantic vector sync runs."""
+
+    entities_total: int
+    entities_synced: int
+    entities_failed: int
+    failed_entity_ids: list[int] = field(default_factory=list)
+    embedding_jobs_total: int = 0
+    embed_seconds_total: float = 0.0
+    write_seconds_total: float = 0.0
+
+
+@dataclass
+class _PreparedEntityVectorSync:
+    """Prepared chunk mutations + embedding jobs for one entity."""
+
+    entity_id: int
+    sync_start: float
+    source_rows_count: int
+    embedding_jobs: list[tuple[int, str]]
+
+
+@dataclass
+class _PendingEmbeddingJob:
+    """Pending embedding write entry with entity ownership metadata."""
+
+    entity_id: int
+    chunk_row_id: int
+    chunk_text: str
+
+
+@dataclass
+class _EntitySyncRuntime:
+    """Per-entity runtime counters used while flushes are in flight."""
+
+    sync_start: float
+    source_rows_count: int
+    embedding_jobs_count: int
+    remaining_jobs: int
+    embed_seconds: float = 0.0
+    write_seconds: float = 0.0
+
+
 class SearchRepositoryBase(ABC):
     """Abstract base class for backend-specific search repository implementations.
 
@@ -54,6 +98,7 @@ class SearchRepositoryBase(ABC):
     _semantic_vector_k: int
     _semantic_min_similarity: float
     _embedding_provider: Optional[EmbeddingProvider]
+    _semantic_embedding_sync_batch_size: int
     _vector_dimensions: int
     _vector_tables_initialized: bool
 
@@ -563,15 +608,205 @@ class SearchRepositoryBase(ABC):
     # ------------------------------------------------------------------
 
     async def sync_entity_vectors(self, entity_id: int) -> None:
-        """Sync semantic chunk rows + embeddings for a single entity.
+        """Sync semantic chunk rows + embeddings for a single entity."""
+        await self._sync_entity_vectors_internal(
+            [entity_id],
+            progress_callback=None,
+            continue_on_error=False,
+        )
 
-        This is the shared orchestration logic. Backend-specific SQL operations
-        are delegated to abstract hooks (_delete_entity_chunks, _write_embeddings, etc.).
-        """
+    async def sync_entity_vectors_batch(
+        self,
+        entity_ids: list[int],
+        progress_callback: Optional[Callable[[int, int, int], Any]] = None,
+    ) -> VectorSyncBatchResult:
+        """Sync semantic chunk rows + embeddings for a batch of entities."""
+        return await self._sync_entity_vectors_internal(
+            entity_ids,
+            progress_callback=progress_callback,
+            continue_on_error=True,
+        )
+
+    async def _sync_entity_vectors_internal(
+        self,
+        entity_ids: list[int],
+        progress_callback: Optional[Callable[[int, int, int], Any]],
+        continue_on_error: bool,
+    ) -> VectorSyncBatchResult:
+        """Run shared vector sync orchestration for one or many entities."""
         self._assert_semantic_available()
         await self._ensure_vector_tables()
         assert self._embedding_provider is not None
 
+        total_entities = len(entity_ids)
+        result = VectorSyncBatchResult(
+            entities_total=total_entities,
+            entities_synced=0,
+            entities_failed=0,
+        )
+        if total_entities == 0:
+            return result
+
+        logger.info(
+            "Vector batch sync start: project_id={project_id} entities_total={entities_total} "
+            "sync_batch_size={sync_batch_size}",
+            project_id=self.project_id,
+            entities_total=total_entities,
+            sync_batch_size=self._semantic_embedding_sync_batch_size,
+        )
+
+        pending_jobs: list[_PendingEmbeddingJob] = []
+        entity_runtime: dict[int, _EntitySyncRuntime] = {}
+        failed_entity_ids: set[int] = set()
+        synced_entity_ids: set[int] = set()
+
+        for index, entity_id in enumerate(entity_ids):
+            if progress_callback is not None:
+                progress_callback(entity_id, index, total_entities)
+
+            try:
+                prepared = await self._prepare_entity_vector_jobs(entity_id)
+            except Exception as exc:
+                if not continue_on_error:
+                    raise
+                failed_entity_ids.add(entity_id)
+                logger.warning(
+                    "Vector batch sync entity prepare failed: project_id={project_id} "
+                    "entity_id={entity_id} error={error}",
+                    project_id=self.project_id,
+                    entity_id=entity_id,
+                    error=str(exc),
+                )
+                continue
+
+            embedding_jobs_count = len(prepared.embedding_jobs)
+            result.embedding_jobs_total += embedding_jobs_count
+
+            if embedding_jobs_count == 0:
+                synced_entity_ids.add(entity_id)
+                total_seconds = time.perf_counter() - prepared.sync_start
+                self._log_vector_sync_complete(
+                    entity_id=entity_id,
+                    total_seconds=total_seconds,
+                    embed_seconds=0.0,
+                    write_seconds=0.0,
+                    source_rows_count=prepared.source_rows_count,
+                    embedding_jobs_count=0,
+                )
+                continue
+
+            entity_runtime[entity_id] = _EntitySyncRuntime(
+                sync_start=prepared.sync_start,
+                source_rows_count=prepared.source_rows_count,
+                embedding_jobs_count=embedding_jobs_count,
+                remaining_jobs=embedding_jobs_count,
+            )
+            pending_jobs.extend(
+                _PendingEmbeddingJob(
+                    entity_id=entity_id, chunk_row_id=row_id, chunk_text=chunk_text
+                )
+                for row_id, chunk_text in prepared.embedding_jobs
+            )
+
+            while len(pending_jobs) >= self._semantic_embedding_sync_batch_size:
+                flush_jobs = pending_jobs[: self._semantic_embedding_sync_batch_size]
+                pending_jobs = pending_jobs[self._semantic_embedding_sync_batch_size :]
+                try:
+                    embed_seconds, write_seconds = await self._flush_embedding_jobs(
+                        flush_jobs=flush_jobs,
+                        entity_runtime=entity_runtime,
+                        synced_entity_ids=synced_entity_ids,
+                    )
+                    result.embed_seconds_total += embed_seconds
+                    result.write_seconds_total += write_seconds
+                except Exception as exc:
+                    if not continue_on_error:
+                        raise
+                    affected_entity_ids = sorted({job.entity_id for job in flush_jobs})
+                    failed_entity_ids.update(affected_entity_ids)
+                    for failed_entity_id in affected_entity_ids:
+                        entity_runtime.pop(failed_entity_id, None)
+                    logger.warning(
+                        "Vector batch sync flush failed: project_id={project_id} "
+                        "affected_entities={affected_entities} chunk_count={chunk_count} error={error}",
+                        project_id=self.project_id,
+                        affected_entities=affected_entity_ids,
+                        chunk_count=len(flush_jobs),
+                        error=str(exc),
+                    )
+
+        if pending_jobs:
+            flush_jobs = list(pending_jobs)
+            pending_jobs = []
+            try:
+                embed_seconds, write_seconds = await self._flush_embedding_jobs(
+                    flush_jobs=flush_jobs,
+                    entity_runtime=entity_runtime,
+                    synced_entity_ids=synced_entity_ids,
+                )
+                result.embed_seconds_total += embed_seconds
+                result.write_seconds_total += write_seconds
+            except Exception as exc:
+                if not continue_on_error:
+                    raise
+                affected_entity_ids = sorted({job.entity_id for job in flush_jobs})
+                failed_entity_ids.update(affected_entity_ids)
+                for failed_entity_id in affected_entity_ids:
+                    entity_runtime.pop(failed_entity_id, None)
+                logger.warning(
+                    "Vector batch sync final flush failed: project_id={project_id} "
+                    "affected_entities={affected_entities} chunk_count={chunk_count} error={error}",
+                    project_id=self.project_id,
+                    affected_entities=affected_entity_ids,
+                    chunk_count=len(flush_jobs),
+                    error=str(exc),
+                )
+
+        # Trigger: this should never happen after all flushes succeed.
+        # Why: remaining jobs mean runtime tracking drifted from queued jobs.
+        # Outcome: fail-safe marks these entities as failed to avoid false positives.
+        if entity_runtime:
+            orphan_runtime_entities = sorted(entity_runtime.keys())
+            failed_entity_ids.update(orphan_runtime_entities)
+            logger.warning(
+                "Vector batch sync left unfinished entities after flushes: "
+                "project_id={project_id} unfinished_entities={unfinished_entities}",
+                project_id=self.project_id,
+                unfinished_entities=orphan_runtime_entities,
+            )
+
+        # Keep result counters aligned with successful/failed terminal states.
+        synced_entity_ids.difference_update(failed_entity_ids)
+        result.failed_entity_ids = sorted(failed_entity_ids)
+        result.entities_failed = len(result.failed_entity_ids)
+        result.entities_synced = len(synced_entity_ids)
+
+        logger.info(
+            "Vector batch sync complete: project_id={project_id} entities_total={entities_total} "
+            "entities_synced={entities_synced} entities_failed={entities_failed} "
+            "embedding_jobs_total={embedding_jobs_total} embed_seconds_total={embed_seconds_total:.3f} "
+            "write_seconds_total={write_seconds_total:.3f}",
+            project_id=self.project_id,
+            entities_total=result.entities_total,
+            entities_synced=result.entities_synced,
+            entities_failed=result.entities_failed,
+            embedding_jobs_total=result.embedding_jobs_total,
+            embed_seconds_total=result.embed_seconds_total,
+            write_seconds_total=result.write_seconds_total,
+        )
+
+        return result
+
+    async def _prepare_entity_vector_jobs(self, entity_id: int) -> _PreparedEntityVectorSync:
+        """Prepare chunk mutations and embedding jobs for one entity."""
+        sync_start = time.perf_counter()
+
+        logger.info(
+            "Vector sync start: project_id={project_id} entity_id={entity_id}",
+            project_id=self.project_id,
+            entity_id=entity_id,
+        )
+
         async with db.scoped_session(self.session_maker) as session:
             await self._prepare_vector_session(session)
 
@@ -597,18 +832,49 @@ class SearchRepositoryBase(ABC):
                 },
             )
             rows = row_result.fetchall()
+            source_rows_count = len(rows)
+            built_chunk_records_count = 0
 
             # No search_index rows → delete all chunk/embedding data for this entity.
             if not rows:
+                logger.info(
+                    "Vector sync source prepared: project_id={project_id} entity_id={entity_id} "
+                    "source_rows_count={source_rows_count} "
+                    "built_chunk_records_count={built_chunk_records_count}",
+                    project_id=self.project_id,
+                    entity_id=entity_id,
+                    source_rows_count=source_rows_count,
+                    built_chunk_records_count=built_chunk_records_count,
+                )
                 await self._delete_entity_chunks(session, entity_id)
                 await session.commit()
-                return
+                return _PreparedEntityVectorSync(
+                    entity_id=entity_id,
+                    sync_start=sync_start,
+                    source_rows_count=source_rows_count,
+                    embedding_jobs=[],
+                )
 
             chunk_records = self._build_chunk_records(rows)
+            built_chunk_records_count = len(chunk_records)
+            logger.info(
+                "Vector sync source prepared: project_id={project_id} entity_id={entity_id} "
+                "source_rows_count={source_rows_count} "
+                "built_chunk_records_count={built_chunk_records_count}",
+                project_id=self.project_id,
+                entity_id=entity_id,
+                source_rows_count=source_rows_count,
+                built_chunk_records_count=built_chunk_records_count,
+            )
             if not chunk_records:
                 await self._delete_entity_chunks(session, entity_id)
                 await session.commit()
-                return
+                return _PreparedEntityVectorSync(
+                    entity_id=entity_id,
+                    sync_start=sync_start,
+                    source_rows_count=source_rows_count,
+                    embedding_jobs=[],
+                )
 
             # --- Diff existing chunks against incoming ---
             existing_rows_result = await session.execute(
@@ -620,6 +886,7 @@ class SearchRepositoryBase(ABC):
                 {"project_id": self.project_id, "entity_id": entity_id},
             )
             existing_by_key = {row.chunk_key: row for row in existing_rows_result.fetchall()}
+            existing_chunks_count = len(existing_by_key)
             incoming_hashes = {
                 record["chunk_key"]: record["source_hash"] for record in chunk_records
             }
@@ -628,6 +895,7 @@ class SearchRepositoryBase(ABC):
                 for chunk_key, row in existing_by_key.items()
                 if chunk_key not in incoming_hashes
             ]
+            stale_chunks_count = len(stale_ids)
 
             if stale_ids:
                 await self._delete_stale_chunks(session, stale_ids, entity_id)
@@ -641,6 +909,8 @@ class SearchRepositoryBase(ABC):
                 {"project_id": self.project_id, "entity_id": entity_id},
             )
             orphan_rows = orphan_result.fetchall()
+            orphan_ids = {int(row.id) for row in orphan_rows}
+            orphan_chunks_count = len(orphan_ids)
 
             # --- Upsert changed / new chunks, collect embedding jobs ---
             timestamp_expr = self._timestamp_now_expr()
@@ -651,7 +921,7 @@ class SearchRepositoryBase(ABC):
                 # Trigger: chunk exists and hash matches (no content change)
                 # but chunk has no embedding (orphan from crash).
                 # Outcome: schedule re-embedding without touching chunk metadata.
-                is_orphan = current and any(o.id == current.id for o in orphan_rows)
+                is_orphan = current and int(current.id) in orphan_ids
                 if current and current.source_hash == record["source_hash"] and not is_orphan:
                     continue
 
@@ -694,20 +964,141 @@ class SearchRepositoryBase(ABC):
                 row_id = int(inserted.scalar_one())
                 embedding_jobs.append((row_id, record["chunk_text"]))
 
+            logger.info(
+                "Vector sync diff complete: project_id={project_id} entity_id={entity_id} "
+                "existing_chunks_count={existing_chunks_count} "
+                "stale_chunks_count={stale_chunks_count} "
+                "orphan_chunks_count={orphan_chunks_count} "
+                "embedding_jobs_count={embedding_jobs_count}",
+                project_id=self.project_id,
+                entity_id=entity_id,
+                existing_chunks_count=existing_chunks_count,
+                stale_chunks_count=stale_chunks_count,
+                orphan_chunks_count=orphan_chunks_count,
+                embedding_jobs_count=len(embedding_jobs),
+            )
             await session.commit()
 
-        if not embedding_jobs:
-            return
+        return _PreparedEntityVectorSync(
+            entity_id=entity_id,
+            sync_start=sync_start,
+            source_rows_count=source_rows_count,
+            embedding_jobs=embedding_jobs,
+        )
 
-        texts = [t for _, t in embedding_jobs]
+    async def _flush_embedding_jobs(
+        self,
+        flush_jobs: list[_PendingEmbeddingJob],
+        entity_runtime: dict[int, _EntitySyncRuntime],
+        synced_entity_ids: set[int],
+    ) -> tuple[float, float]:
+        """Embed and persist one queued flush chunk."""
+        if not flush_jobs:
+            return 0.0, 0.0
+        assert self._embedding_provider is not None
+
+        embed_start = time.perf_counter()
+        texts = [job.chunk_text for job in flush_jobs]
         embeddings = await self._embedding_provider.embed_documents(texts)
-        if len(embeddings) != len(embedding_jobs):
+        embed_seconds = time.perf_counter() - embed_start
+        embed_rate = (len(flush_jobs) / embed_seconds) if embed_seconds > 0 else 0.0
+        logger.info(
+            "Vector batch embed flush: project_id={project_id} chunk_count={chunk_count} "
+            "embed_seconds={embed_seconds:.3f} embed_rate_chunks_per_second={embed_rate:.2f}",
+            project_id=self.project_id,
+            chunk_count=len(flush_jobs),
+            embed_seconds=embed_seconds,
+            embed_rate=embed_rate,
+        )
+        if len(embeddings) != len(flush_jobs):
             raise RuntimeError("Embedding provider returned an unexpected number of vectors.")
 
+        write_start = time.perf_counter()
         async with db.scoped_session(self.session_maker) as session:
             await self._prepare_vector_session(session)
-            await self._write_embeddings(session, embedding_jobs, embeddings)
+            write_jobs = [(job.chunk_row_id, job.chunk_text) for job in flush_jobs]
+            await self._write_embeddings(session, write_jobs, embeddings)
             await session.commit()
+        write_seconds = time.perf_counter() - write_start
+        write_rate = (len(flush_jobs) / write_seconds) if write_seconds > 0 else 0.0
+        logger.info(
+            "Vector batch write flush: project_id={project_id} row_count={row_count} "
+            "write_seconds={write_seconds:.3f} write_rate_rows_per_second={write_rate:.2f}",
+            project_id=self.project_id,
+            row_count=len(flush_jobs),
+            write_seconds=write_seconds,
+            write_rate=write_rate,
+        )
+
+        flush_size = len(flush_jobs)
+        entity_job_counts: dict[int, int] = {}
+        for job in flush_jobs:
+            entity_job_counts[job.entity_id] = entity_job_counts.get(job.entity_id, 0) + 1
+
+        for entity_id, entity_job_count in entity_job_counts.items():
+            runtime = entity_runtime.get(entity_id)
+            if runtime is None:
+                continue
+            runtime.remaining_jobs -= entity_job_count
+
+            # Attribute flush wall-clock to entities in proportion to rows written.
+            flush_share = entity_job_count / flush_size
+            runtime.embed_seconds += embed_seconds * flush_share
+            runtime.write_seconds += write_seconds * flush_share
+
+            if runtime.remaining_jobs <= 0:
+                synced_entity_ids.add(entity_id)
+                total_seconds = time.perf_counter() - runtime.sync_start
+                self._log_vector_sync_complete(
+                    entity_id=entity_id,
+                    total_seconds=total_seconds,
+                    embed_seconds=runtime.embed_seconds,
+                    write_seconds=runtime.write_seconds,
+                    source_rows_count=runtime.source_rows_count,
+                    embedding_jobs_count=runtime.embedding_jobs_count,
+                )
+                entity_runtime.pop(entity_id, None)
+
+        return embed_seconds, write_seconds
+
+    def _log_vector_sync_complete(
+        self,
+        *,
+        entity_id: int,
+        total_seconds: float,
+        embed_seconds: float,
+        write_seconds: float,
+        source_rows_count: int,
+        embedding_jobs_count: int,
+    ) -> None:
+        """Log completion and slow-entity warnings with a consistent format."""
+        logger.info(
+            "Vector sync complete: project_id={project_id} entity_id={entity_id} "
+            "total_seconds={total_seconds:.3f} embed_seconds={embed_seconds:.3f} "
+            "write_seconds={write_seconds:.3f} source_rows_count={source_rows_count} "
+            "embedding_jobs_count={embedding_jobs_count}",
+            project_id=self.project_id,
+            entity_id=entity_id,
+            total_seconds=total_seconds,
+            embed_seconds=embed_seconds,
+            write_seconds=write_seconds,
+            source_rows_count=source_rows_count,
+            embedding_jobs_count=embedding_jobs_count,
+        )
+        if total_seconds > 10:
+            logger.warning(
+                "Vector sync slow entity: project_id={project_id} entity_id={entity_id} "
+                "total_seconds={total_seconds:.3f} embed_seconds={embed_seconds:.3f} "
+                "write_seconds={write_seconds:.3f} source_rows_count={source_rows_count} "
+                "embedding_jobs_count={embedding_jobs_count}",
+                project_id=self.project_id,
+                entity_id=entity_id,
+                total_seconds=total_seconds,
+                embed_seconds=embed_seconds,
+                write_seconds=write_seconds,
+                source_rows_count=source_rows_count,
+                embedding_jobs_count=embedding_jobs_count,
+            )
 
     async def _prepare_vector_session(self, session: AsyncSession) -> None:
         """Hook for per-session setup (e.g. loading sqlite-vec extension).
@@ -852,6 +1243,7 @@ class SearchRepositoryBase(ABC):
         min_similarity: Optional[float] = None,
         limit: int,
         offset: int,
+        _emit_observability_log: bool = True,
     ) -> List[SearchIndexRow]:
         """Run vector-only search returning chunk-level results.
 
@@ -862,16 +1254,65 @@ class SearchRepositoryBase(ABC):
         self._assert_semantic_available()
         await self._ensure_vector_tables()
         assert self._embedding_provider is not None
-        query_embedding = await self._embedding_provider.embed_query(search_text.strip())
+        query_text = search_text.strip()
         candidate_limit = max(self._semantic_vector_k, (limit + offset) * 10)
+        query_start = time.perf_counter()
+        embed_start = time.perf_counter()
+        query_embedding = await self._embedding_provider.embed_query(query_text)
+        embed_ms = (time.perf_counter() - embed_start) * 1000
+        vector_query_start = time.perf_counter()
 
         async with db.scoped_session(self.session_maker) as session:
             await self._prepare_vector_session(session)
             vector_rows = await self._run_vector_query(session, query_embedding, candidate_limit)
+        vector_query_ms = (time.perf_counter() - vector_query_start) * 1000
+        vector_row_count = len(vector_rows)
+        hydrate_ms = 0.0
+
+        def _log_vector_summary() -> None:
+            if not _emit_observability_log:
+                return
+
+            total_ms = (time.perf_counter() - query_start) * 1000
+            logger.info(
+                "Semantic query timing: project_id={project_id} retrieval_mode={retrieval_mode} "
+                "query_length={query_length} candidate_limit={candidate_limit} "
+                "vector_row_count={vector_row_count} embed_ms={embed_ms:.2f} "
+                "vector_query_ms={vector_query_ms:.2f} hydrate_ms={hydrate_ms:.2f} "
+                "total_ms={total_ms:.2f}",
+                project_id=self.project_id,
+                retrieval_mode="vector",
+                query_length=len(query_text),
+                candidate_limit=candidate_limit,
+                vector_row_count=vector_row_count,
+                embed_ms=embed_ms,
+                vector_query_ms=vector_query_ms,
+                hydrate_ms=hydrate_ms,
+                total_ms=total_ms,
+            )
+            if total_ms > 2000:
+                logger.warning(
+                    "[SEMANTIC_SLOW_QUERY] Semantic query timing: project_id={project_id} "
+                    "retrieval_mode={retrieval_mode} query_length={query_length} "
+                    "candidate_limit={candidate_limit} vector_row_count={vector_row_count} "
+                    "embed_ms={embed_ms:.2f} vector_query_ms={vector_query_ms:.2f} "
+                    "hydrate_ms={hydrate_ms:.2f} total_ms={total_ms:.2f}",
+                    project_id=self.project_id,
+                    retrieval_mode="vector",
+                    query_length=len(query_text),
+                    candidate_limit=candidate_limit,
+                    vector_row_count=vector_row_count,
+                    embed_ms=embed_ms,
+                    vector_query_ms=vector_query_ms,
+                    hydrate_ms=hydrate_ms,
+                    total_ms=total_ms,
+                )
 
         if not vector_rows:
+            _log_vector_summary()
             return []
 
+        hydrate_start = time.perf_counter()
         # Build per-search_index_row similarity scores from chunk-level results.
         # Each chunk_key encodes the search_index row type and id.
         # Track the best similarity per row (for ranking) and all chunks (for context).
@@ -893,6 +1334,8 @@ class SearchRepositoryBase(ABC):
             chunks_by_si_id.setdefault(si_id, []).append((similarity, chunk_text))
 
         if not similarity_by_si_id:
+            hydrate_ms = (time.perf_counter() - hydrate_start) * 1000
+            _log_vector_summary()
             return []
 
         # Filter out results below the minimum similarity threshold.
@@ -905,6 +1348,8 @@ class SearchRepositoryBase(ABC):
                 k: v for k, v in similarity_by_si_id.items() if v >= effective_min_similarity
             }
             if not similarity_by_si_id:
+                hydrate_ms = (time.perf_counter() - hydrate_start) * 1000
+                _log_vector_summary()
                 return []
 
         # Fetch the actual search_index rows
@@ -971,6 +1416,8 @@ class SearchRepositoryBase(ABC):
             )
 
         ranked_rows.sort(key=lambda item: item.score or 0.0, reverse=True)
+        hydrate_ms = (time.perf_counter() - hydrate_start) * 1000
+        _log_vector_summary()
         return ranked_rows[offset : offset + limit]
 
     async def _fetch_entity_rows_by_ids(self, entity_ids: list[int]) -> dict[int, SearchIndexRow]:
@@ -1093,7 +1540,10 @@ class SearchRepositoryBase(ABC):
         the dominant signal and rewards dual-source agreement.
         """
         self._assert_semantic_available()
+        query_text = search_text.strip()
+        query_start = time.perf_counter()
         candidate_limit = max(self._semantic_vector_k, (limit + offset) * 10)
+        fts_start = time.perf_counter()
         fts_results = await self.search(
             search_text=search_text,
             permalink=permalink,
@@ -1107,6 +1557,8 @@ class SearchRepositoryBase(ABC):
             limit=candidate_limit,
             offset=0,
         )
+        fts_ms = (time.perf_counter() - fts_start) * 1000
+        vector_start = time.perf_counter()
         vector_results = await self._search_vector_only(
             search_text=search_text,
             permalink=permalink,
@@ -1119,7 +1571,10 @@ class SearchRepositoryBase(ABC):
             min_similarity=min_similarity,
             limit=candidate_limit,
             offset=0,
+            _emit_observability_log=False,
         )
+        vector_ms = (time.perf_counter() - vector_start) * 1000
+        fusion_start = time.perf_counter()
 
         # --- Score-based fusion keyed on search_index row id ---
         # FTS scores are normalized to [0, 1] (BM25 is unbounded).
@@ -1171,4 +1626,40 @@ class SearchRepositoryBase(ABC):
             if row.matched_chunk_text is None and row.content_snippet:
                 row = replace(row, matched_chunk_text=row.content_snippet)
             output.append(replace(row, score=fused_score))
+        fusion_ms = (time.perf_counter() - fusion_start) * 1000
+        total_ms = (time.perf_counter() - query_start) * 1000
+        logger.info(
+            "Semantic query timing: project_id={project_id} retrieval_mode={retrieval_mode} "
+            "query_length={query_length} candidate_limit={candidate_limit} "
+            "fts_count={fts_count} vector_count={vector_count} fts_ms={fts_ms:.2f} "
+            "vector_ms={vector_ms:.2f} fusion_ms={fusion_ms:.2f} total_ms={total_ms:.2f}",
+            project_id=self.project_id,
+            retrieval_mode="hybrid",
+            query_length=len(query_text),
+            candidate_limit=candidate_limit,
+            fts_count=len(fts_results),
+            vector_count=len(vector_results),
+            fts_ms=fts_ms,
+            vector_ms=vector_ms,
+            fusion_ms=fusion_ms,
+            total_ms=total_ms,
+        )
+        if total_ms > 2500:
+            logger.warning(
+                "[SEMANTIC_SLOW_QUERY] Semantic query timing: project_id={project_id} "
+                "retrieval_mode={retrieval_mode} query_length={query_length} "
+                "candidate_limit={candidate_limit} fts_count={fts_count} "
+                "vector_count={vector_count} fts_ms={fts_ms:.2f} vector_ms={vector_ms:.2f} "
+                "fusion_ms={fusion_ms:.2f} total_ms={total_ms:.2f}",
+                project_id=self.project_id,
+                retrieval_mode="hybrid",
+                query_length=len(query_text),
+                candidate_limit=candidate_limit,
+                fts_count=len(fts_results),
+                vector_count=len(vector_results),
+                fts_ms=fts_ms,
+                vector_ms=vector_ms,
+                fusion_ms=fusion_ms,
+                total_ms=total_ms,
+            )
         return output

src/basic_memory/repository/sqlite_search_repository.py

@@ -52,6 +52,9 @@ class SQLiteSearchRepository(SearchRepositoryBase):
         self._semantic_enabled = self._app_config.semantic_search_enabled
         self._semantic_vector_k = self._app_config.semantic_vector_k
         self._semantic_min_similarity = self._app_config.semantic_min_similarity
+        self._semantic_embedding_sync_batch_size = (
+            self._app_config.semantic_embedding_sync_batch_size
+        )
         self._embedding_provider = embedding_provider
         self._sqlite_vec_lock = asyncio.Lock()
         self._vector_tables_initialized = False

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/services/link_resolver.py

@@ -301,8 +301,10 @@ class LinkResolver:
             )
 
             if results:
-                # Look for best match
-                best_match = min(results, key=lambda x: x.score)  # pyright: ignore
+                # Both SQLite and Postgres return results sorted best-first in SQL
+                # (SQLite: ORDER BY score ASC for negative BM25, Postgres: ORDER BY score DESC
+                # for positive ts_rank). Using results[0] is backend-agnostic and correct.
+                best_match = results[0]
                 logger.trace(
                     f"Selected best match from {len(results)} results: {best_match.permalink}"
                 )

src/basic_memory/services/project_service.py

@@ -82,6 +82,21 @@ class ProjectService:
         """
         return self.config_manager.default_project
 
+    async def get_default_project_name(self) -> str:
+        """Get the default project name, falling back to the database.
+
+        ConfigManager reads from the local config file, which doesn't exist
+        in cloud mode. When it returns None, fall back to the is_default
+        flag stored in the database.
+        """
+        default = self.config_manager.default_project
+        if default is not None:
+            return default
+        db_default = await self.repository.get_default_project()
+        if db_default is not None:
+            return db_default.name
+        raise ValueError("No default project configured")
+
     @property
     def current_project(self) -> Optional[str]:
         """Get the name of the currently active project.

src/basic_memory/services/search_service.py

@@ -13,7 +13,11 @@ from sqlalchemy import text
 
 from basic_memory.models import Entity
 from basic_memory.repository import EntityRepository
-from basic_memory.repository.search_repository import SearchRepository, SearchIndexRow
+from basic_memory.repository.search_repository import (
+    SearchIndexRow,
+    SearchRepository,
+    VectorSyncBatchResult,
+)
 from basic_memory.schemas.search import SearchQuery, SearchItemType, SearchRetrievalMode
 from basic_memory.services import FileService
 
@@ -377,6 +381,17 @@ class SearchService:
         """Refresh vector chunks for one entity in repositories that support semantic indexing."""
         await self.repository.sync_entity_vectors(entity_id)
 
+    async def sync_entity_vectors_batch(
+        self,
+        entity_ids: list[int],
+        progress_callback=None,
+    ) -> VectorSyncBatchResult:
+        """Refresh vector chunks for a batch of entities."""
+        return await self.repository.sync_entity_vectors_batch(
+            entity_ids,
+            progress_callback=progress_callback,
+        )
+
     async def reindex_vectors(self, progress_callback=None) -> dict:
         """Rebuild vector embeddings for all entities.
 
@@ -387,17 +402,20 @@ class SearchService:
             dict with stats: total_entities, embedded, skipped, errors
         """
         entities = await self.entity_repository.find_all()
-        stats = {"total_entities": len(entities), "embedded": 0, "skipped": 0, "errors": 0}
+        entity_ids = [entity.id for entity in entities]
+        batch_result = await self.repository.sync_entity_vectors_batch(
+            entity_ids,
+            progress_callback=progress_callback,
+        )
+        stats = {
+            "total_entities": batch_result.entities_total,
+            "embedded": batch_result.entities_synced,
+            "skipped": 0,
+            "errors": batch_result.entities_failed,
+        }
 
-        for i, entity in enumerate(entities):
-            if progress_callback:
-                progress_callback(entity.id, i, len(entities))
-            try:
-                await self.repository.sync_entity_vectors(entity.id)
-                stats["embedded"] += 1
-            except Exception as e:
-                logger.warning(f"Failed to embed entity {entity.id} ({entity.permalink}): {e}")
-                stats["errors"] += 1
+        for failed_entity_id in batch_result.failed_entity_ids:
+            logger.warning(f"Failed to embed entity {failed_entity_id}")
 
         return stats
 

test-int/mcp/test_default_project_mode_integration.py

@@ -93,32 +93,33 @@ async def test_explicit_project_overrides_default(
 
 
 @pytest.mark.asyncio
-async def test_no_default_project_requires_project(mcp_server, app, test_project):
-    """Test that tools require project parameter when no default_project is configured."""
+async def test_no_config_default_falls_back_to_db(mcp_server, app, test_project):
+    """When ConfigManager has no default_project, tools fall back to the database is_default flag."""
 
     mock_config = BasicMemoryConfig(
-        default_project=None,  # No default
+        default_project=None,  # No config default
         projects={test_project.name: test_project.path},
     )
 
+    # test_project has is_default=True in the database, so write_note should
+    # resolve to it via the API fallback in resolve_project_parameter.
     with patch.object(ConfigManager, "config", mock_config):
         async with Client(mcp_server) as client:
-            with pytest.raises(Exception) as exc_info:
-                await client.call_tool(
-                    "write_note",
-                    {
-                        "title": "Should Fail",
-                        "directory": "test",
-                        "content": "# Should Fail\n\nThis should fail because no project specified.",
-                    },
-                )
-
-            error_message = str(exc_info.value)
-            assert (
-                "No project specified" in error_message
-                or "project parameter" in error_message.lower()
+            result = await client.call_tool(
+                "write_note",
+                {
+                    "title": "DB Fallback Test",
+                    "directory": "test",
+                    "content": "# DB Fallback Test\n\nShould resolve to the database default project.",
+                },
             )
 
+            assert len(result.content) == 1
+            response_text = result.content[0].text  # pyright: ignore [reportAttributeAccessIssue]
+
+            assert f"project: {test_project.name}" in response_text
+            assert "# Created note" in response_text
+
 
 @pytest.mark.asyncio
 async def test_cli_constraint_overrides_default_project(

test-int/mcp/test_delete_note_integration.py

@@ -105,11 +105,8 @@ async def test_delete_note_by_permalink(mcp_server, app, test_project):
             },
         )
 
-        # Should have no results
-        assert (
-            '"results": []' in search_result.content[0].text
-            or '"results":[]' in search_result.content[0].text
-        )
+        # Default text format returns "No results found" when empty
+        assert "No results found" in search_result.content[0].text
 
 
 @pytest.mark.asyncio
@@ -387,11 +384,8 @@ async def test_delete_multiple_notes_sequentially(mcp_server, app, test_project)
             },
         )
 
-        # Should have no results
-        assert (
-            '"results": []' in search_result.content[0].text
-            or '"results":[]' in search_result.content[0].text
-        )
+        # Default text format returns "No results found" when empty
+        assert "No results found" in search_result.content[0].text
 
 
 @pytest.mark.asyncio

test-int/mcp/test_search_integration.py

@@ -362,9 +362,9 @@ async def test_search_pagination(mcp_server, app, test_project):
         )
 
         result_text = search_result.content[0].text
-        # Should contain 5 results and pagination info
-        assert '"current_page":1' in result_text
-        assert '"page_size":5' in result_text
+        # Text format includes pagination info in footer
+        assert "page 1" in result_text
+        assert "page_size 5" in result_text
 
         # Search page 2
         search_result = await client.call_tool(
@@ -378,7 +378,7 @@ async def test_search_pagination(mcp_server, app, test_project):
         )
 
         result_text = search_result.content[0].text
-        assert '"current_page":2' in result_text
+        assert "page 2" in result_text
 
 
 @pytest.mark.asyncio
@@ -407,8 +407,9 @@ async def test_search_no_results(mcp_server, app, test_project):
             },
         )
 
+        # Default text format returns "No results found" when empty
         result_text = search_result.content[0].text
-        assert '"results": []' in result_text or '"results":[]' in result_text
+        assert "No results found" in result_text
 
 
 @pytest.mark.asyncio

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/api/v2/test_project_router.py

@@ -11,6 +11,21 @@ from basic_memory.schemas.project_info import ProjectItem, ProjectStatusResponse
 from basic_memory.schemas.v2 import ProjectResolveResponse
 
 
+@pytest.mark.asyncio
+async def test_list_projects(client: AsyncClient, test_project: Project, v2_projects_url):
+    """Test listing projects returns default_project from the database."""
+    response = await client.get(f"{v2_projects_url}/")
+
+    assert response.status_code == 200
+    data = response.json()
+
+    # default_project must be populated from the is_default flag in the database
+    assert data["default_project"] == test_project.name
+
+    project_names = [p["name"] for p in data["projects"]]
+    assert test_project.name in project_names
+
+
 @pytest.mark.asyncio
 async def test_get_project_by_id(client: AsyncClient, test_project: Project, v2_projects_url):
     """Test getting a project by its external_id UUID."""
@@ -361,9 +376,10 @@ async def test_legacy_v1_list_projects_endpoint(client: AsyncClient, test_projec
     assert response.status_code == 200
     data = response.json()
     assert "projects" in data
-    assert "default_project" in data
 
-    # Verify the test project is in the list
+    # default_project must be populated, not null
+    assert data["default_project"] == test_project.name
+
     project_names = [p["name"] for p in data["projects"]]
     assert test_project.name in project_names
 

tests/api/v2/test_schema_router.py

@@ -668,7 +668,8 @@ async def test_validate_reads_schema_from_file_not_database(
 
     # Overwrite the file on disk with validation=strict
     file_path = Path(file_service.base_path) / schema_entity.file_path
-    file_path.write_text(dedent("""\
+    file_path.write_text(
+        dedent("""\
         ---
         title: Editable Schema
         permalink: schemas/editable-schema
@@ -685,7 +686,8 @@ async def test_validate_reads_schema_from_file_not_database(
 
         ## Observations
         - [note] Schema that will be edited on disk
-    """))
+    """)
+    )
 
     # Create a note missing "role" — strict mode should produce errors, not warnings
     note_entity, _ = await entity_service.create_or_update_entity(
@@ -749,7 +751,8 @@ async def test_validate_falls_back_to_db_on_incomplete_frontmatter(
 
     # Overwrite file with frontmatter missing the 'schema' key
     file_path = Path(file_service.base_path) / schema_entity.file_path
-    file_path.write_text(dedent("""\
+    file_path.write_text(
+        dedent("""\
         ---
         title: Incomplete Schema
         permalink: schemas/incomplete-schema
@@ -761,7 +764,8 @@ async def test_validate_falls_back_to_db_on_incomplete_frontmatter(
 
         ## Observations
         - [note] Mid-edit state
-    """))
+    """)
+    )
 
     # Create a note to validate against this schema
     note_entity, _ = await entity_service.create_or_update_entity(

tests/cli/test_analytics.py

@@ -76,7 +76,7 @@ class TestTrack:
 
         captured_target = None
 
-        def fake_thread(target, daemon):
+        def fake_thread(target):
             nonlocal captured_target
             captured_target = target
             mock = MagicMock()
@@ -103,7 +103,7 @@ class TestTrack:
         with patch("basic_memory.cli.analytics.urllib.request.urlopen", fake_urlopen):
             with patch("basic_memory.cli.analytics.threading.Thread") as mock_thread:
                 # Capture the target function and call it directly
-                def run_target(target, daemon):
+                def run_target(target):
                     target()  # Execute synchronously
                     return MagicMock()
 
@@ -113,6 +113,7 @@ class TestTrack:
         assert captured_request is not None
         assert captured_request.full_url == "https://analytics.example.com/api/send"
         body = json.loads(captured_request.data)
+        assert body["type"] == "event"
         assert body["payload"]["name"] == "cli-cloud-login-started"
         assert body["payload"]["website"] == "test-site-id"
         assert body["payload"]["hostname"] == "cli.basicmemory.com"
@@ -129,7 +130,7 @@ class TestTrack:
         with patch("basic_memory.cli.analytics.urllib.request.urlopen", fake_urlopen):
             with patch("basic_memory.cli.analytics.threading.Thread") as mock_thread:
 
-                def run_target(target, daemon):
+                def run_target(target):
                     target()  # Should not raise
                     return MagicMock()
 

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/cli/test_cloud_status.py

@@ -63,9 +63,7 @@ def _patch_status_deps(monkeypatch, *, tokens=None, api_side_effect=None):
     monkeypatch.setattr(
         "basic_memory.cli.commands.cloud.core_commands.ConfigManager", FakeConfigManager
     )
-    monkeypatch.setattr(
-        "basic_memory.cli.commands.cloud.core_commands.CLIAuth", FakeAuth
-    )
+    monkeypatch.setattr("basic_memory.cli.commands.cloud.core_commands.CLIAuth", FakeAuth)
     monkeypatch.setattr(
         "basic_memory.cli.commands.cloud.core_commands.get_cloud_config",
         lambda: ("cid", "domain", "https://cloud.example.com"),

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,17 +31,34 @@ 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
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context._resolve_default_project_from_api",
+        _no_api_fallback,
+    )
     assert await resolve_project_parameter(project=None, allow_discovery=False) is None
 
 
 @pytest.mark.asyncio
-async def test_allows_discovery_when_enabled(config_manager):
+async def test_allows_discovery_when_enabled(config_manager, monkeypatch):
     from basic_memory.mcp.project_context import resolve_project_parameter
 
     cfg = config_manager.load_config()
     cfg.default_project = None
     config_manager.save_config(cfg)
 
+    # Prevent API fallback from returning a project via stale dependency overrides
+    async def _no_api_fallback():
+        return None
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context._resolve_default_project_from_api",
+        _no_api_fallback,
+    )
     assert await resolve_project_parameter(project=None, allow_discovery=True) is None
 
 
@@ -101,6 +118,15 @@ 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
+
+    monkeypatch.setattr(
+        "basic_memory.mcp.project_context._resolve_default_project_from_api",
+        _no_api_fallback,
+    )
     assert await resolve_project_parameter(project=None) is None
 
 

tests/mcp/test_prompt_tool_delegation.py

@@ -9,7 +9,6 @@ import pytest
 
 from basic_memory.mcp.prompts.search import search_prompt
 from basic_memory.mcp.prompts.continue_conversation import continue_conversation
-from basic_memory.schemas.search import SearchResponse, SearchResult
 
 
 # --- search_prompt ---
@@ -20,19 +19,20 @@ async def test_search_prompt_delegates_to_search_notes(monkeypatch):
     """Search prompt should call search_notes tool and wrap output."""
     captured_kwargs = {}
 
-    fake_result = SearchResponse(
-        results=[
-            SearchResult(
-                type="entity",
-                title="Test Note",
-                permalink="test-note",
-                file_path="test-note.md",
-                score=0.95,
-            )
+    # Prompts use output_format="json", so mock returns a dict
+    fake_result = {
+        "results": [
+            {
+                "type": "entity",
+                "title": "Test Note",
+                "permalink": "test-note",
+                "file_path": "test-note.md",
+                "score": 0.95,
+            }
         ],
-        current_page=1,
-        page_size=10,
-    )
+        "current_page": 1,
+        "page_size": 10,
+    }
 
     async def fake_search_notes(**kwargs):
         captured_kwargs.update(kwargs)
@@ -45,6 +45,7 @@ async def test_search_prompt_delegates_to_search_notes(monkeypatch):
     # Verify delegation
     assert captured_kwargs["query"] == "my query"
     assert captured_kwargs["after_date"] == "1w"
+    assert captured_kwargs["output_format"] == "json"
 
     # Verify output wrapping
     assert 'Search Results: "my query"' in out
@@ -55,7 +56,7 @@ async def test_search_prompt_delegates_to_search_notes(monkeypatch):
 @pytest.mark.asyncio
 async def test_search_prompt_handles_no_results(monkeypatch):
     """Search prompt should handle empty results gracefully."""
-    fake_result = SearchResponse(results=[], current_page=1, page_size=10)
+    fake_result = {"results": [], "current_page": 1, "page_size": 10}
 
     async def fake_search_notes(**kwargs):
         return fake_result
@@ -91,19 +92,20 @@ async def test_continue_conversation_delegates_to_search_notes(monkeypatch):
     """Continue conversation with topic should call search_notes."""
     captured_kwargs = {}
 
-    fake_result = SearchResponse(
-        results=[
-            SearchResult(
-                type="entity",
-                title="Previous Discussion",
-                permalink="discussions/previous",
-                file_path="discussions/previous.md",
-                score=0.9,
-            )
+    # Prompts use output_format="json", so mock returns a dict
+    fake_result = {
+        "results": [
+            {
+                "type": "entity",
+                "title": "Previous Discussion",
+                "permalink": "discussions/previous",
+                "file_path": "discussions/previous.md",
+                "score": 0.9,
+            }
         ],
-        current_page=1,
-        page_size=10,
-    )
+        "current_page": 1,
+        "page_size": 10,
+    }
 
     async def fake_search_notes(**kwargs):
         captured_kwargs.update(kwargs)
@@ -117,6 +119,7 @@ async def test_continue_conversation_delegates_to_search_notes(monkeypatch):
 
     assert captured_kwargs["query"] == "my topic"
     assert captured_kwargs["after_date"] == "3d"
+    assert captured_kwargs["output_format"] == "json"
 
     assert "'my topic'" in out
     assert "Previous Discussion" in out
@@ -164,7 +167,7 @@ async def test_continue_conversation_no_topic_default_timeframe(monkeypatch):
 @pytest.mark.asyncio
 async def test_continue_conversation_no_results_for_topic(monkeypatch):
     """Continue conversation should show capture opportunity when no results found."""
-    fake_result = SearchResponse(results=[], current_page=1, page_size=10)
+    fake_result = {"results": [], "current_page": 1, "page_size": 10}
 
     async def fake_search_notes(**kwargs):
         return fake_result

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

@@ -128,69 +128,16 @@ async def test_recent_activity_type_invalid(client, test_project, test_graph):
 
 
 @pytest.mark.asyncio
-async def test_recent_activity_discovery_mode(client, test_project, test_graph, config_manager):
-    """Test that recent_activity discovery mode works without project parameter."""
-    # Clear default_project to test discovery mode
-    cfg = config_manager.load_config()
-    cfg.default_project = None
-    config_manager.save_config(cfg)
-
-    # Test discovery mode (no project parameter)
+async def test_recent_activity_uses_default_project(client, test_project, test_graph):
+    """When no project parameter is given, recent_activity uses the default project."""
+    # Call without explicit project — should resolve to the default
     result = await recent_activity()
     assert result is not None
     assert isinstance(result, str)
 
-    # Check that we get a formatted summary
-    assert "Recent Activity Summary" in result
-    assert "Most Active Project:" in result or "Other Active Projects:" in result
-    assert "Summary:" in result
-    assert "active projects" in result
-
-    # Should contain project discovery guidance
-    assert "Suggested project:" in result or "Multiple active projects" in result
-    assert "Session reminder:" in result
-
-
-@pytest.mark.asyncio
-async def test_recent_activity_discovery_mode_no_activity(client, test_project, config_manager):
-    """If there is no activity in any project, discovery mode should say so."""
-    # Clear default_project to test discovery mode
-    cfg = config_manager.load_config()
-    cfg.default_project = None
-    config_manager.save_config(cfg)
-
-    result = await recent_activity()
-    assert "Recent Activity Summary" in result
-    assert "No recent activity found in any project." in result
-
-
-@pytest.mark.asyncio
-async def test_recent_activity_discovery_mode_multiple_active_projects(
-    app, client, test_project, tmp_path_factory, config_manager
-):
-    """Discovery mode should use the multi-project guidance when multiple projects have activity."""
-    # Clear default_project to test discovery mode
-    cfg = config_manager.load_config()
-    cfg.default_project = None
-    config_manager.save_config(cfg)
-
-    from basic_memory.mcp.tools import create_memory_project, write_note
-
-    second_root = tmp_path_factory.mktemp("second-project-home")
-
-    result = await create_memory_project(
-        project_name="second-project",
-        project_path=str(second_root),
-        set_default=False,
-    )
-    assert result.startswith("✓")
-
-    await write_note(project=test_project.name, title="One", directory="notes", content="one")
-    await write_note(project="second-project", title="Two", directory="notes", content="two")
-
-    out = await recent_activity()
-    assert "Recent Activity Summary" in out
-    assert "or would you prefer a different project" in out
+    # Should return project-specific output for the default project
+    assert "Recent Activity:" in result
+    assert "Activity Summary:" in result
 
 
 def test_recent_activity_format_relative_time_and_truncate_helpers():

tests/mcp/test_tool_search.py

@@ -5,7 +5,11 @@ from contextlib import asynccontextmanager
 from datetime import datetime, timedelta
 
 from basic_memory.mcp.tools import write_note
-from basic_memory.mcp.tools.search import search_notes, _format_search_error_response
+from basic_memory.mcp.tools.search import (
+    search_notes,
+    _format_search_error_response,
+    _format_search_markdown,
+)
 from basic_memory.schemas.search import SearchResponse
 
 
@@ -22,15 +26,18 @@ async def test_search_text(client, test_project):
     )
     assert result
 
-    # Search for it
-    response = await search_notes(project=test_project.name, query="searchable")
+    # Search for it (use json format to inspect structured results)
+    response = await search_notes(
+        project=test_project.name, query="searchable", output_format="json"
+    )
 
     # Verify results - handle both success and error cases
-    if isinstance(response, SearchResponse):
-        # Success case - verify SearchResponse
-        assert len(response.results) > 0
+    if isinstance(response, dict):
+        # Success case - verify dict response
+        assert len(response["results"]) > 0
         assert any(
-            r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results
+            r["permalink"] == f"{test_project.name}/test/test-search-note"
+            for r in response["results"]
         )
     else:
         # If search failed and returned error message, test should fail with informative message
@@ -50,21 +57,22 @@ async def test_search_title(client, test_project):
     )
     assert result
 
-    # Search for it
+    # Search for it (use json format to inspect structured results)
     response = await search_notes(
-        project=test_project.name, query="Search Note", search_type="title"
+        project=test_project.name, query="Search Note", search_type="title", output_format="json"
     )
 
     # Verify results - handle both success and error cases
-    if isinstance(response, str):
+    if isinstance(response, dict):
+        # Success case - verify dict response
+        assert len(response["results"]) > 0
+        assert any(
+            r["permalink"] == f"{test_project.name}/test/test-search-note"
+            for r in response["results"]
+        )
+    else:
         # If search failed and returned error message, test should fail with informative message
         pytest.fail(f"Search failed with error: {response}")
-    else:
-        # Success case - verify SearchResponse
-        assert len(response.results) > 0
-        assert any(
-            r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results
-        )
 
 
 @pytest.mark.asyncio
@@ -80,19 +88,21 @@ async def test_search_permalink(client, test_project):
     )
     assert result
 
-    # Search for it
+    # Search for it (use json format to inspect structured results)
     response = await search_notes(
         project=test_project.name,
         query=f"{test_project.name}/test/test-search-note",
         search_type="permalink",
+        output_format="json",
     )
 
     # Verify results - handle both success and error cases
-    if isinstance(response, SearchResponse):
-        # Success case - verify SearchResponse
-        assert len(response.results) > 0
+    if isinstance(response, dict):
+        # Success case - verify dict response
+        assert len(response["results"]) > 0
         assert any(
-            r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results
+            r["permalink"] == f"{test_project.name}/test/test-search-note"
+            for r in response["results"]
         )
     else:
         # If search failed and returned error message, test should fail with informative message
@@ -112,19 +122,21 @@ async def test_search_permalink_match(client, test_project):
     )
     assert result
 
-    # Search for it
+    # Search for it (use json format to inspect structured results)
     response = await search_notes(
         project=test_project.name,
         query=f"{test_project.name}/test/test-search-*",
         search_type="permalink",
+        output_format="json",
     )
 
     # Verify results - handle both success and error cases
-    if isinstance(response, SearchResponse):
-        # Success case - verify SearchResponse
-        assert len(response.results) > 0
+    if isinstance(response, dict):
+        # Success case - verify dict response
+        assert len(response["results"]) > 0
         assert any(
-            r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results
+            r["permalink"] == f"{test_project.name}/test/test-search-note"
+            for r in response["results"]
         )
     else:
         # If search failed and returned error message, test should fail with informative message
@@ -142,13 +154,15 @@ async def test_search_memory_url_with_project_prefix(client, test_project):
     )
     assert result
 
-    response = await search_notes(query=f"memory://{test_project.name}/test/memory-url-search-note")
+    response = await search_notes(
+        query=f"memory://{test_project.name}/test/memory-url-search-note", output_format="json"
+    )
 
-    if isinstance(response, SearchResponse):
-        assert len(response.results) > 0
+    if isinstance(response, dict):
+        assert len(response["results"]) > 0
         assert any(
-            r.permalink == f"{test_project.name}/test/memory-url-search-note"
-            for r in response.results
+            r["permalink"] == f"{test_project.name}/test/memory-url-search-note"
+            for r in response["results"]
         )
     else:
         pytest.fail(f"Search failed with error: {response}")
@@ -167,17 +181,18 @@ async def test_search_pagination(client, test_project):
     )
     assert result
 
-    # Search for it
+    # Search for it (use json format to inspect structured results)
     response = await search_notes(
-        project=test_project.name, query="searchable", page=1, page_size=1
+        project=test_project.name, query="searchable", page=1, page_size=1, output_format="json"
     )
 
     # Verify results - handle both success and error cases
-    if isinstance(response, SearchResponse):
-        # Success case - verify SearchResponse
-        assert len(response.results) == 1
+    if isinstance(response, dict):
+        # Success case - verify dict response
+        assert len(response["results"]) == 1
         assert any(
-            r.permalink == f"{test_project.name}/test/test-search-note" for r in response.results
+            r["permalink"] == f"{test_project.name}/test/test-search-note"
+            for r in response["results"]
         )
     else:
         # If search failed and returned error message, test should fail with informative message
@@ -195,13 +210,15 @@ async def test_search_with_type_filter(client, test_project):
         content="# Test\nFiltered by type",
     )
 
-    # Search with note type filter
-    response = await search_notes(project=test_project.name, query="type", note_types=["note"])
+    # Search with note type filter (use json format to inspect structured results)
+    response = await search_notes(
+        project=test_project.name, query="type", note_types=["note"], output_format="json"
+    )
 
     # Verify results - handle both success and error cases
-    if isinstance(response, SearchResponse):
+    if isinstance(response, dict):
         # Success case - verify all results are entities
-        assert all(r.type == "entity" for r in response.results)
+        assert all(r["type"] == "entity" for r in response["results"])
     else:
         # If search failed and returned error message, test should fail with informative message
         pytest.fail(f"Search failed with error: {response}")
@@ -218,13 +235,15 @@ async def test_search_with_entity_type_filter(client, test_project):
         content="# Test\nFiltered by type",
     )
 
-    # Search with entity_types (SearchItemType) filter
-    response = await search_notes(project=test_project.name, query="type", entity_types=["entity"])
+    # Search with entity_types (SearchItemType) filter (use json format)
+    response = await search_notes(
+        project=test_project.name, query="type", entity_types=["entity"], output_format="json"
+    )
 
     # Verify results - handle both success and error cases
-    if isinstance(response, SearchResponse):
+    if isinstance(response, dict):
         # Success case - verify all results are entities
-        assert all(r.type == "entity" for r in response.results)
+        assert all(r["type"] == "entity" for r in response["results"])
     else:
         # If search failed and returned error message, test should fail with informative message
         pytest.fail(f"Search failed with error: {response}")
@@ -241,16 +260,19 @@ async def test_search_with_date_filter(client, test_project):
         content="# Test\nRecent content",
     )
 
-    # Search with date filter
+    # Search with date filter (use json format to inspect structured results)
     one_hour_ago = datetime.now() - timedelta(hours=1)
     response = await search_notes(
-        project=test_project.name, query="recent", after_date=one_hour_ago.isoformat()
+        project=test_project.name,
+        query="recent",
+        after_date=one_hour_ago.isoformat(),
+        output_format="json",
     )
 
     # Verify results - handle both success and error cases
-    if isinstance(response, SearchResponse):
+    if isinstance(response, dict):
         # Success case - verify we get results within timeframe
-        assert len(response.results) > 0
+        assert len(response["results"]) > 0
     else:
         # If search failed and returned error message, test should fail with informative message
         pytest.fail(f"Search failed with error: {response}")
@@ -468,7 +490,8 @@ async def test_search_notes_sets_retrieval_mode_for_semantic_types(monkeypatch,
         search_type=search_type,
     )
 
-    assert isinstance(result, SearchResponse)
+    # Default text format returns a formatted string for empty results
+    assert isinstance(result, str)
     assert captured_payload["text"] == "semantic lookup"
     # "semantic" is an alias for "vector" retrieval mode
     expected_mode = "vector" if search_type in ("vector", "semantic") else search_type
@@ -563,7 +586,8 @@ async def test_search_notes_filter_only_metadata(monkeypatch):
         metadata_filters={"status": "in-progress"},
     )
 
-    assert isinstance(result, SearchResponse)
+    # Default text format returns a formatted string for empty results
+    assert isinstance(result, str)
     assert captured_payload["metadata_filters"] == {"status": "in-progress"}
     # No text/title/permalink should be set
     assert captured_payload.get("text") is None
@@ -605,7 +629,8 @@ async def test_search_notes_filter_only_tags(monkeypatch):
         tags=["security", "oauth"],
     )
 
-    assert isinstance(result, SearchResponse)
+    # Default text format returns a formatted string for empty results
+    assert isinstance(result, str)
     assert captured_payload["tags"] == ["security", "oauth"]
     assert captured_payload.get("text") is None
 
@@ -1158,7 +1183,8 @@ async def test_search_notes_tag_prefix_converts_to_tags_filter(monkeypatch):
         query="tag:security",
     )
 
-    assert isinstance(result, SearchResponse)
+    # Default text format returns a formatted string for empty results
+    assert isinstance(result, str)
     assert captured_payload["tags"] == ["security"]
     # No text query should be set — tag: prefix was consumed
     assert captured_payload.get("text") is None
@@ -1199,7 +1225,8 @@ async def test_search_notes_tag_prefix_merges_with_explicit_tags(monkeypatch):
         tags=["oauth"],
     )
 
-    assert isinstance(result, SearchResponse)
+    # Default text format returns a formatted string for empty results
+    assert isinstance(result, str)
     assert set(captured_payload["tags"]) == {"security", "oauth"}
     assert captured_payload.get("text") is None
 
@@ -1238,7 +1265,8 @@ async def test_search_notes_multiple_tag_prefixes(monkeypatch):
         query="tag:coffee AND tag:brewing",
     )
 
-    assert isinstance(result, SearchResponse)
+    # Default text format returns a formatted string for empty results
+    assert isinstance(result, str)
     assert set(captured_payload["tags"]) == {"coffee", "brewing"}
     # Boolean connector AND should be stripped, leaving no text query
     assert captured_payload.get("text") is None
@@ -1283,7 +1311,211 @@ async def test_search_notes_tag_prefix_with_remaining_text(monkeypatch):
         query="authentication tag:security",
     )
 
-    assert isinstance(result, SearchResponse)
+    # Default text format returns a formatted string for empty results
+    assert isinstance(result, str)
     assert captured_payload["tags"] == ["security"]
     # Remaining text should be preserved as the query
     assert captured_payload["text"] == "authentication"
+
+
+# --- Tests for text output format (#641) -----------------------------------
+
+
+def test_format_search_markdown_with_results():
+    """_format_search_markdown returns readable markdown for non-empty results."""
+    from basic_memory.schemas.search import SearchResult, SearchItemType
+
+    result = SearchResponse(
+        results=[
+            SearchResult(
+                title="My Note",
+                type=SearchItemType.ENTITY,
+                score=0.85,
+                permalink="docs/my-note",
+                file_path="docs/My Note.md",
+                matched_chunk="This is a matching snippet",
+            ),
+            SearchResult(
+                title="Other Note",
+                type=SearchItemType.ENTITY,
+                score=0.42,
+                permalink="docs/other-note",
+                file_path="docs/Other Note.md",
+            ),
+        ],
+        current_page=1,
+        page_size=10,
+    )
+
+    text = _format_search_markdown(result, "test-project", "my query")
+    assert isinstance(text, str)
+    assert "# Search Results: my query" in text
+    assert "test-project" in text
+    assert "### My Note" in text
+    assert "permalink: docs/my-note" in text
+    assert "0.8500" in text
+    assert "match: This is a matching snippet" in text
+    assert "### Other Note" in text
+    assert "2 results" in text
+    assert "page 1" in text
+
+
+def test_format_search_markdown_empty_results():
+    """_format_search_markdown returns a no-results message when results are empty."""
+    result = SearchResponse(results=[], current_page=1, page_size=10)
+    text = _format_search_markdown(result, "test-project", "missing")
+    assert isinstance(text, str)
+    assert "No results found" in text
+    assert "missing" in text
+
+
+@pytest.mark.asyncio
+async def test_search_notes_text_format_returns_string(monkeypatch):
+    """search_notes with output_format='text' returns a formatted markdown string."""
+    import importlib
+
+    from basic_memory.schemas.search import SearchResult, SearchItemType
+
+    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
+
+    class MockSearchClient:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def search(self, payload, page, page_size):
+            return SearchResponse(
+                results=[
+                    SearchResult(
+                        title="Found Note",
+                        type=SearchItemType.ENTITY,
+                        score=0.9,
+                        permalink="docs/found-note",
+                        file_path="docs/Found Note.md",
+                        matched_chunk="snippet",
+                    ),
+                ],
+                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)
+
+    result = await search_mod.search_notes(
+        project="test-project",
+        query="test",
+        output_format="text",
+    )
+
+    assert isinstance(result, str)
+    assert "# Search Results: test" in result
+    assert "### Found Note" in result
+    assert "permalink: docs/found-note" in result
+
+
+# --- Tests for metadata_filters key aliasing (#642) ----------------------------
+
+
+@pytest.mark.asyncio
+async def test_search_notes_metadata_filters_aliases_note_type(monkeypatch):
+    """metadata_filters={'note_type': 'note'} is aliased to {'type': 'note'}."""
+    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",
+        metadata_filters={"note_type": "note"},
+    )
+
+    # "note_type" should be aliased to "type" in the payload
+    assert captured_payload["metadata_filters"] == {"type": "note"}
+
+
+@pytest.mark.asyncio
+async def test_search_notes_metadata_filters_preserves_non_aliased_keys(monkeypatch):
+    """metadata_filters with non-aliased keys pass through unchanged."""
+    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",
+        metadata_filters={"note_type": "spec", "priority": "high"},
+    )
+
+    # "note_type" aliased to "type", "priority" passes through unchanged
+    assert captured_payload["metadata_filters"] == {"type": "spec", "priority": "high"}

tests/repository/test_fastembed_provider.py

@@ -19,14 +19,22 @@ class _StubVector:
 
 class _StubTextEmbedding:
     init_count = 0
+    last_init_kwargs: dict = {}
+    last_embed_kwargs: dict = {}
 
-    def __init__(self, model_name: str):
+    def __init__(self, model_name: str, cache_dir: str | None = None, threads: int | None = None):
         self.model_name = model_name
         self.embed_calls = 0
+        _StubTextEmbedding.last_init_kwargs = {
+            "model_name": model_name,
+            "cache_dir": cache_dir,
+            "threads": threads,
+        }
         _StubTextEmbedding.init_count += 1
 
-    def embed(self, texts: list[str], batch_size: int = 64):
+    def embed(self, texts: list[str], batch_size: int = 64, **kwargs):
         self.embed_calls += 1
+        _StubTextEmbedding.last_embed_kwargs = {"batch_size": batch_size, **kwargs}
         for text in texts:
             if "wide" in text:
                 yield _StubVector([1.0, 0.0, 0.0, 0.0, 0.5])
@@ -85,3 +93,58 @@ async def test_fastembed_provider_missing_dependency_raises_actionable_error(mon
         await provider.embed_query("test")
 
     assert "pip install -U basic-memory" in str(error.value)
+
+
+@pytest.mark.asyncio
+async def test_fastembed_provider_passes_runtime_knobs_to_fastembed(monkeypatch):
+    """Provider should pass optional runtime tuning knobs through to FastEmbed."""
+    module = type(sys)("fastembed")
+    module.TextEmbedding = _StubTextEmbedding
+    monkeypatch.setitem(sys.modules, "fastembed", module)
+    _StubTextEmbedding.last_init_kwargs = {}
+    _StubTextEmbedding.last_embed_kwargs = {}
+
+    provider = FastEmbedEmbeddingProvider(
+        model_name="stub-model",
+        dimensions=4,
+        batch_size=8,
+        cache_dir="/tmp/fastembed-cache",
+        threads=3,
+        parallel=2,
+    )
+    await provider.embed_documents(["runtime knobs"])
+
+    assert _StubTextEmbedding.last_init_kwargs == {
+        "model_name": "stub-model",
+        "cache_dir": "/tmp/fastembed-cache",
+        "threads": 3,
+    }
+    assert _StubTextEmbedding.last_embed_kwargs == {"batch_size": 8, "parallel": 2}
+
+
+@pytest.mark.asyncio
+async def test_fastembed_provider_parallel_one_disables_multiprocessing(monkeypatch):
+    """parallel=1 should not pass FastEmbed multiprocessing kwargs."""
+    module = type(sys)("fastembed")
+    module.TextEmbedding = _StubTextEmbedding
+    monkeypatch.setitem(sys.modules, "fastembed", module)
+    _StubTextEmbedding.last_embed_kwargs = {}
+
+    provider = FastEmbedEmbeddingProvider(model_name="stub-model", dimensions=4, parallel=1)
+    await provider.embed_documents(["parallel guardrail"])
+
+    assert _StubTextEmbedding.last_embed_kwargs == {"batch_size": 64}
+
+
+@pytest.mark.asyncio
+async def test_fastembed_provider_parallel_two_passes_multiprocessing(monkeypatch):
+    """parallel>1 should keep passing FastEmbed multiprocessing kwargs."""
+    module = type(sys)("fastembed")
+    module.TextEmbedding = _StubTextEmbedding
+    monkeypatch.setitem(sys.modules, "fastembed", module)
+    _StubTextEmbedding.last_embed_kwargs = {}
+
+    provider = FastEmbedEmbeddingProvider(model_name="stub-model", dimensions=4, parallel=2)
+    await provider.embed_documents(["parallel enabled"])
+
+    assert _StubTextEmbedding.last_embed_kwargs == {"batch_size": 64, "parallel": 2}

tests/repository/test_openai_provider.py

@@ -7,7 +7,10 @@ from types import SimpleNamespace
 import pytest
 
 from basic_memory.config import BasicMemoryConfig
-from basic_memory.repository.embedding_provider_factory import create_embedding_provider
+from basic_memory.repository.embedding_provider_factory import (
+    create_embedding_provider,
+    reset_embedding_provider_cache,
+)
 from basic_memory.repository.fastembed_provider import FastEmbedEmbeddingProvider
 from basic_memory.repository.openai_provider import OpenAIEmbeddingProvider
 from basic_memory.repository.semantic_errors import SemanticDependenciesMissingError
@@ -37,6 +40,13 @@ class _StubAsyncOpenAI:
         _StubAsyncOpenAI.init_count += 1
 
 
+@pytest.fixture(autouse=True)
+def _reset_embedding_provider_cache_fixture():
+    reset_embedding_provider_cache()
+    yield
+    reset_embedding_provider_cache()
+
+
 @pytest.mark.asyncio
 async def test_openai_provider_lazy_loads_and_reuses_client(monkeypatch):
     """Provider should instantiate AsyncOpenAI lazily and reuse a single client."""
@@ -204,3 +214,89 @@ def test_embedding_provider_factory_uses_provider_defaults_when_dimensions_not_s
     openai_provider = create_embedding_provider(openai_config)
     assert isinstance(openai_provider, OpenAIEmbeddingProvider)
     assert openai_provider.dimensions == 1536
+
+
+def test_embedding_provider_factory_forwards_fastembed_runtime_knobs():
+    """Factory should forward FastEmbed runtime tuning config fields."""
+    config = BasicMemoryConfig(
+        env="test",
+        projects={"test-project": "/tmp/basic-memory-test"},
+        default_project="test-project",
+        semantic_search_enabled=True,
+        semantic_embedding_provider="fastembed",
+        semantic_embedding_cache_dir="/tmp/fastembed-cache",
+        semantic_embedding_threads=3,
+        semantic_embedding_parallel=2,
+    )
+    provider = create_embedding_provider(config)
+    assert isinstance(provider, FastEmbedEmbeddingProvider)
+    assert provider.cache_dir == "/tmp/fastembed-cache"
+    assert provider.threads == 3
+    assert provider.parallel == 2
+
+
+def test_embedding_provider_factory_reuses_provider_for_same_cache_key():
+    """Factory should reuse the same provider instance for identical config values."""
+    config_a = BasicMemoryConfig(
+        env="test",
+        projects={"test-project": "/tmp/basic-memory-test"},
+        default_project="test-project",
+        semantic_search_enabled=True,
+        semantic_embedding_provider="fastembed",
+        semantic_embedding_threads=2,
+    )
+    config_b = BasicMemoryConfig(
+        env="test",
+        projects={"test-project": "/tmp/basic-memory-test"},
+        default_project="test-project",
+        semantic_search_enabled=True,
+        semantic_embedding_provider="fastembed",
+        semantic_embedding_threads=2,
+    )
+
+    provider_a = create_embedding_provider(config_a)
+    provider_b = create_embedding_provider(config_b)
+
+    assert provider_a is provider_b
+
+
+def test_embedding_provider_factory_creates_new_provider_for_different_cache_key():
+    """Factory should create distinct providers when cache key fields differ."""
+    config_a = BasicMemoryConfig(
+        env="test",
+        projects={"test-project": "/tmp/basic-memory-test"},
+        default_project="test-project",
+        semantic_search_enabled=True,
+        semantic_embedding_provider="fastembed",
+        semantic_embedding_threads=2,
+    )
+    config_b = BasicMemoryConfig(
+        env="test",
+        projects={"test-project": "/tmp/basic-memory-test"},
+        default_project="test-project",
+        semantic_search_enabled=True,
+        semantic_embedding_provider="fastembed",
+        semantic_embedding_threads=4,
+    )
+
+    provider_a = create_embedding_provider(config_a)
+    provider_b = create_embedding_provider(config_b)
+
+    assert provider_a is not provider_b
+
+
+def test_embedding_provider_factory_reset_clears_cache():
+    """Cache reset helper should force provider recreation for the same config."""
+    config = BasicMemoryConfig(
+        env="test",
+        projects={"test-project": "/tmp/basic-memory-test"},
+        default_project="test-project",
+        semantic_search_enabled=True,
+        semantic_embedding_provider="fastembed",
+    )
+
+    provider_first = create_embedding_provider(config)
+    reset_embedding_provider_cache()
+    provider_second = create_embedding_provider(config)
+
+    assert provider_first is not provider_second

tests/repository/test_semantic_search_base.py

@@ -11,6 +11,7 @@ import pytest
 from basic_memory.repository.search_repository_base import (
     MAX_VECTOR_CHUNK_CHARS,
     SearchRepositoryBase,
+    _PreparedEntityVectorSync,
 )
 from basic_memory.repository.sqlite_search_repository import SQLiteSearchRepository
 from basic_memory.repository.semantic_errors import SemanticSearchDisabledError
@@ -50,6 +51,7 @@ class _ConcreteRepo(SearchRepositoryBase):
     _semantic_enabled = False
     _semantic_vector_k = 100
     _embedding_provider = None
+    _semantic_embedding_sync_batch_size = 64
     _vector_dimensions = 4
     _vector_tables_initialized = False
 
@@ -269,3 +271,82 @@ async def test_sqlite_hybrid_search_raises_disabled_error(search_repository):
             limit=5,
             offset=0,
         )
+
+
+@pytest.mark.asyncio
+async def test_sync_entity_vectors_batch_flushes_at_configured_threshold(monkeypatch):
+    """Batch sync should flush queued jobs at semantic_embedding_sync_batch_size boundaries."""
+    repo = _ConcreteRepo()
+    repo._semantic_enabled = True
+    repo._embedding_provider = object()
+    repo._semantic_embedding_sync_batch_size = 2
+
+    prepared_by_entity = {
+        1: _PreparedEntityVectorSync(1, 1.0, 1, [(101, "chunk-1")]),
+        2: _PreparedEntityVectorSync(2, 2.0, 1, [(102, "chunk-2")]),
+        3: _PreparedEntityVectorSync(3, 3.0, 1, [(103, "chunk-3")]),
+    }
+    flush_sizes: list[int] = []
+
+    async def _stub_prepare(entity_id: int) -> _PreparedEntityVectorSync:
+        return prepared_by_entity[entity_id]
+
+    async def _stub_flush(flush_jobs, entity_runtime, synced_entity_ids):
+        flush_sizes.append(len(flush_jobs))
+        for job in flush_jobs:
+            runtime = entity_runtime[job.entity_id]
+            runtime.remaining_jobs -= 1
+            if runtime.remaining_jobs <= 0:
+                synced_entity_ids.add(job.entity_id)
+                entity_runtime.pop(job.entity_id, None)
+        return (0.1, 0.2)
+
+    monkeypatch.setattr(repo, "_prepare_entity_vector_jobs", _stub_prepare)
+    monkeypatch.setattr(repo, "_flush_embedding_jobs", _stub_flush)
+
+    result = await repo.sync_entity_vectors_batch([1, 2, 3])
+
+    assert flush_sizes == [2, 1]
+    assert result.entities_total == 3
+    assert result.entities_synced == 3
+    assert result.entities_failed == 0
+    assert result.failed_entity_ids == []
+    assert result.embedding_jobs_total == 3
+    assert result.embed_seconds_total == pytest.approx(0.2)
+    assert result.write_seconds_total == pytest.approx(0.4)
+
+
+@pytest.mark.asyncio
+async def test_sync_entity_vectors_batch_continue_on_error(monkeypatch):
+    """Batch sync should continue after per-entity and per-flush failures."""
+    repo = _ConcreteRepo()
+    repo._semantic_enabled = True
+    repo._embedding_provider = object()
+    repo._semantic_embedding_sync_batch_size = 1
+
+    async def _stub_prepare(entity_id: int) -> _PreparedEntityVectorSync:
+        if entity_id == 2:
+            raise RuntimeError("prepare failed")
+        return _PreparedEntityVectorSync(
+            entity_id, float(entity_id), 1, [(100 + entity_id, "chunk")]
+        )
+
+    async def _stub_flush(flush_jobs, entity_runtime, synced_entity_ids):
+        entity_id = flush_jobs[0].entity_id
+        if entity_id == 3:
+            raise RuntimeError("flush failed")
+        runtime = entity_runtime[entity_id]
+        runtime.remaining_jobs = 0
+        synced_entity_ids.add(entity_id)
+        entity_runtime.pop(entity_id, None)
+        return (0.05, 0.05)
+
+    monkeypatch.setattr(repo, "_prepare_entity_vector_jobs", _stub_prepare)
+    monkeypatch.setattr(repo, "_flush_embedding_jobs", _stub_flush)
+
+    result = await repo.sync_entity_vectors_batch([1, 2, 3])
+
+    assert result.entities_total == 3
+    assert result.entities_synced == 1
+    assert result.entities_failed == 2
+    assert result.failed_entity_ids == [2, 3]

tests/services/test_initialization.py

@@ -319,8 +319,20 @@ async def test_semantic_embedding_backfill_syncs_each_entity(
         def __init__(self, _session_maker, project_id: int, app_config=None):
             self.project_id = project_id
 
-        async def sync_entity_vectors(self, entity_id: int) -> None:
-            synced_pairs.append((self.project_id, entity_id))
+        async def sync_entity_vectors_batch(self, entity_ids: list[int], progress_callback=None):
+            for entity_id in entity_ids:
+                synced_pairs.append((self.project_id, entity_id))
+            from basic_memory.repository.search_repository_base import VectorSyncBatchResult
+
+            return VectorSyncBatchResult(
+                entities_total=len(entity_ids),
+                entities_synced=len(entity_ids),
+                entities_failed=0,
+                failed_entity_ids=[],
+                embedding_jobs_total=0,
+                embed_seconds_total=0.0,
+                write_seconds_total=0.0,
+            )
 
     monkeypatch.setattr("basic_memory.db.SQLiteSearchRepository", StubSearchRepository)
     monkeypatch.setattr("basic_memory.db.PostgresSearchRepository", StubSearchRepository)
@@ -347,8 +359,18 @@ async def test_semantic_embedding_backfill_skips_when_semantic_disabled(
             nonlocal called
             called = True
 
-        async def sync_entity_vectors(self, entity_id: int) -> None:  # pragma: no cover
-            return None
+        async def sync_entity_vectors_batch(self, entity_ids: list[int], progress_callback=None):
+            from basic_memory.repository.search_repository_base import VectorSyncBatchResult
+
+            return VectorSyncBatchResult(
+                entities_total=len(entity_ids),
+                entities_synced=len(entity_ids),
+                entities_failed=0,
+                failed_entity_ids=[],
+                embedding_jobs_total=0,
+                embed_seconds_total=0.0,
+                write_seconds_total=0.0,
+            )
 
     monkeypatch.setattr("basic_memory.db.SQLiteSearchRepository", StubSearchRepository)
     monkeypatch.setattr("basic_memory.db.PostgresSearchRepository", StubSearchRepository)

tests/services/test_link_resolver.py

@@ -901,3 +901,21 @@ async def test_resolve_link_non_uuid_falls_through(link_resolver, test_entities,
     result = await link_resolver.resolve_link("Core Service")
     assert result is not None
     assert result.permalink == f"{project_prefix}/components/core-service"
+
+
+# ============================================================================
+# Fuzzy search best-match selection tests (#640)
+# ============================================================================
+
+
+@pytest.mark.asyncio
+async def test_fuzzy_search_selects_first_result(link_resolver, project_prefix):
+    """Test that fuzzy search uses results[0] (best-ranked by the DB) regardless of score sign.
+
+    Both SQLite (BM25, negative scores, ASC) and Postgres (ts_rank, positive scores, DESC)
+    return the best match first. Using results[0] is backend-agnostic and correct.
+    """
+    result = await link_resolver.resolve_link("Auth Serv")
+    assert result is not None
+    # The best match for "Auth Serv" should be Auth Service
+    assert result.permalink == f"{project_prefix}/components/auth-service"

tests/services/test_search_service.py

@@ -1172,14 +1172,16 @@ async def test_index_entity_markdown_strips_nul_bytes(search_service, session_ma
 
 
 @pytest.mark.asyncio
-async def test_reindex_vectors(search_service, session_maker, test_project):
+async def test_reindex_vectors(search_service, session_maker, test_project, monkeypatch):
     """Test that reindex_vectors processes all entities and reports stats."""
     from basic_memory.repository import EntityRepository
+    from basic_memory.repository.search_repository_base import VectorSyncBatchResult
     from datetime import datetime
 
     entity_repo = EntityRepository(session_maker, project_id=test_project.id)
 
     # Create some entities
+    created_entity_ids: list[int] = []
     for i in range(3):
         entity = await entity_repo.create(
             {
@@ -1194,8 +1196,30 @@ async def test_reindex_vectors(search_service, session_maker, test_project):
                 "updated_at": datetime.now(),
             }
         )
+        created_entity_ids.append(entity.id)
         await search_service.index_entity(entity, content=f"Content for entity {i}")
 
+    async def _stub_sync_entity_vectors_batch(entity_ids: list[int], progress_callback=None):
+        assert entity_ids == created_entity_ids
+        if progress_callback:
+            for i, entity_id in enumerate(entity_ids):
+                progress_callback(entity_id, i, len(entity_ids))
+        return VectorSyncBatchResult(
+            entities_total=len(entity_ids),
+            entities_synced=len(entity_ids),
+            entities_failed=0,
+            failed_entity_ids=[],
+            embedding_jobs_total=9,
+            embed_seconds_total=1.2,
+            write_seconds_total=0.4,
+        )
+
+    monkeypatch.setattr(
+        search_service.repository,
+        "sync_entity_vectors_batch",
+        _stub_sync_entity_vectors_batch,
+    )
+
     # Track progress calls
     progress_calls = []
 
@@ -1217,9 +1241,12 @@ async def test_reindex_vectors(search_service, session_maker, test_project):
 
 
 @pytest.mark.asyncio
-async def test_reindex_vectors_no_callback(search_service, session_maker, test_project):
+async def test_reindex_vectors_no_callback(
+    search_service, session_maker, test_project, monkeypatch
+):
     """Test reindex_vectors works without a progress callback."""
     from basic_memory.repository import EntityRepository
+    from basic_memory.repository.search_repository_base import VectorSyncBatchResult
     from datetime import datetime
 
     entity_repo = EntityRepository(session_maker, project_id=test_project.id)
@@ -1238,6 +1265,24 @@ async def test_reindex_vectors_no_callback(search_service, session_maker, test_p
     )
     await search_service.index_entity(entity, content="Test content")
 
+    async def _stub_sync_entity_vectors_batch(entity_ids: list[int], progress_callback=None):
+        assert progress_callback is None
+        return VectorSyncBatchResult(
+            entities_total=len(entity_ids),
+            entities_synced=len(entity_ids),
+            entities_failed=0,
+            failed_entity_ids=[],
+            embedding_jobs_total=3,
+            embed_seconds_total=0.5,
+            write_seconds_total=0.1,
+        )
+
+    monkeypatch.setattr(
+        search_service.repository,
+        "sync_entity_vectors_batch",
+        _stub_sync_entity_vectors_batch,
+    )
+
     stats = await search_service.reindex_vectors()
     assert stats["total_entities"] >= 1
     assert stats["embedded"] + stats["errors"] == stats["total_entities"]

tests/test_config.py

@@ -624,7 +624,6 @@ class TestConfigManager:
             raw = json.loads(config_manager.config_file.read_text(encoding="utf-8"))
             assert "cloud_mode" not in raw
 
-
     def test_migration_creates_backup_of_old_config(self):
         """Config migration should create a .bak backup before overwriting."""
         with tempfile.TemporaryDirectory() as temp_dir:
@@ -671,9 +670,7 @@ class TestConfigManager:
             # Write config in the current ProjectEntry format — no migration needed
             current_config_data = {
                 "env": "dev",
-                "projects": {
-                    "main": {"path": str(temp_path / "main"), "mode": "local"}
-                },
+                "projects": {"main": {"path": str(temp_path / "main"), "mode": "local"}},
                 "default_project": "main",
             }
             config_manager.config_file.write_text(json.dumps(current_config_data, indent=2))