feat: improve cloud promo copy and split init line from upsell panel

- Split "Basic Memory initialized" into a pre-command plain line (shown once on first run)
- Rewrite cloud upsell with benefit-led copy: pain point → solution → CTA
- Use package version instead of manual CLOUD_PROMO_VERSION for promo gating
- Set discount code to BMFOSS
- Add "Learn more" link and dimmed opt-out hint below panel
- Remove cloud emoji from panel title to fix box alignment
- Collapse two message builders into single _build_cloud_promo_message()
- Update tests for new behavior (15 tests, all passing)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

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

@@ -10,10 +10,7 @@
 
 from fastapi import APIRouter, Path, Query
 
-from basic_memory.deps import (
-    SearchServiceV2ExternalDep,
-    EntityRepositoryV2ExternalDep,
-)
+from basic_memory.deps import EntityRepositoryV2ExternalDep
 from basic_memory.models.knowledge import Entity
 from basic_memory.schemas.schema import (
     ValidationReport,
@@ -24,7 +21,6 @@
     FieldFrequencyResponse,
     DriftFieldResponse,
 )
-from basic_memory.schemas.search import SearchQuery
 from basic_memory.schema.resolver import resolve_schema
 from basic_memory.schema.validator import validate_note
 from basic_memory.schema.inference import infer_schema, NoteData, ObservationData, RelationData
@@ -81,7 +77,6 @@ def _entity_frontmatter(entity: Entity) -> dict:
 @router.post("/schema/validate", response_model=ValidationReport)
 async def validate_schema(
     entity_repository: EntityRepositoryV2ExternalDep,
-    search_service: SearchServiceV2ExternalDep,
     project_id: str = Path(..., description="Project external UUID"),
     entity_type: str | None = Query(None, description="Entity type to validate"),
     identifier: str | None = Query(None, description="Specific note identifier"),
@@ -93,18 +88,12 @@ async def validate_schema(
     """
     results: list[NoteValidationResponse] = []
 
-    async def search_fn(query: str) -> list:
-        # Search for schema notes, then load full entity_metadata from the entity table.
-        # The search index only stores minimal metadata (e.g., {"entity_type": "schema"}),
-        # but parse_schema_note needs the full frontmatter with entity/schema/version keys.
-        results = await search_service.search(SearchQuery(text=query, types=["schema"]), limit=5)
-        frontmatters = []
-        for row in results:
-            if row.permalink:
-                entity = await entity_repository.get_by_permalink(row.permalink)
-                if entity:
-                    frontmatters.append(_entity_frontmatter(entity))
-        return frontmatters
+    async def search_fn(query: str) -> list[dict]:
+        # Trigger: resolve_schema passes the entity type name as query
+        # Why: direct metadata match avoids fragile text search that fails when
+        #      schema title doesn't match entity type (e.g. "Strict Person" vs "strictperson")
+        entities = await _find_schema_entities(entity_repository, query)
+        return [_entity_frontmatter(e) for e in entities]
 
     # --- Single note validation ---
     if identifier:
@@ -205,7 +194,6 @@ async def infer_schema_endpoint(
 @router.get("/schema/diff/{entity_type}", response_model=DriftReport)
 async def diff_schema_endpoint(
     entity_repository: EntityRepositoryV2ExternalDep,
-    search_service: SearchServiceV2ExternalDep,
     entity_type: str = Path(..., description="Entity type to check for drift"),
     project_id: str = Path(..., description="Project external UUID"),
 ):
@@ -216,25 +204,16 @@ async def diff_schema_endpoint(
     fields, and cardinality changes.
     """
 
-    async def search_fn(query: str) -> list:
-        # Search for schema notes, then load full entity_metadata from the entity table.
-        # The search index only stores minimal metadata (e.g., {"entity_type": "schema"}),
-        # but parse_schema_note needs the full frontmatter with entity/schema/version keys.
-        results = await search_service.search(SearchQuery(text=query, types=["schema"]), limit=5)
-        frontmatters = []
-        for row in results:
-            if row.permalink:
-                entity = await entity_repository.get_by_permalink(row.permalink)
-                if entity:
-                    frontmatters.append(_entity_frontmatter(entity))
-        return frontmatters
+    async def search_fn(query: str) -> list[dict]:
+        entities = await _find_schema_entities(entity_repository, query)
+        return [_entity_frontmatter(e) for e in entities]
 
     # Resolve schema by entity type
     schema_frontmatter = {"type": entity_type}
     schema_def = await resolve_schema(schema_frontmatter, search_fn)
 
     if not schema_def:
-        return DriftReport(entity_type=entity_type)
+        return DriftReport(entity_type=entity_type, schema_found=False)
 
     # Collect all notes of this type
     entities = await _find_by_entity_type(entity_repository, entity_type)
@@ -281,6 +260,26 @@ async def _find_by_entity_type(
     return list(result.scalars().all())
 
 
+async def _find_schema_entities(
+    entity_repository: EntityRepositoryV2ExternalDep,
+    target_entity_type: str,
+) -> list[Entity]:
+    """Find schema entities whose entity_metadata['entity'] matches the target type.
+
+    Queries all schema-type entities, then filters by metadata in Python.
+    JSON column filtering syntax varies across SQLite/Postgres; an in-memory
+    filter on the small set of schema notes is simple and portable.
+    """
+    query = entity_repository.select().where(Entity.entity_type == "schema")
+    result = await entity_repository.execute_query(query)
+    entities = list(result.scalars().all())
+    return [
+        e
+        for e in entities
+        if e.entity_metadata and e.entity_metadata.get("entity") == target_entity_type
+    ]
+
+
 def _to_note_validation_response(result) -> NoteValidationResponse:
     """Convert a core ValidationResult to a Pydantic response model."""
     return NoteValidationResponse(

src/basic_memory/cli/app.py

@@ -9,7 +9,7 @@
 import typer  # noqa: E402
 
 from basic_memory.cli.container import CliContainer, set_container  # noqa: E402
-from basic_memory.cli.promo import maybe_show_cloud_promo  # noqa: E402
+from basic_memory.cli.promo import maybe_show_cloud_promo, maybe_show_init_line  # noqa: E402
 from basic_memory.config import init_cli_logging  # noqa: E402
 
 
@@ -47,7 +47,15 @@ def app_callback(
     container = CliContainer.create()
     set_container(container)
 
-    maybe_show_cloud_promo(ctx.invoked_subcommand)
+    # Trigger: first-run init confirmation before command output.
+    # Why: informational "initialized" message belongs above command results, not in the upsell panel.
+    # Outcome: one-time plain line printed before the subcommand runs.
+    maybe_show_init_line(ctx.invoked_subcommand)
+
+    # Trigger: register promo as a post-command callback.
+    # Why: promo output should appear after the command's own output, not before.
+    # Outcome: promo panel renders below the command results (status tree, table, etc.).
+    ctx.call_on_close(lambda: maybe_show_cloud_promo(ctx.invoked_subcommand))
 
     # Run initialization for commands that don't use the API
     # Skip for 'mcp' command - it has its own lifespan that handles initialization

src/basic_memory/cli/commands/command_utils.py

@@ -9,8 +9,8 @@
 from rich.console import Console
 
 from basic_memory import db
+from basic_memory.config import ConfigManager
 from basic_memory.mcp.async_client import get_client
-
 from basic_memory.mcp.tools.utils import call_post, call_get
 from basic_memory.mcp.project_context import get_active_project
 from basic_memory.schemas import ProjectInfoResponse
@@ -55,8 +55,11 @@ async def run_sync(
         run_in_background: If True, return immediately; if False, wait for completion
     """
 
+    # Resolve default project so get_client() can route per-project
+    project = project or ConfigManager().default_project
+
     try:
-        async with get_client() as client:
+        async with get_client(project_name=project) as client:
             project_item = await get_active_project(client, project, None)
             url = f"/v2/projects/{project_item.external_id}/sync"
             params = []
@@ -88,9 +91,8 @@ async def run_sync(
 
 async def get_project_info(project: str):
     """Get project information via API endpoint."""
-
     try:
-        async with get_client() as client:
+        async with get_client(project_name=project) as client:
             project_item = await get_active_project(client, project, None)
             response = await call_get(client, f"/v2/projects/{project_item.external_id}/info")
             return ProjectInfoResponse.model_validate(response.json())

src/basic_memory/cli/commands/project.py

@@ -79,34 +79,38 @@ async def _list_projects():
         table.add_column("Path", style="green")
         table.add_column("Mode", style="blue")
 
-        # Add Local Path column if in cloud mode and not forcing local
+        # Add cloud-specific columns when in cloud mode
         if config.cloud_mode_enabled and not local:
             table.add_column("Local Path", style="yellow", no_wrap=True, overflow="fold")
+            table.add_column("Sync", style="green")
 
-        # Show Default column in local mode or if default_project_mode is enabled in cloud mode
-        show_default_column = local or not config.cloud_mode_enabled or config.default_project_mode
-        if show_default_column:
-            table.add_column("Default", style="magenta")
+        table.add_column("Default", style="magenta")
 
         for project in result.projects:
             is_default = "[X]" if project.is_default else ""
             normalized_path = normalize_project_path(project.path)
-            project_mode = config.get_project_mode(project.name).value
+            # Trigger: cloud mode and project not in local config
+            # Why: cloud-discovered projects default to LOCAL in get_project_mode
+            # Outcome: show "cloud" for projects only known to the cloud API
+            entry = config.projects.get(project.name)
+            if config.cloud_mode_enabled and not local and entry is None:
+                project_mode = ProjectMode.CLOUD.value
+            else:
+                project_mode = config.get_project_mode(project.name).value
 
             # Build row based on mode
             row = [project.name, format_path(normalized_path), project_mode]
 
-            # Add local path if in cloud mode and not forcing local
+            # Add cloud-specific columns
             if config.cloud_mode_enabled and not local:
                 local_path = ""
-                entry = config.projects.get(project.name)
-                if entry and entry.cloud_sync_path:
-                    local_path = format_path(entry.cloud_sync_path)
+                if entry:
+                    local_path = format_path(entry.cloud_sync_path or entry.path)
                 row.append(local_path)
+                has_sync = "[X]" if entry and entry.cloud_sync_path else ""
+                row.append(has_sync)
 
-            # Add default indicator if showing default column
-            if show_default_column:
-                row.append(is_default)
+            row.append(is_default)
 
             table.add_row(*row)
 
@@ -379,7 +383,7 @@ def set_default_project(
         False, "--local", help="Force local API routing (required in cloud mode)"
     ),
 ) -> None:
-    """Set the default project when 'config.default_project_mode' is set.
+    """Set the default project used as fallback when no project is specified.
 
     In cloud mode, use --local to modify the local configuration.
     """

src/basic_memory/cli/commands/routing.py

@@ -9,6 +9,8 @@
 
 The routing is controlled via environment variables:
 - BASIC_MEMORY_FORCE_LOCAL: When "true", forces local ASGI transport
+- BASIC_MEMORY_EXPLICIT_ROUTING: When "true", signals that --local/--cloud
+  was explicitly passed, overriding per-project routing in get_client()
 - These are checked in basic_memory.mcp.async_client.get_client()
 """
 
@@ -24,6 +26,11 @@ def force_routing(local: bool = False, cloud: bool = False) -> Generator[None, N
     Sets environment variables that are checked by get_client() to determine
     whether to use local ASGI transport or cloud proxy transport.
 
+    When either flag is set, BASIC_MEMORY_EXPLICIT_ROUTING is also set so
+    that get_client() skips per-project routing and honors the flag directly.
+    This only affects CLI commands — the MCP server sets FORCE_LOCAL directly
+    (without EXPLICIT_ROUTING), so per-project routing still works for MCP tools.
+
     Args:
         local: If True, force local ASGI transport (ignores cloud_mode_enabled)
         cloud: If True, clear force_local to allow cloud routing
@@ -41,23 +48,30 @@ def force_routing(local: bool = False, cloud: bool = False) -> Generator[None, N
 
     # Save original values
     original_force_local = os.environ.get("BASIC_MEMORY_FORCE_LOCAL")
+    original_explicit = os.environ.get("BASIC_MEMORY_EXPLICIT_ROUTING")
 
     try:
         if local:
-            # Force local routing by setting the env var
             os.environ["BASIC_MEMORY_FORCE_LOCAL"] = "true"
+            os.environ["BASIC_MEMORY_EXPLICIT_ROUTING"] = "true"
         elif cloud:
             # Ensure force_local is NOT set, let cloud_mode_enabled take effect
             os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None)
+            os.environ["BASIC_MEMORY_EXPLICIT_ROUTING"] = "true"
         # If neither is set, don't change anything (use default behavior)
         yield
     finally:
-        # Restore original value
+        # Restore original values
         if original_force_local is None:
             os.environ.pop("BASIC_MEMORY_FORCE_LOCAL", None)
         else:
             os.environ["BASIC_MEMORY_FORCE_LOCAL"] = original_force_local
 
+        if original_explicit is None:
+            os.environ.pop("BASIC_MEMORY_EXPLICIT_ROUTING", None)
+        else:
+            os.environ["BASIC_MEMORY_EXPLICIT_ROUTING"] = original_explicit
+
 
 def validate_routing_flags(local: bool, cloud: bool) -> None:
     """Validate that --local and --cloud flags are not both specified.