fix(cli): resolve workspace slugs and names for bm cloud share --workspace

Live prod QA found `bm cloud share ... --workspace <slug>` failed with an
opaque 400: resolve_configured_workspace returns the explicit --workspace
value verbatim, but the cloud's X-Workspace-ID resolver only accepts a
workspace/tenant UUID. Users see slugs and display names (list-workspaces
output, memory:// URLs), so the natural input never routed.

The share commands now resolve the workspace header client-side: a UUID is
forwarded verbatim (covers per-project config workspace_id and the default
chain, zero extra API calls), while any other value is treated as a human
identifier and mapped to the tenant UUID via a single get_available_workspaces
lookup, matching with slug > tenant_id > name precedence (mirroring #979).
Ambiguous and unknown identifiers fail fast with errors that name the
candidate / available workspace slugs, and the list command now re-raises
typer.Exit ahead of its broad handler so the resolution errors aren't
re-wrapped as "Unexpected error".

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: Drew Cain <groksrc@gmail.com>

src/basic_memory/cli/commands/cloud/__init__.py

@@ -11,9 +11,11 @@ from basic_memory.cli.commands.cloud.project_sync import *  # noqa: F401,F403
 # Register snapshot sub-command group
 from basic_memory.cli.commands.cloud.snapshot import snapshot_app
 from basic_memory.cli.commands.cloud.workspace import workspace_app
+from basic_memory.cli.commands.cloud.shares import share_app
 
 cloud_app.add_typer(snapshot_app, name="snapshot")
 cloud_app.add_typer(workspace_app, name="workspace")
+cloud_app.add_typer(share_app, name="share")
 
 # Register restore command (directly on cloud_app via decorator)
 from basic_memory.cli.commands.cloud.restore import restore  # noqa: F401, E402

src/basic_memory/cli/commands/cloud/shares.py

@@ -0,0 +1,533 @@
+"""Public share CLI commands for Basic Memory Cloud.
+
+Surfaces the cloud `/api/shares` endpoints so users can manage public share
+links for notes without leaving the terminal:
+
+- POST   /api/shares          -> create
+- GET    /api/shares          -> list
+- PATCH  /api/shares/{token}   -> update (enable/disable, set expiration)
+- DELETE /api/shares/{token}   -> revoke
+
+Auth, config lookup, and error handling reuse the shared `make_api_request()`
+helper, matching the `snapshot.py` command group.
+"""
+
+import asyncio
+from datetime import datetime
+from typing import Optional
+from urllib.parse import urlencode
+from uuid import UUID
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from basic_memory.cli.commands.cloud.api_client import (
+    CloudAPIError,
+    SubscriptionRequiredError,
+    make_api_request,
+)
+from basic_memory.config import ConfigManager
+from basic_memory.mcp.async_client import resolve_configured_workspace
+from basic_memory.schemas.cloud import WorkspaceInfo
+
+console = Console()
+share_app = typer.Typer(help="Manage public share links for notes")
+
+# Header the cloud uses to route a request to a specific tenant's workspace.
+# Mirrors basic_memory.cli.commands.cloud.cloud_utils._workspace_headers: the
+# cloud /api/shares endpoints resolve the workspace from X-Workspace-ID (see
+# resolve_workspace in basic-memory-cloud deps.py), so without it a team
+# workspace project would be evaluated against the caller's default tenant.
+WORKSPACE_ID_HEADER = "X-Workspace-ID"
+
+
+def _is_uuid(value: str) -> bool:
+    """Return True when value parses as a UUID in any standard textual form."""
+    try:
+        UUID(value)
+    except ValueError:
+        return False
+    return True
+
+
+def _match_workspace_identifier(
+    workspaces: list[WorkspaceInfo], identifier: str
+) -> Optional[WorkspaceInfo]:
+    """Match a human workspace identifier with slug > tenant_id > name precedence.
+
+    Mirrors project_context._match_workspace_identifier (PR #979): try the stable
+    slug first (case-insensitive), then the exact tenant_id, then the display name
+    (case-insensitive). The first tier that yields any match wins, so a display
+    name colliding with another workspace's slug never shadows the slug match.
+    """
+    slug_matches = [ws for ws in workspaces if ws.slug.casefold() == identifier.casefold()]
+    if slug_matches:
+        return slug_matches[0] if len(slug_matches) == 1 else _ambiguous(slug_matches, identifier)
+
+    tenant_matches = [ws for ws in workspaces if ws.tenant_id == identifier]
+    if tenant_matches:
+        return tenant_matches[0]
+
+    name_matches = [ws for ws in workspaces if ws.name.casefold() == identifier.casefold()]
+    if name_matches:
+        return name_matches[0] if len(name_matches) == 1 else _ambiguous(name_matches, identifier)
+
+    return None
+
+
+def _ambiguous(matches: list[WorkspaceInfo], identifier: str) -> WorkspaceInfo:
+    """Fail with a clear, copyable error when an identifier matches >1 workspace.
+
+    Trigger: a display name (or slug, defensively) resolves to multiple workspaces.
+    Why: silently picking one would route a share to the wrong tenant.
+    Outcome: list candidate slugs/tenant_ids and exit non-zero so the user re-runs
+        with an unambiguous slug or tenant_id.
+    """
+    candidates = "\n".join(f"  - {ws.slug} (tenant_id: {ws.tenant_id})" for ws in matches)
+    console.print(
+        f"[red]Workspace '{identifier}' is ambiguous; it matches multiple workspaces.[/red]\n"
+        "[yellow]Re-run with a unique workspace slug or tenant_id:[/yellow]\n"
+        f"{candidates}"
+    )
+    raise typer.Exit(1)
+
+
+async def _resolve_workspace_to_tenant_id(identifier: str) -> str:
+    """Resolve a human workspace identifier (slug/name/tenant_id) to a tenant UUID.
+
+    Constraint: the cloud's X-Workspace-ID resolver only accepts a workspace/tenant
+    UUID, but users see slugs and display names (in list-workspaces output and
+    memory:// URLs). So when --workspace (or the configured default) is not already
+    a UUID, fetch the caller's workspaces once and map it to the tenant_id here.
+
+    get_available_workspaces is the same workspace-fetch seam project.py uses for
+    CLI workspace resolution; it is awaited directly here because the share
+    commands already run inside their own event loop.
+    """
+    from basic_memory.mcp.project_context import get_available_workspaces
+
+    workspaces = await get_available_workspaces()
+    match = _match_workspace_identifier(workspaces, identifier)
+    if match is None:
+        available = "\n".join(f"  - {ws.slug}" for ws in workspaces)
+        console.print(
+            f"[red]Workspace '{identifier}' was not found.[/red]\n"
+            "[yellow]Use one of these workspace slugs (or a tenant_id):[/yellow]\n"
+            f"{available}"
+        )
+        raise typer.Exit(1)
+    return match.tenant_id
+
+
+async def _workspace_headers(
+    *,
+    project_name: Optional[str] = None,
+    workspace: Optional[str] = None,
+) -> dict[str, str]:
+    """Resolve the target workspace and build the routing header, if any.
+
+    Resolution chain (see resolve_configured_workspace): explicit --workspace,
+    then the project's configured workspace_id, then the global default. Returns
+    an empty dict when nothing resolves so the request falls back to the
+    caller's default tenant exactly as before.
+
+    The cloud's X-Workspace-ID resolver only accepts a workspace/tenant UUID. A
+    UUID is forwarded verbatim (covers per-project config workspace_id values and
+    the default chain, zero extra API calls); any other value is treated as a
+    human identifier and resolved to the tenant UUID via one workspace lookup.
+    """
+    resolved = resolve_configured_workspace(project_name=project_name, workspace=workspace)
+    if resolved is None:
+        return {}
+    if _is_uuid(resolved):
+        return {WORKSPACE_ID_HEADER: resolved}
+    tenant_id = await _resolve_workspace_to_tenant_id(resolved)
+    return {WORKSPACE_ID_HEADER: tenant_id}
+
+
+def _format_timestamp(iso_timestamp: Optional[str]) -> str:
+    """Format an ISO timestamp to a human-readable form, or '-' when absent."""
+    if not iso_timestamp:
+        return "-"
+    try:
+        dt = datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00"))
+        return dt.strftime("%Y-%m-%d %H:%M:%S")
+    except (ValueError, AttributeError):
+        return iso_timestamp
+
+
+def _parse_expires_at(value: str) -> str:
+    """Validate an --expires-at value and normalize it to an ISO 8601 string.
+
+    Accepts either a full ISO timestamp ("2025-12-31T23:59:00") or a bare date
+    ("2025-12-31"). Exits with a clear error on anything we can't parse so the
+    server never sees a malformed payload.
+    """
+    try:
+        dt = datetime.fromisoformat(value)
+    except ValueError:
+        console.print(
+            f"[red]Invalid --expires-at value '{value}'. "
+            "Use ISO format, e.g. 2025-12-31 or 2025-12-31T23:59:00.[/red]"
+        )
+        raise typer.Exit(1)
+    return dt.isoformat()
+
+
+def _print_share_details(data: dict) -> None:
+    """Print a single share's fields in the snapshot-style detail layout."""
+    console.print(f"  Token: {data.get('token', 'unknown')}")
+    console.print(f"  URL: [blue underline]{data.get('share_url', '-')}[/blue underline]")
+    console.print(f"  Project: {data.get('project_name', '-')}")
+    console.print(f"  Note: {data.get('note_permalink', '-')}")
+    console.print(f"  Enabled: {'yes' if data.get('enabled', False) else 'no'}")
+    console.print(f"  Expires: {_format_timestamp(data.get('expires_at'))}")
+    console.print(f"  Views: {data.get('view_count', 0)}")
+    console.print(f"  Created: {_format_timestamp(data.get('created_at'))}")
+
+
+@share_app.command("create")
+def create(
+    project: str = typer.Argument(
+        ...,
+        help="Name of the project the note belongs to",
+    ),
+    permalink: str = typer.Argument(
+        ...,
+        help="Permalink of the note to share",
+    ),
+    expires_at: Optional[str] = typer.Option(
+        None,
+        "--expires-at",
+        "-e",
+        help="Optional expiration date/time (ISO 8601, e.g. 2025-12-31)",
+    ),
+    workspace: Optional[str] = typer.Option(
+        None,
+        "--workspace",
+        help="Workspace to route to: a workspace slug, display name, or tenant ID",
+    ),
+) -> None:
+    """Create a public share link for a note.
+
+    Examples:
+      bm cloud share create my-project notes/my-idea
+      bm cloud share create my-project notes/my-idea --expires-at 2025-12-31
+      bm cloud share create my-project notes/my-idea --workspace acme
+    """
+
+    # Validate --expires-at before any async/API work so a parse error surfaces
+    # a single clean message and exits, rather than being re-wrapped by the broad
+    # handler below as "Unexpected error: 1" (typer.Exit subclasses Exception).
+    payload: dict = {
+        "project_name": project,
+        "note_permalink": permalink,
+    }
+    if expires_at is not None:
+        payload["expires_at"] = _parse_expires_at(expires_at)
+
+    async def _create():
+        try:
+            config_manager = ConfigManager()
+            config = config_manager.config
+            host_url = config.cloud_host.rstrip("/")
+
+            console.print("[blue]Creating share link...[/blue]")
+
+            response = await make_api_request(
+                method="POST",
+                url=f"{host_url}/api/shares",
+                json_data=payload,
+                headers=await _workspace_headers(project_name=project, workspace=workspace),
+            )
+
+            data = response.json()
+
+            console.print("[green]Share link created successfully[/green]")
+            _print_share_details(data)
+
+        except typer.Exit:
+            raise
+        except SubscriptionRequiredError as e:
+            console.print("\n[red]Subscription Required[/red]\n")
+            console.print(f"[yellow]{e.args[0]}[/yellow]\n")
+            console.print(f"Subscribe at: [blue underline]{e.subscribe_url}[/blue underline]\n")
+            raise typer.Exit(1)
+        except CloudAPIError as e:
+            if e.status_code == 404:
+                console.print(f"[red]Note not found: {permalink} (project: {project})[/red]")
+            else:
+                console.print(f"[red]Failed to create share link: {e}[/red]")
+            raise typer.Exit(1)
+        except Exception as e:
+            console.print(f"[red]Unexpected error: {e}[/red]")
+            raise typer.Exit(1)
+
+    asyncio.run(_create())
+
+
+@share_app.command("list")
+def list_shares(
+    project: Optional[str] = typer.Option(
+        None,
+        "--project",
+        "-p",
+        help="Filter shares by project name",
+    ),
+    workspace: Optional[str] = typer.Option(
+        None,
+        "--workspace",
+        help="Workspace to route to: a workspace slug, display name, or tenant ID",
+    ),
+) -> None:
+    """List public share links.
+
+    Examples:
+      bm cloud share list
+      bm cloud share list --project my-project
+      bm cloud share list --project my-project --workspace acme
+    """
+
+    async def _list():
+        try:
+            config_manager = ConfigManager()
+            config = config_manager.config
+            host_url = config.cloud_host.rstrip("/")
+
+            url = f"{host_url}/api/shares"
+            if project:
+                # Encode the filter so project names with query-reserved
+                # characters (&, +, #, spaces) reach the server intact rather
+                # than being parsed as extra query parameters.
+                url += f"?{urlencode({'project_name': project})}"
+
+            console.print("[blue]Fetching share links...[/blue]")
+
+            response = await make_api_request(
+                method="GET",
+                url=url,
+                headers=await _workspace_headers(project_name=project, workspace=workspace),
+            )
+
+            data = response.json()
+            shares = data.get("shares", [])
+            total = data.get("total", len(shares))
+
+            if not shares:
+                console.print("[yellow]No share links found[/yellow]")
+                console.print(
+                    "\n[dim]Create a share with: bm cloud share create <project> <permalink>[/dim]"
+                )
+                return
+
+            table = Table(title=f"Public Shares ({total} total)")
+            table.add_column("Token", style="cyan", no_wrap=True)
+            table.add_column("Project", style="yellow")
+            table.add_column("Note", style="white")
+            table.add_column("Enabled", style="green")
+            table.add_column("Expires", style="green")
+            table.add_column("Views", style="magenta", justify="right")
+            table.add_column("URL", style="blue", overflow="fold")
+
+            for share in shares:
+                table.add_row(
+                    share.get("token", "unknown"),
+                    share.get("project_name", "-"),
+                    share.get("note_permalink", "-"),
+                    "yes" if share.get("enabled", False) else "no",
+                    _format_timestamp(share.get("expires_at")),
+                    str(share.get("view_count", 0)),
+                    share.get("share_url", "-"),
+                )
+
+            console.print(table)
+
+        # Re-raise typer.Exit before the broad handler below: workspace resolution
+        # raises typer.Exit (a subclass of Exception) for ambiguous/unknown
+        # identifiers, and that must not be re-wrapped as "Unexpected error: 1".
+        except typer.Exit:
+            raise
+        except SubscriptionRequiredError as e:
+            console.print("\n[red]Subscription Required[/red]\n")
+            console.print(f"[yellow]{e.args[0]}[/yellow]\n")
+            console.print(f"Subscribe at: [blue underline]{e.subscribe_url}[/blue underline]\n")
+            raise typer.Exit(1)
+        except CloudAPIError as e:
+            console.print(f"[red]Failed to list share links: {e}[/red]")
+            raise typer.Exit(1)
+        except Exception as e:
+            console.print(f"[red]Unexpected error: {e}[/red]")
+            raise typer.Exit(1)
+
+    asyncio.run(_list())
+
+
+@share_app.command("update")
+def update(
+    token: str = typer.Argument(
+        ...,
+        help="The token of the share to update",
+    ),
+    enable: bool = typer.Option(
+        False,
+        "--enable",
+        help="Enable the share link",
+    ),
+    disable: bool = typer.Option(
+        False,
+        "--disable",
+        help="Disable the share link without deleting it",
+    ),
+    expires_at: Optional[str] = typer.Option(
+        None,
+        "--expires-at",
+        "-e",
+        help="New expiration date/time (ISO 8601). Use 'none' to clear it.",
+    ),
+    workspace: Optional[str] = typer.Option(
+        None,
+        "--workspace",
+        help="Workspace the share belongs to: a workspace slug, display name, or tenant ID",
+    ),
+) -> None:
+    """Update a share link: enable/disable it or change its expiration.
+
+    Examples:
+      bm cloud share update abc123 --disable
+      bm cloud share update abc123 --enable
+      bm cloud share update abc123 --expires-at 2026-01-01
+      bm cloud share update abc123 --expires-at none
+      bm cloud share update abc123 --disable --workspace acme
+    """
+
+    async def _update():
+        try:
+            # --- Validate flags ---
+            # Trigger: both toggles passed, or neither toggle and no expiry change.
+            # Why: PATCH needs at least one concrete field, and enable/disable
+            #      conflict; reject up front so we don't send an empty/ambiguous body.
+            if enable and disable:
+                console.print("[red]Cannot use --enable and --disable together[/red]")
+                raise typer.Exit(1)
+            if not enable and not disable and expires_at is None:
+                console.print(
+                    "[red]Nothing to update. Pass --enable, --disable, or --expires-at.[/red]"
+                )
+                raise typer.Exit(1)
+
+            payload: dict = {}
+            if enable:
+                payload["enabled"] = True
+            if disable:
+                payload["enabled"] = False
+            if expires_at is not None:
+                # "none" clears the expiration; anything else is parsed as a date.
+                payload["expires_at"] = (
+                    None if expires_at.lower() == "none" else _parse_expires_at(expires_at)
+                )
+
+            config_manager = ConfigManager()
+            config = config_manager.config
+            host_url = config.cloud_host.rstrip("/")
+
+            console.print("[blue]Updating share link...[/blue]")
+
+            response = await make_api_request(
+                method="PATCH",
+                url=f"{host_url}/api/shares/{token}",
+                json_data=payload,
+                headers=await _workspace_headers(workspace=workspace),
+            )
+
+            data = response.json()
+
+            console.print("[green]Share link updated successfully[/green]")
+            _print_share_details(data)
+
+        except typer.Exit:
+            raise
+        except SubscriptionRequiredError as e:
+            console.print("\n[red]Subscription Required[/red]\n")
+            console.print(f"[yellow]{e.args[0]}[/yellow]\n")
+            console.print(f"Subscribe at: [blue underline]{e.subscribe_url}[/blue underline]\n")
+            raise typer.Exit(1)
+        except CloudAPIError as e:
+            if e.status_code == 404:
+                console.print(f"[red]Share not found: {token}[/red]")
+            else:
+                console.print(f"[red]Failed to update share link: {e}[/red]")
+            raise typer.Exit(1)
+        except Exception as e:
+            console.print(f"[red]Unexpected error: {e}[/red]")
+            raise typer.Exit(1)
+
+    asyncio.run(_update())
+
+
+@share_app.command("revoke")
+def revoke(
+    token: str = typer.Argument(
+        ...,
+        help="The token of the share to revoke",
+    ),
+    force: bool = typer.Option(
+        False,
+        "--force",
+        "-f",
+        help="Skip confirmation prompt",
+    ),
+    workspace: Optional[str] = typer.Option(
+        None,
+        "--workspace",
+        help="Workspace the share belongs to: a workspace slug, display name, or tenant ID",
+    ),
+) -> None:
+    """Revoke (delete) a public share link.
+
+    Examples:
+      bm cloud share revoke abc123
+      bm cloud share revoke abc123 --force
+      bm cloud share revoke abc123 --force --workspace acme
+    """
+
+    async def _revoke():
+        try:
+            config_manager = ConfigManager()
+            config = config_manager.config
+            host_url = config.cloud_host.rstrip("/")
+
+            if not force:
+                confirmed = typer.confirm(f"Are you sure you want to revoke share '{token}'?")
+                if not confirmed:
+                    console.print("[yellow]Revocation cancelled[/yellow]")
+                    raise typer.Exit(0)
+
+            console.print("[blue]Revoking share link...[/blue]")
+
+            await make_api_request(
+                method="DELETE",
+                url=f"{host_url}/api/shares/{token}",
+                headers=await _workspace_headers(workspace=workspace),
+            )
+
+            console.print(f"[green]Share {token} revoked successfully[/green]")
+
+        except typer.Exit:
+            raise
+        except SubscriptionRequiredError as e:
+            console.print("\n[red]Subscription Required[/red]\n")
+            console.print(f"[yellow]{e.args[0]}[/yellow]\n")
+            console.print(f"Subscribe at: [blue underline]{e.subscribe_url}[/blue underline]\n")
+            raise typer.Exit(1)
+        except CloudAPIError as e:
+            if e.status_code == 404:
+                console.print(f"[red]Share not found: {token}[/red]")
+            else:
+                console.print(f"[red]Failed to revoke share link: {e}[/red]")
+            raise typer.Exit(1)
+        except Exception as e:
+            console.print(f"[red]Unexpected error: {e}[/red]")
+            raise typer.Exit(1)
+
+    asyncio.run(_revoke())

src/basic_memory/cli/commands/tool.py

@@ -1,42 +1,18 @@
 """CLI tool commands for Basic Memory.
 
 Every command calls its MCP tool with output_format="json" and prints the result.
-Commands that benefit from human-readable output (search-notes, read-note,
-build-context, recent-activity) support three output modes:
-
-- **JSON** — raw machine-readable JSON.  Used when ``--json`` is passed, or
-  automatically when stdout is not a TTY (piped/redirected), so scripts stay
-  parseable.  This follows the same bm status / bm project list precedent.
-- **Rich** — colored Panel/Table/Tree/Markdown output.  The default interactive
-  experience when stdout is a TTY.
-- **Plain** — undecorated, greppable text (no ANSI colors, no box-drawing, no
-  markup).  Forced with ``--plain`` even when piped.
-
-Precedence, highest first: ``--json`` > ``--plain`` > non-TTY (JSON) > TTY
-(config ``cli_output_style``, ``rich`` by default).  Passing both ``--json`` and
-``--plain`` is an error.  The interactive default for a TTY is controlled by the
-``cli_output_style`` config option (``rich``/``plain``; env
-``BASIC_MEMORY_CLI_OUTPUT_STYLE``).
+No text formatting, no separate code paths, no duplicate data fetching.
 """
 
 import json
 import sys
-from typing import Annotated, Any, Dict, List, Literal, Optional
+from typing import Annotated, Any, Dict, List, Optional
 
 import typer
 from loguru import logger
-from rich.console import Console
-from rich.markdown import Markdown
-from rich.markup import escape as markup_escape
-from rich.panel import Panel
-from rich.table import Table
-from rich.text import Text
-from rich.tree import Tree
 
 from basic_memory.cli.app import app
 from basic_memory.cli.commands.command_utils import run_with_cleanup
-from basic_memory.config import ConfigManager
-from basic_memory.file_utils import has_frontmatter, remove_frontmatter
 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 delete_note as mcp_delete_note
@@ -56,410 +32,15 @@ app.add_typer(tool_app, name="tool", help="Access to MCP tools via CLI")
 
 VALID_EDIT_OPERATIONS = ["append", "prepend", "find_replace", "replace_section"]
 
-# Shared Rich console (stderr=False so output goes to stdout, matching _print_json).
-console = Console()
-
 
 # --- Shared helpers ---
 
-OutputMode = Literal["json", "rich", "plain"]
-
-
-def _use_rich() -> bool:
-    """Return True when stdout is an interactive TTY.
-
-    Why: piped output (scripts, jq, etc.) must stay machine-parseable; the
-         interactive (Rich/plain) renderers are only the default in a terminal.
-    Outcome: a formatted renderer in a terminal; raw JSON when piped or redirected.
-
-    Note: tests patch this to simulate a TTY, so the precedence logic in
-    ``_resolve_output_mode`` routes its terminal check through here.
-    """
-    return sys.stdout.isatty()
-
-
-def _validate_output_flags(json_output: bool, plain: bool) -> None:
-    """Reject the contradictory --json/--plain combination.
-
-    Trigger: both --json and --plain were passed.
-    Why: they request mutually exclusive output modes (raw JSON vs undecorated
-         human text); silently picking one would hide a user mistake.
-    Outcome: a clear typer error with a non-zero exit.
-    """
-    if json_output and plain:
-        typer.echo("Error: --json and --plain are mutually exclusive.", err=True)
-        raise typer.Exit(1)
-
-
-def _resolve_output_mode(json_output: bool, plain: bool) -> OutputMode:
-    """Resolve the effective output mode from flags, TTY state, and config.
-
-    Precedence, highest first:
-      1. --json            → raw JSON (wins over everything else)
-      2. --plain           → undecorated plain text (even when piped)
-      3. non-TTY stdout    → raw JSON (script compatibility, unchanged)
-      4. TTY               → config ``cli_output_style`` (rich by default)
-
-    Callers must invoke ``_validate_output_flags`` first; this helper assumes the
-    --json/--plain combination has already been rejected.
-    """
-    if json_output:
-        return "json"
-    if plain:
-        return "plain"
-    if not _use_rich():
-        return "json"
-    # Trigger: interactive TTY with no explicit mode flag.
-    # Why: let users choose their default terminal experience without a flag.
-    # Outcome: honor cli_output_style (rich out of the box, plain if configured).
-    return ConfigManager().config.cli_output_style
-
 
 def _print_json(result: Any) -> None:
     """Print a result as formatted JSON."""
     print(json.dumps(result, indent=2, ensure_ascii=True, default=str))
 
 
-# --- Rich formatters ---
-
-
-def _display_search_results(result: dict[str, Any], query: str = "") -> None:
-    """Render search-notes results as a Rich table.
-
-    Real SearchResponse.model_dump() shape:
-      results: list of SearchResult dicts (title, type, permalink, score, matched_chunk, content)
-      current_page: int   (NOT "page")
-      page_size: int
-      total: int
-      has_more: bool
-    """
-    results = result.get("results", [])
-    # Trigger: API returns total=0 even when results is non-empty (upstream quirk).
-    # Why: the key exists with value 0, so result.get("total", len(results)) never
-    #      applies its default, leaving the subtitle as "0 result(s)" under a
-    #      populated table.
-    # Outcome: fall back to len(results) when total is falsy but results is non-empty.
-    raw_total = result.get("total", len(results))
-    total = raw_total if raw_total else len(results)
-    # Real key is "current_page"; fall back to "page" for forward-compat.
-    page = result.get("current_page") or result.get("page", 1)
-    page_size = result.get("page_size", len(results)) or 1
-
-    # Trigger: query is user-supplied text that may contain Rich markup characters.
-    # Why: interpolating it directly into a markup string causes brackets to be
-    #      parsed as style tags, swallowing or restyling bracketed content.
-    # Outcome: escape the query so its literal characters are always displayed.
-    escaped_query = markup_escape(query) if query else query
-    title = f"Search: [bold cyan]{escaped_query}[/bold cyan]" if query else "Search results"
-    subtitle = f"{total} result(s)  •  page {page} of {max(1, -(-total // page_size))}"
-
-    if not results:
-        console.print(Panel(Text("No results found.", style="dim"), title=title, expand=False))
-        return
-
-    table = Table(show_header=True, header_style="bold", expand=False)
-    table.add_column("Type", style="dim", width=12)
-    table.add_column("Title", style="bold cyan")
-    table.add_column("Score", style="yellow", width=7)
-    table.add_column("Permalink", style="green")
-    table.add_column("Snippet", style="dim", max_width=60)
-
-    for item in results:
-        item_type = item.get("type", "")
-        # Trigger: user-sourced title/permalink may contain bracketed text.
-        # Why: Rich table cells with a style column interpret markup in cell values,
-        #      swallowing brackets (e.g. "Spec [draft] v2" → "Spec  v2").
-        # Outcome: escape every user-sourced cell value before adding to the table.
-        item_title = markup_escape(item.get("title") or item.get("permalink", ""))
-        permalink = markup_escape(item.get("permalink", ""))
-        score = item.get("score")
-        score_str = f"{score:.2f}" if score is not None else ""
-        # Prefer matched_chunk as the most relevant snippet; fall back to content.
-        raw_snippet = item.get("matched_chunk") or item.get("content") or ""
-        # Truncate to ~200 chars so the table stays readable.
-        snippet = markup_escape(raw_snippet[:200].replace("\n", " ")) if raw_snippet else ""
-        table.add_row(item_type, item_title, score_str, permalink, snippet)
-
-    console.print(Panel(table, title=title, subtitle=subtitle, expand=False))
-
-
-def _display_read_note(result: dict[str, Any], *, include_frontmatter: bool = False) -> None:
-    """Render read-note result: header panel + optional frontmatter + rendered Markdown content."""
-    title = result.get("title", "")
-    permalink = result.get("permalink", "")
-    content = result.get("content", "")
-    frontmatter: dict[str, Any] = result.get("frontmatter") or {}
-
-    # The header already uses Text.append so title is never markup-interpreted.
-    header = Text()
-    header.append(title, style="bold cyan")
-    if permalink:
-        header.append(f"  [{permalink}]", style="dim green")
-
-    console.print(Panel(header, expand=False))
-
-    # Trigger: --frontmatter was passed; the MCP tool populates "frontmatter".
-    # Why: the JSON payload always carries a "frontmatter" key regardless of the flag,
-    #      so checking non-empty alone would render it even without the flag.  The flag
-    #      must be threaded in to gate the panel.
-    # Outcome: print a dim key/value block above the content only when the flag is set.
-    if include_frontmatter and frontmatter:
-        fm_table = Table(show_header=False, box=None, padding=(0, 1), expand=False)
-        fm_table.add_column("key", style="dim")
-        fm_table.add_column("value", style="dim")
-        for key, value in frontmatter.items():
-            # Trigger: frontmatter keys/values are user-sourced and may contain markup.
-            # Why: Rich table cells with a style column parse markup, so bracketed
-            #      keys or values would be silently consumed or restyled.
-            # Outcome: escape both key and value before adding them to the table.
-            fm_table.add_row(markup_escape(str(key)), markup_escape(str(value)))
-        console.print(Panel(fm_table, title="[dim]frontmatter[/dim]", expand=False))
-
-    # Trigger: --frontmatter makes the API return the literal file, so
-    # content starts with the frontmatter block the panel above already shows.
-    # Why: rendering it again through Markdown duplicates the frontmatter (and
-    #      Markdown mangles the --- fences into rules/headings).
-    # Outcome: strip the block from the body; the panel is the frontmatter view.
-    body = content
-    if include_frontmatter and content and has_frontmatter(content):
-        body = remove_frontmatter(content)
-    if body and body.strip():
-        console.print(Markdown(body))
-    else:
-        console.print(Text("(no content)", style="dim"))
-
-
-def _display_build_context(result: dict[str, Any]) -> None:
-    """Render build-context result as a Rich tree.
-
-    Real GraphContext.model_dump() shape:
-      results: list of ContextResult dicts, each with:
-        primary_result: EntitySummary | RelationSummary | ObservationSummary
-        observations:   list of ObservationSummary
-        related_results: list of EntitySummary | RelationSummary | ObservationSummary
-      metadata: {"uri": ..., ...}
-      page/page_size/has_more
-
-    Each summary has: type, title (EntitySummary/RelationSummary), permalink,
-    and relation_type (RelationSummary only).
-    """
-    metadata = result.get("metadata", {})
-    uri = metadata.get("uri", "")
-    context_items: list[dict[str, Any]] = list(result.get("results", []))
-
-    # Trigger: uri is user-sourced and may contain Rich markup characters.
-    # Why: interpolating it directly into a markup string causes brackets to be
-    #      parsed as style tags, swallowing or restyling bracketed content.
-    # Outcome: escape the uri so its literal characters are always displayed.
-    label = f"[bold cyan]{markup_escape(uri)}[/bold cyan]" if uri else "Context"
-    tree = Tree(f"[bold]Context:[/bold] {label}")
-
-    if not context_items:
-        tree.add("[dim]No related content found.[/dim]")
-    else:
-        for context_result in context_items:
-            # --- Primary result node ---
-            primary = context_result.get("primary_result", {})
-            # Trigger: p_title and p_type are user-sourced values from the knowledge graph.
-            # Why: embedding them in markup strings without escaping would cause any
-            #      bracketed text (e.g. an entity titled "Spec [draft]") to be consumed
-            #      by the Rich markup parser and silently dropped from output.
-            # Outcome: escape all user values before interpolating into markup strings.
-            p_title = markup_escape(primary.get("title") or primary.get("permalink", ""))
-            p_type = markup_escape(primary.get("type", ""))
-            primary_label = f"[cyan]{p_title}[/cyan]"
-            if p_type:
-                primary_label = f"[dim]{p_type}[/dim]  {primary_label}"
-            primary_node = tree.add(primary_label)
-
-            # --- Observations as children (category + truncated content) ---
-            # Trigger: ContextResult.observations exists in the JSON output but was
-            #          never rendered in the Rich path.
-            # Why: users running interactively lost core entity facts (observations)
-            #      that the --json path exposes; the TTY view must be at least as
-            #      informative as the JSON view for the primary entity.
-            # Outcome: each observation appears as a dim "[category] content" leaf
-            #          under its primary node, truncated at 120 chars.
-            observations: list[dict[str, Any]] = list(context_result.get("observations", []))
-            for obs in observations:
-                category = obs.get("category", "")
-                obs_content = obs.get("content", "")
-                # Truncate long observations so the tree stays readable.
-                if len(obs_content) > 120:
-                    obs_content = obs_content[:117] + "..."
-                # Trigger: category and obs_content are user-sourced strings that may
-                #          contain Rich markup characters.  The category is also wrapped
-                #          in literal "[" "]" brackets in the label, which must be
-                #          escaped too so Rich does not treat "[fact]" as a style tag.
-                # Why: embedding "[fact]" in a markup string causes Rich to parse it as
-                #      an unknown tag and silently drop the text.
-                # Outcome: escape the full "[category] content" fragment including the
-                #          surrounding brackets before embedding it in a styled label.
-                obs_label = f"[dim]{markup_escape(f'[{category}] {obs_content}')}[/dim]"
-                primary_node.add(obs_label)
-
-            # --- Related items as children ---
-            related: list[dict[str, Any]] = list(context_result.get("related_results", []))
-            for rel_item in related:
-                rel_title = markup_escape(rel_item.get("title") or rel_item.get("permalink", ""))
-                rel_type = markup_escape(rel_item.get("type", ""))
-                relation = markup_escape(rel_item.get("relation_type", ""))
-
-                parts = []
-                if relation:
-                    parts.append(f"[yellow]{relation}[/yellow]")
-                if rel_type:
-                    parts.append(f"[dim]{rel_type}[/dim]")
-                parts.append(f"[cyan]{rel_title}[/cyan]")
-                primary_node.add(" ".join(parts))
-
-    # Count total related items across all primary results.
-    total_related = sum(len(cr.get("related_results", [])) for cr in context_items)
-    total_observations = sum(len(cr.get("observations", [])) for cr in context_items)
-    subtitle = f"{len(context_items)} primary  •  {total_observations} observations  •  {total_related} related"
-    console.print(Panel(tree, subtitle=subtitle, expand=False))
-
-
-def _display_recent_activity(result: list[dict[str, Any]]) -> None:
-    """Render recent-activity results as a Rich table."""
-    if not result:
-        console.print(
-            Panel(Text("No recent activity.", style="dim"), title="Recent Activity", expand=False)
-        )
-        return
-
-    table = Table(show_header=True, header_style="bold", expand=False)
-    table.add_column("Type", style="dim", width=12)
-    table.add_column("Title", style="bold cyan")
-    table.add_column("Permalink", style="green")
-    table.add_column("Updated", style="dim")
-
-    for item in result:
-        item_type = item.get("type", "")
-        # Trigger: title, permalink, and timestamps are user-sourced strings from the
-        #          knowledge graph and may contain Rich markup characters.
-        # Why: Rich table cells with a style column parse markup in cell values, so
-        #      bracketed content would be silently consumed or restyled.
-        # Outcome: escape all user-sourced cell values before adding to the table.
-        item_title = markup_escape(item.get("title") or item.get("permalink", ""))
-        permalink = markup_escape(item.get("permalink", ""))
-        updated = str(item.get("updated_at") or item.get("created_at") or "")
-        table.add_row(item_type, item_title, permalink, updated)
-
-    console.print(Panel(table, title="Recent Activity", expand=False))
-
-
-# --- Plain formatters ---
-#
-# Plain output is NOT Rich markup: it is undecorated, greppable text printed via
-# the builtin print().  Literal brackets ([draft], [fact]) must survive verbatim,
-# so we deliberately do NOT call rich.markup.escape here -- escaping is only for
-# the Rich path and would corrupt literal brackets in plain text.
-
-
-def _plain_search_results(result: dict[str, Any], query: str = "") -> None:
-    """Render search-notes results as numbered plain-text entries.
-
-    Mirrors the Rich table content: a header line, then one numbered block per
-    result (title / score / permalink) with an indented snippet line.
-    """
-    results = result.get("results", [])
-    # Mirror the Rich path's total fix: the API can return total=0 with a
-    # populated results list, so fall back to len(results) when total is falsy.
-    raw_total = result.get("total", len(results))
-    total = raw_total if raw_total else len(results)
-
-    header = f"Search: {query}" if query else "Search results"
-    print(header)
-    print(f"{total} result(s)")
-
-    if not results:
-        print("No results found.")
-        return
-
-    for index, item in enumerate(results, start=1):
-        item_title = item.get("title") or item.get("permalink", "")
-        permalink = item.get("permalink", "")
-        score = item.get("score")
-        score_str = f"{score:.2f}" if score is not None else ""
-        print(f"{index}. {item_title}  (score: {score_str})  {permalink}")
-        raw_snippet = item.get("matched_chunk") or item.get("content") or ""
-        if raw_snippet:
-            snippet = raw_snippet[:200].replace("\n", " ")
-            print(f"    {snippet}")
-
-
-def _plain_read_note(result: dict[str, Any]) -> None:
-    """Render read-note content faithfully: the note body, or the literal file.
-
-    Plain mode adds NO decoration: no header line, no synthesized frontmatter
-    block, no placeholder for empty notes. Without --frontmatter the
-    API returns the note body; with it, the literal file (frontmatter block
-    included). Either is printed verbatim, trimmed only of the surrounding
-    newline artifacts the API keeps from frontmatter stripping, so the output
-    round-trips (e.g. ``read-note X --plain --frontmatter > note.md``).
-    """
-    content = result.get("content", "")
-    body = content.strip("\n") if content else ""
-    if body:
-        print(body)
-
-
-def _plain_build_context(result: dict[str, Any]) -> None:
-    """Render build-context as an ASCII-indented outline.
-
-    Each primary result is a top-level line; its observations and related items
-    are two-space indented beneath it, mirroring the Rich tree content.
-    """
-    metadata = result.get("metadata", {})
-    uri = metadata.get("uri", "")
-    context_items: list[dict[str, Any]] = list(result.get("results", []))
-
-    print(f"Context: {uri}" if uri else "Context")
-
-    if not context_items:
-        print("No related content found.")
-        return
-
-    for context_result in context_items:
-        primary = context_result.get("primary_result", {})
-        p_title = primary.get("title") or primary.get("permalink", "")
-        p_type = primary.get("type", "")
-        primary_line = f"{p_type}  {p_title}" if p_type else p_title
-        print(primary_line)
-
-        observations: list[dict[str, Any]] = list(context_result.get("observations", []))
-        for obs in observations:
-            category = obs.get("category", "")
-            obs_content = obs.get("content", "")
-            if len(obs_content) > 120:
-                obs_content = obs_content[:117] + "..."
-            print(f"  [{category}] {obs_content}")
-
-        related: list[dict[str, Any]] = list(context_result.get("related_results", []))
-        for rel_item in related:
-            rel_title = rel_item.get("title") or rel_item.get("permalink", "")
-            rel_type = rel_item.get("type", "")
-            relation = rel_item.get("relation_type", "")
-            parts = [part for part in (relation, rel_type, rel_title) if part]
-            print(f"  {'  '.join(parts)}")
-
-
-def _plain_recent_activity(result: list[dict[str, Any]]) -> None:
-    """Render recent-activity as plain "- title (type) permalink updated" lines."""
-    if not result:
-        print("No recent activity.")
-        return
-
-    print("Recent Activity")
-    for item in result:
-        item_type = item.get("type", "")
-        item_title = item.get("title") or item.get("permalink", "")
-        permalink = item.get("permalink", "")
-        updated = str(item.get("updated_at") or item.get("created_at") or "")
-        print(f"- {item_title} ({item_type}) {permalink} {updated}".rstrip())
-
-
 def _delete_note_failure_message(result: dict[str, Any]) -> str | None:
     """Return the CLI failure message for delete-note JSON results, if any."""
     error = result.get("error")
@@ -600,16 +181,7 @@ def write_note(
 def read_note(
     identifier: str,
     include_frontmatter: bool = typer.Option(
-        False,
-        "--frontmatter",
-        "--include-frontmatter",
-        help="Include YAML frontmatter in output (--include-frontmatter is a deprecated alias)",
-    ),
-    json_output: bool = typer.Option(
-        False, "--json", help="Output raw JSON instead of formatted display"
-    ),
-    plain: bool = typer.Option(
-        False, "--plain", help="Output undecorated plain text (no colors/markup), even when piped"
+        False, "--include-frontmatter", help="Include YAML frontmatter in output"
     ),
     project: Annotated[
         Optional[str],
@@ -629,21 +201,13 @@ def read_note(
 ):
     """Read a markdown note from the knowledge base.
 
-    Three output modes: Rich formatted Markdown (default in a terminal), plain
-    undecorated text (--plain), and raw JSON (--json, or automatically when
-    piped). The interactive default is set by the cli_output_style config option
-    (rich/plain). --json and --plain are mutually exclusive.
-
     Examples:
 
     bm tool read-note my-note
-    bm tool read-note my-note --frontmatter
-    bm tool read-note my-note --plain
-    bm tool read-note my-note --json
+    bm tool read-note my-note --include-frontmatter
     """
     try:
         validate_routing_flags(local, cloud)
-        _validate_output_flags(json_output, plain)
 
         with force_routing(local=local, cloud=cloud):
             result = run_with_cleanup(
@@ -668,15 +232,7 @@ def read_note(
             _print_json(result)
             raise typer.Exit(1)
 
-        # A string result (e.g. a not-found message) has no structured shape to
-        # format, so always fall back to JSON regardless of the resolved mode.
-        mode = _resolve_output_mode(json_output, plain)
-        if mode == "json" or isinstance(result, str):
-            _print_json(result)
-        elif mode == "plain":
-            _plain_read_note(result)
-        else:
-            _display_read_note(result, include_frontmatter=include_frontmatter)
+        _print_json(result)
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)
@@ -834,12 +390,6 @@ def build_context(
     page: int = typer.Option(1, "--page", help="Page number for pagination"),
     page_size: int = typer.Option(10, "--page-size", help="Number of results per page"),
     max_related: int = typer.Option(10, "--max-related", help="Maximum related items to return"),
-    json_output: bool = typer.Option(
-        False, "--json", help="Output raw JSON instead of formatted display"
-    ),
-    plain: bool = typer.Option(
-        False, "--plain", help="Output undecorated plain text (no colors/markup), even when piped"
-    ),
     project: Annotated[
         Optional[str],
         typer.Option(help="The project to use. If not provided, the default project will be used."),
@@ -858,21 +408,13 @@ def build_context(
 ):
     """Get context needed to continue a discussion.
 
-    Three output modes: a Rich tree view (default in a terminal), a plain
-    ASCII-indented outline (--plain), and raw JSON (--json, or automatically when
-    piped). The interactive default is set by the cli_output_style config option
-    (rich/plain). --json and --plain are mutually exclusive.
-
     Examples:
 
     bm tool build-context memory://specs/search
     bm tool build-context specs/search --depth 2 --timeframe 30d
-    bm tool build-context memory://specs/search --plain
-    bm tool build-context memory://specs/search --json
     """
     try:
         validate_routing_flags(local, cloud)
-        _validate_output_flags(json_output, plain)
 
         with force_routing(local=local, cloud=cloud):
             result = run_with_cleanup(
@@ -888,15 +430,7 @@ def build_context(
                     output_format="json",
                 )
             )
-
-        # A string result has no structured shape to format, so fall back to JSON.
-        mode = _resolve_output_mode(json_output, plain)
-        if mode == "json" or isinstance(result, str):
-            _print_json(result)
-        elif mode == "plain":
-            _plain_build_context(result)
-        else:
-            _display_build_context(result)
+        _print_json(result)
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)
@@ -918,12 +452,6 @@ def recent_activity(
     # Match the MCP recent_activity default (page_size=10) so identical default
     # invocations return the same number of rows from CLI and MCP.
     page_size: int = typer.Option(10, "--page-size", help="Number of results per page"),
-    json_output: bool = typer.Option(
-        False, "--json", help="Output raw JSON instead of formatted display"
-    ),
-    plain: bool = typer.Option(
-        False, "--plain", help="Output undecorated plain text (no colors/markup), even when piped"
-    ),
     project: Annotated[
         Optional[str],
         typer.Option(help="The project to use. If not provided, the default project will be used."),
@@ -942,22 +470,14 @@ def recent_activity(
 ):
     """Get recent activity across the knowledge base.
 
-    Three output modes: a formatted Rich table (default in a terminal), plain
-    undecorated lines (--plain), and raw JSON (--json, or automatically when
-    piped). The interactive default is set by the cli_output_style config option
-    (rich/plain). --json and --plain are mutually exclusive.
-
     Examples:
 
     bm tool recent-activity
     bm tool recent-activity --timeframe 30d --page-size 20
     bm tool recent-activity --type entity --type observation
-    bm tool recent-activity --plain
-    bm tool recent-activity --json
     """
     try:
         validate_routing_flags(local, cloud)
-        _validate_output_flags(json_output, plain)
 
         with force_routing(local=local, cloud=cloud):
             result = run_with_cleanup(
@@ -972,15 +492,7 @@ def recent_activity(
                     output_format="json",
                 )
             )
-
-        # A string result has no structured shape to format, so fall back to JSON.
-        mode = _resolve_output_mode(json_output, plain)
-        if mode == "json" or isinstance(result, str):
-            _print_json(result)
-        elif mode == "plain":
-            _plain_recent_activity(result)
-        else:
-            _display_recent_activity(result)
+        _print_json(result)
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)
@@ -1044,12 +556,6 @@ def search_notes(
     ] = None,
     page: int = typer.Option(1, "--page", help="Page number for pagination"),
     page_size: int = typer.Option(10, "--page-size", help="Number of results per page"),
-    json_output: bool = typer.Option(
-        False, "--json", help="Output raw JSON instead of formatted display"
-    ),
-    plain: bool = typer.Option(
-        False, "--plain", help="Output undecorated plain text (no colors/markup), even when piped"
-    ),
     project: Annotated[
         Optional[str],
         typer.Option(help="The project to use. If not provided, the default project will be used."),
@@ -1068,11 +574,6 @@ def search_notes(
 ):
     """Search across all content in the knowledge base.
 
-    Three output modes: a formatted Rich table (default in a terminal), plain
-    numbered text results (--plain), and raw JSON (--json, or automatically when
-    piped). The interactive default is set by the cli_output_style config option
-    (rich/plain). --json and --plain are mutually exclusive.
-
     Examples:
 
     bm tool search-notes "my query"
@@ -1080,12 +581,9 @@ def search_notes(
     bm tool search-notes --tag python --tag async
     bm tool search-notes --meta status=draft
     bm tool search-notes "auth" --entity-type observation --category requirement
-    bm tool search-notes "my query" --plain
-    bm tool search-notes "my query" --json
     """
     try:
         validate_routing_flags(local, cloud)
-        _validate_output_flags(json_output, plain)
 
         mode_flags = [permalink, title, vector, hybrid]
         if sum(1 for enabled in mode_flags if enabled) > 1:  # pragma: no cover
@@ -1160,13 +658,7 @@ def search_notes(
             typer.echo(result, err=True)
             raise typer.Exit(1)
 
-        mode = _resolve_output_mode(json_output, plain)
-        if mode == "json":
-            _print_json(result)
-        elif mode == "plain":
-            _plain_search_results(result, query=query or "")
-        else:
-            _display_search_results(result, query=query or "")
+        _print_json(result)
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)

src/basic_memory/config.py

@@ -442,18 +442,6 @@ class BasicMemoryConfig(BaseSettings):
         ),
     )
 
-    cli_output_style: Literal["rich", "plain"] = Field(
-        default="rich",
-        description=(
-            "Default human-readable output style for interactive `bm tool` commands "
-            "(search-notes, read-note, build-context, recent-activity) when stdout is a TTY. "
-            "'rich' (default) renders colored Panel/Table/Tree/Markdown output; "
-            "'plain' renders undecorated greppable text with no ANSI colors or box-drawing. "
-            "Overridden per-invocation by --json (raw JSON) or --plain (forces plain). "
-            "Env: BASIC_MEMORY_CLI_OUTPUT_STYLE"
-        ),
-    )
-
     ensure_frontmatter_on_sync: bool = Field(
         default=True,
         description="Ensure markdown files have frontmatter during sync by adding derived title/type/permalink when missing. When combined with disable_permalinks=True, this setting takes precedence for missing-frontmatter files and still writes permalinks.",

tests/cli/test_cli_tool_rich_output.py

@@ -1,991 +0,0 @@
-"""Tests for Rich (human-readable) output mode for bm tool commands.
-
-Commands default to Rich output when stdout is a TTY and fall back to raw JSON
-when stdout is piped or --json is supplied.  These tests verify both modes.
-"""
-
-import json
-from types import SimpleNamespace
-from unittest.mock import AsyncMock, patch
-
-import pytest
-from typer.testing import CliRunner
-
-from basic_memory.cli.main import app as cli_app
-
-runner = CliRunner()
-
-# ---------------------------------------------------------------------------
-# Shared mock payloads (mirrors test_cli_tool_json_output.py for symmetry)
-# ---------------------------------------------------------------------------
-
-READ_NOTE_RESULT = {
-    "title": "Test Note",
-    "permalink": "notes/test-note",
-    "file_path": "notes/Test Note.md",
-    # Real payloads keep the leading newline left by frontmatter stripping.
-    "content": "\n# Test Note\n\nhello world",
-    "frontmatter": {"title": "Test Note", "tags": ["test"]},
-}
-
-# With --include-frontmatter the API returns the LITERAL FILE as content
-# (frontmatter block included) alongside the parsed frontmatter dict.
-READ_NOTE_RESULT_WITH_FRONTMATTER = {
-    "title": "Test Note",
-    "permalink": "notes/test-note",
-    "file_path": "notes/Test Note.md",
-    "content": "---\ntitle: Test Note\ntags:\n- test\n---\n\n# Test Note\n\nhello world",
-    "frontmatter": {"title": "Test Note", "tags": ["test"]},
-}
-
-SEARCH_RESULT = {
-    # Real SearchResponse.model_dump() uses "current_page", not "page".
-    # No "query" key in the response -- the query comes from the CLI argument.
-    "total": 2,
-    "current_page": 1,
-    "page_size": 10,
-    "has_more": False,
-    "results": [
-        {
-            "type": "entity",
-            "title": "Test Note",
-            "permalink": "notes/test-note",
-            "file_path": "notes/Test Note.md",
-            "score": 0.95,
-            "matched_chunk": "A snippet about test notes",
-            "content": None,
-        },
-        {
-            "type": "observation",
-            "title": "Another Note",
-            "permalink": "notes/another-note",
-            "file_path": "notes/Another Note.md",
-            "score": 0.72,
-            "matched_chunk": None,
-            "content": "Full content here",
-        },
-    ],
-}
-
-SEARCH_RESULT_EMPTY = {
-    "total": 0,
-    "current_page": 1,
-    "page_size": 10,
-    "has_more": False,
-    "results": [],
-}
-
-BUILD_CONTEXT_RESULT = {
-    # Real GraphContext.model_dump() shape: results is a list of ContextResult dicts.
-    # Each ContextResult has primary_result + observations + related_results.
-    # ObservationSummary fields: type, category, content, permalink, file_path, created_at.
-    "results": [
-        {
-            "primary_result": {
-                "type": "entity",
-                "external_id": "abc123",
-                "title": "Test Note",
-                "permalink": "notes/test-note",
-                "file_path": "notes/Test Note.md",
-                "created_at": "2025-01-01T00:00:00",
-            },
-            "observations": [
-                {
-                    "type": "observation",
-                    "category": "fact",
-                    "content": "This is a key fact about the test note",
-                    "permalink": "notes/test-note",
-                    "file_path": "notes/Test Note.md",
-                    "created_at": "2025-01-01T00:00:00",
-                }
-            ],
-            "related_results": [
-                {
-                    "type": "relation",
-                    "title": "Related Note",
-                    "permalink": "notes/related",
-                    "file_path": "notes/Related Note.md",
-                    "relation_type": "references",
-                    "created_at": "2025-01-01T00:00:00",
-                }
-            ],
-        }
-    ],
-    "metadata": {"uri": "notes/test-note", "depth": 1},
-    "page": 1,
-    "page_size": 10,
-    "has_more": False,
-}
-
-BUILD_CONTEXT_EMPTY = {
-    "results": [],
-    "metadata": {"uri": "notes/test-note", "depth": 1},
-    "page": 1,
-    "page_size": 10,
-    "has_more": False,
-}
-
-RECENT_ACTIVITY_RESULT = [
-    # Real _extract_recent_rows output keys: type/title/permalink/file_path/created_at
-    # (optional: project).  No "updated_at" key in the real output.
-    {
-        "type": "entity",
-        "title": "Note A",
-        "permalink": "notes/note-a",
-        "file_path": "notes/Note A.md",
-        "created_at": "2025-01-01 00:00:00",
-    },
-    {
-        "type": "entity",
-        "title": "Note B",
-        "permalink": "notes/note-b",
-        "file_path": "notes/Note B.md",
-        "created_at": "2025-01-02 00:00:00",
-    },
-]
-
-
-# ---------------------------------------------------------------------------
-# Helper: simulate a TTY by patching _use_rich to return True
-# ---------------------------------------------------------------------------
-
-
-def _tty_runner(args, **kwargs):
-    """Invoke CLI as if stdout is a TTY (interactive output enabled)."""
-    with patch("basic_memory.cli.commands.tool._use_rich", return_value=True):
-        return runner.invoke(cli_app, args, **kwargs)
-
-
-def _tty_runner_with_style(args, style, **kwargs):
-    """Invoke CLI as if stdout is a TTY with a given cli_output_style config value.
-
-    Patches both _use_rich (simulate TTY) and tool.ConfigManager so that
-    _resolve_output_mode reads the configured interactive style.
-    """
-    fake_cm = patch(
-        "basic_memory.cli.commands.tool.ConfigManager",
-        return_value=SimpleNamespace(config=SimpleNamespace(cli_output_style=style)),
-    )
-    with patch("basic_memory.cli.commands.tool._use_rich", return_value=True), fake_cm:
-        return runner.invoke(cli_app, args, **kwargs)
-
-
-# ---------------------------------------------------------------------------
-# search-notes – Rich output
-# ---------------------------------------------------------------------------
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT,
-)
-def test_search_notes_rich_output_default(mock_mcp):
-    """search-notes produces Rich table output when stdout is a TTY."""
-    result = _tty_runner(["tool", "search-notes", "test"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    # Rich output should NOT be valid JSON
-    with pytest.raises((json.JSONDecodeError, ValueError)):
-        json.loads(result.output)
-    # But it should contain the result titles and partial permalink
-    assert "Test Note" in result.output
-    assert "Another Note" in result.output
-    # Rich may truncate long permalinks with ellipsis; check the prefix.
-    assert "notes/test-no" in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT_EMPTY,
-)
-def test_search_notes_rich_empty(mock_mcp):
-    """search-notes Rich output handles empty results gracefully."""
-    result = _tty_runner(["tool", "search-notes", "nothing"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "No results found" in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT,
-)
-def test_search_notes_json_flag_overrides_tty(mock_mcp):
-    """search-notes --json outputs raw JSON even when stdout is a TTY."""
-    result = _tty_runner(["tool", "search-notes", "test", "--json"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    data = json.loads(result.output)
-    assert data["total"] == 2
-    assert data["results"][0]["title"] == "Test Note"
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT,
-)
-def test_search_notes_non_tty_gives_json(mock_mcp):
-    """search-notes outputs JSON when stdout is not a TTY (default runner behaviour)."""
-    # CliRunner does not set isatty(); _use_rich() returns False → JSON path.
-    result = runner.invoke(cli_app, ["tool", "search-notes", "test"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    data = json.loads(result.output)
-    assert data["total"] == 2
-
-
-# ---------------------------------------------------------------------------
-# read-note – Rich output
-# ---------------------------------------------------------------------------
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_read_note",
-    new_callable=AsyncMock,
-    return_value=READ_NOTE_RESULT,
-)
-def test_read_note_rich_output_default(mock_mcp):
-    """read-note produces Rich formatted output when stdout is a TTY."""
-    result = _tty_runner(["tool", "read-note", "test-note"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    # Rich output contains the note title
-    assert "Test Note" in result.output
-    # And the rendered markdown content
-    assert "hello world" in result.output
-    # Not raw JSON
-    with pytest.raises((json.JSONDecodeError, ValueError)):
-        json.loads(result.output)
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_read_note",
-    new_callable=AsyncMock,
-    return_value=READ_NOTE_RESULT,
-)
-def test_read_note_json_flag_overrides_tty(mock_mcp):
-    """read-note --json outputs raw JSON even when stdout is a TTY."""
-    result = _tty_runner(["tool", "read-note", "test-note", "--json"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    data = json.loads(result.output)
-    assert data["title"] == "Test Note"
-    # JSON mode is byte-faithful: the payload's leading newline (frontmatter-strip
-    # artifact) is preserved here even though display modes trim it.
-    assert data["content"] == "\n# Test Note\n\nhello world"
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_read_note",
-    new_callable=AsyncMock,
-    return_value={"title": "", "permalink": "", "content": "", "frontmatter": {}},
-)
-def test_read_note_rich_empty_content(mock_mcp):
-    """read-note Rich output handles empty content without crashing."""
-    result = _tty_runner(["tool", "read-note", "empty-note"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "no content" in result.output.lower()
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_read_note",
-    new_callable=AsyncMock,
-    return_value=READ_NOTE_RESULT,
-)
-def test_read_note_non_tty_gives_json(mock_mcp):
-    """read-note outputs JSON when stdout is not a TTY."""
-    result = runner.invoke(cli_app, ["tool", "read-note", "test-note"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    data = json.loads(result.output)
-    assert data["title"] == "Test Note"
-
-
-# ---------------------------------------------------------------------------
-# build-context – Rich output
-# ---------------------------------------------------------------------------
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_build_context",
-    new_callable=AsyncMock,
-    return_value=BUILD_CONTEXT_RESULT,
-)
-def test_build_context_rich_output_default(mock_mcp):
-    """build-context produces Rich tree output when stdout is a TTY."""
-    result = _tty_runner(["tool", "build-context", "memory://notes/test-note"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "notes/test-note" in result.output
-    assert "Related Note" in result.output
-    # Not raw JSON
-    with pytest.raises((json.JSONDecodeError, ValueError)):
-        json.loads(result.output)
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_build_context",
-    new_callable=AsyncMock,
-    return_value=BUILD_CONTEXT_EMPTY,
-)
-def test_build_context_rich_empty(mock_mcp):
-    """build-context Rich output handles empty results gracefully."""
-    result = _tty_runner(["tool", "build-context", "memory://notes/test-note"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "No related content found" in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_build_context",
-    new_callable=AsyncMock,
-    return_value=BUILD_CONTEXT_RESULT,
-)
-def test_build_context_json_flag_overrides_tty(mock_mcp):
-    """build-context --json outputs raw JSON even when stdout is a TTY."""
-    result = _tty_runner(["tool", "build-context", "memory://notes/test-note", "--json"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    data = json.loads(result.output)
-    assert "results" in data
-    # Real shape: results[i] is a ContextResult with primary_result nested inside.
-    assert data["results"][0]["primary_result"]["title"] == "Test Note"
-    assert data["results"][0]["related_results"][0]["title"] == "Related Note"
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_build_context",
-    new_callable=AsyncMock,
-    return_value=BUILD_CONTEXT_RESULT,
-)
-def test_build_context_non_tty_gives_json(mock_mcp):
-    """build-context outputs JSON when stdout is not a TTY."""
-    result = runner.invoke(cli_app, ["tool", "build-context", "memory://notes/test-note"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    data = json.loads(result.output)
-    assert "results" in data
-
-
-# ---------------------------------------------------------------------------
-# recent-activity – Rich output
-# ---------------------------------------------------------------------------
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_recent_activity",
-    new_callable=AsyncMock,
-    return_value=RECENT_ACTIVITY_RESULT,
-)
-def test_recent_activity_rich_output_default(mock_mcp):
-    """recent-activity produces Rich table output when stdout is a TTY."""
-    result = _tty_runner(["tool", "recent-activity"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "Note A" in result.output
-    assert "Note B" in result.output
-    assert "notes/note-a" in result.output
-    # Not raw JSON
-    with pytest.raises((json.JSONDecodeError, ValueError)):
-        json.loads(result.output)
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_recent_activity",
-    new_callable=AsyncMock,
-    return_value=[],
-)
-def test_recent_activity_rich_empty(mock_mcp):
-    """recent-activity Rich output handles empty results gracefully."""
-    result = _tty_runner(["tool", "recent-activity"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "No recent activity" in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_recent_activity",
-    new_callable=AsyncMock,
-    return_value=RECENT_ACTIVITY_RESULT,
-)
-def test_recent_activity_json_flag_overrides_tty(mock_mcp):
-    """recent-activity --json outputs raw JSON even when stdout is a TTY."""
-    result = _tty_runner(["tool", "recent-activity", "--json"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    data = json.loads(result.output)
-    assert isinstance(data, list)
-    assert data[0]["title"] == "Note A"
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_recent_activity",
-    new_callable=AsyncMock,
-    return_value=RECENT_ACTIVITY_RESULT,
-)
-def test_recent_activity_non_tty_gives_json(mock_mcp):
-    """recent-activity outputs JSON when stdout is not a TTY."""
-    result = runner.invoke(cli_app, ["tool", "recent-activity"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    data = json.loads(result.output)
-    assert isinstance(data, list)
-    assert len(data) == 2
-
-
-# ---------------------------------------------------------------------------
-# read-note – frontmatter rendering (issue #678)
-# ---------------------------------------------------------------------------
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_read_note",
-    new_callable=AsyncMock,
-    return_value=READ_NOTE_RESULT_WITH_FRONTMATTER,
-)
-def test_read_note_rich_include_frontmatter(mock_mcp):
-    """read-note --include-frontmatter renders the panel once, not twice.
-
-    Regression 1: the Rich renderer silently dropped frontmatter even with the
-    flag. Regression 2: with the flag, content is the LITERAL FILE, so the
-    frontmatter block must be stripped from the Markdown body or it renders
-    again under the panel.
-    """
-    result = _tty_runner(["tool", "read-note", "test-note", "--frontmatter"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    # Frontmatter panel appears with key/value data
-    assert "frontmatter" in result.output
-    assert "tags" in result.output
-    assert "test" in result.output
-    # The note content still appears
-    assert "hello world" in result.output
-    # The frontmatter block is NOT rendered a second time through Markdown:
-    # the raw fence is stripped, and the title key appears only in the panel.
-    assert "---" not in result.output
-    assert result.output.count("title") == 1
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_read_note",
-    new_callable=AsyncMock,
-    return_value=READ_NOTE_RESULT,
-)
-def test_read_note_rich_no_frontmatter_without_flag(mock_mcp):
-    """read-note WITHOUT --include-frontmatter must not render the frontmatter panel.
-
-    Regression (Bug 2): the JSON payload always contains a non-empty "frontmatter"
-    key, so the previous `if frontmatter:` guard rendered it even without the flag.
-    The flag must be threaded into _display_read_note to gate the panel.
-    """
-    result = _tty_runner(["tool", "read-note", "test-note"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    # The note title and content must still appear
-    assert "Test Note" in result.output
-    assert "hello world" in result.output
-    # The frontmatter panel must NOT appear
-    assert "frontmatter" not in result.output
-
-
-# ---------------------------------------------------------------------------
-# build-context – observations rendering (issue #678)
-# ---------------------------------------------------------------------------
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_build_context",
-    new_callable=AsyncMock,
-    return_value=BUILD_CONTEXT_RESULT,
-)
-def test_build_context_rich_renders_observations(mock_mcp):
-    """build-context Rich tree includes observations under each primary node.
-
-    Regression: ContextResult.observations was exposed in JSON output but never
-    rendered in the Rich path, so interactive users lost core entity facts.
-    """
-    result = _tty_runner(["tool", "build-context", "memory://notes/test-note"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    # The observation category should appear in the tree
-    assert "fact" in result.output
-    # The observation content should appear (possibly truncated)
-    assert "key fact" in result.output
-    # The subtitle should include an observations count
-    assert "observations" in result.output
-
-
-# ---------------------------------------------------------------------------
-# Rich markup injection – bracketed user text must survive (Bug 1, issue #678)
-# ---------------------------------------------------------------------------
-
-# Search result whose title contains a bracket expression like "[draft]".
-SEARCH_RESULT_BRACKETED_TITLE = {
-    "total": 1,
-    "current_page": 1,
-    "page_size": 10,
-    "has_more": False,
-    "results": [
-        {
-            "type": "entity",
-            "title": "Spec [draft] v2",
-            "permalink": "specs/spec-draft-v2",
-            "file_path": "specs/Spec [draft] v2.md",
-            "score": 0.90,
-            "matched_chunk": "An important [red] section",
-            "content": None,
-        },
-    ],
-}
-
-# build-context payload where the observation category is "fact" — previously
-# `[fact]` in the obs_label markup was interpreted as an unknown Rich tag and
-# the text was swallowed.
-BUILD_CONTEXT_BRACKETED_OBS = {
-    "results": [
-        {
-            "primary_result": {
-                "type": "entity",
-                "external_id": "xyz",
-                "title": "Joanna",
-                "permalink": "people/joanna",
-                "file_path": "people/Joanna.md",
-                "created_at": "2025-01-01T00:00:00",
-            },
-            "observations": [
-                {
-                    "type": "observation",
-                    "category": "fact",
-                    "content": "Joanna lives in Austin",
-                    "permalink": "people/joanna",
-                    "file_path": "people/Joanna.md",
-                    "created_at": "2025-01-01T00:00:00",
-                }
-            ],
-            "related_results": [],
-        }
-    ],
-    "metadata": {"uri": "people/joanna", "depth": 1},
-    "page": 1,
-    "page_size": 10,
-    "has_more": False,
-}
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT_BRACKETED_TITLE,
-)
-def test_search_notes_rich_title_with_brackets_survives(mock_mcp):
-    """Bracketed text in a search result title must appear literally in Rich output.
-
-    Regression (Bug 1): user-sourced titles were interpolated directly into Rich
-    markup strings, so "[draft]" was treated as an unknown style tag and stripped.
-    After escaping, the literal text "[draft]" must be present in the output.
-    """
-    result = _tty_runner(["tool", "search-notes", "spec"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    # The full title including the bracket expression must survive
-    assert "[draft]" in result.output
-    # The snippet "[red]" should also survive (not restyle the output)
-    assert "[red]" in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_build_context",
-    new_callable=AsyncMock,
-    return_value=BUILD_CONTEXT_BRACKETED_OBS,
-)
-def test_build_context_rich_observation_category_bracket_survives(mock_mcp):
-    """Observation category "[fact]" must appear literally in build-context Rich tree.
-
-    Regression (Bug 1): the obs_label was built as f"[dim][{category}] content[/dim]",
-    which caused the inner "[fact]" to be parsed as an unknown Rich tag and dropped,
-    rendering "Joanna lives in Austin" without the category prefix.
-    After escaping the category, "[fact]" must be present in the tree output.
-    """
-    result = _tty_runner(["tool", "build-context", "memory://people/joanna"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    # The observation category prefix must appear literally
-    assert "[fact]" in result.output
-    # The observation content must also appear
-    assert "Joanna lives in Austin" in result.output
-
-
-# ---------------------------------------------------------------------------
-# search-notes – total=0 with non-empty results subtitle (Bug 3, issue #678)
-# ---------------------------------------------------------------------------
-
-# Fixture that mirrors the upstream quirk: total=0 but results is non-empty.
-SEARCH_RESULT_ZERO_TOTAL = {
-    "total": 0,
-    "current_page": 1,
-    "page_size": 10,
-    "has_more": False,
-    "results": [
-        {
-            "type": "entity",
-            "title": "Found Note",
-            "permalink": "notes/found-note",
-            "file_path": "notes/Found Note.md",
-            "score": 0.80,
-            "matched_chunk": "some content",
-            "content": None,
-        },
-        {
-            "type": "entity",
-            "title": "Another Found",
-            "permalink": "notes/another-found",
-            "file_path": "notes/Another Found.md",
-            "score": 0.70,
-            "matched_chunk": "more content",
-            "content": None,
-        },
-    ],
-}
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT_ZERO_TOTAL,
-)
-def test_search_notes_rich_zero_total_falls_back_to_result_count(mock_mcp):
-    """When the API returns total=0 but results is non-empty, subtitle shows real count.
-
-    Regression (Bug 3): result.get("total", len(results)) never triggered its
-    default because the "total" key exists (with value 0), so the subtitle read
-    "0 result(s)" under a table showing rows.  The fix detects a falsy total with
-    non-empty results and falls back to len(results).
-    """
-    result = _tty_runner(["tool", "search-notes", "found"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    # Both result rows must appear
-    assert "Found Note" in result.output
-    assert "Another Found" in result.output
-    # The subtitle must show the real count (2), not 0
-    assert "2 result(s)" in result.output
-    assert "0 result(s)" not in result.output
-
-
-# ---------------------------------------------------------------------------
-# --plain output mode (issue #678 follow-up)
-# ---------------------------------------------------------------------------
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT,
-)
-def test_search_notes_plain_output(mock_mcp):
-    """search-notes --plain emits undecorated numbered text, not JSON or Rich boxes."""
-    result = _tty_runner(["tool", "search-notes", "test", "--plain"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    # Not JSON
-    with pytest.raises((json.JSONDecodeError, ValueError)):
-        json.loads(result.output)
-    # Numbered, greppable entries with titles, scores, and permalinks
-    assert "1. Test Note" in result.output
-    assert "2. Another Note" in result.output
-    assert "notes/test-note" in result.output
-    assert "0.95" in result.output
-    # Snippet line present
-    assert "A snippet about test notes" in result.output
-    # No Rich box-drawing characters
-    assert "─" not in result.output
-    assert "│" not in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT_BRACKETED_TITLE,
-)
-def test_search_notes_plain_brackets_survive(mock_mcp):
-    """Literal brackets must survive verbatim in plain output (no markup escaping)."""
-    result = _tty_runner(["tool", "search-notes", "spec", "--plain"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    # The literal title and snippet brackets must be present unmangled
-    assert "Spec [draft] v2" in result.output
-    assert "[red]" in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT_ZERO_TOTAL,
-)
-def test_search_notes_plain_zero_total_fallback(mock_mcp):
-    """Plain search output shows the corrected count when the API returns total=0."""
-    result = _tty_runner(["tool", "search-notes", "found", "--plain"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "2 result(s)" in result.output
-    assert "0 result(s)" not in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT_EMPTY,
-)
-def test_search_notes_plain_empty(mock_mcp):
-    """Plain search output handles empty results."""
-    result = _tty_runner(["tool", "search-notes", "nothing", "--plain"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "No results found." in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_read_note",
-    new_callable=AsyncMock,
-    return_value=READ_NOTE_RESULT,
-)
-def test_read_note_plain_output(mock_mcp):
-    """read-note --plain emits the note body faithfully, with no decoration."""
-    result = _tty_runner(["tool", "read-note", "test-note", "--plain"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    # No synthesized header line -- plain mode is the payload, faithfully.
-    assert "[notes/test-note]" not in result.output
-    # The body verbatim, trimmed of the API's frontmatter-strip newline artifact.
-    assert result.output == "# Test Note\n\nhello world\n"
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_read_note",
-    new_callable=AsyncMock,
-    return_value=READ_NOTE_RESULT_WITH_FRONTMATTER,
-)
-def test_read_note_plain_include_frontmatter(mock_mcp):
-    """read-note --plain --include-frontmatter shows the literal file, once.
-
-    With the flag, content IS the file (frontmatter block included); plain mode
-    prints it verbatim and must not prepend a synthesized key/value block.
-    """
-    result = _tty_runner(["tool", "read-note", "test-note", "--plain", "--frontmatter"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    # ONLY the literal file: no header line, output starts at the fence
-    assert "Test Note  [notes/test-note]" not in result.output
-    assert result.output.startswith("---\ntitle: Test Note\ntags:\n- test\n---")
-    assert "hello world" in result.output
-    # No duplicated frontmatter from a synthesized block or header
-    assert result.output.count("title: Test Note") == 1
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_read_note",
-    new_callable=AsyncMock,
-    return_value=READ_NOTE_RESULT_WITH_FRONTMATTER,
-)
-def test_read_note_include_frontmatter_alias(mock_mcp):
-    """--include-frontmatter still works as a deprecated alias for --frontmatter."""
-    result = _tty_runner(["tool", "read-note", "test-note", "--plain", "--include-frontmatter"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert result.output.startswith("---\ntitle: Test Note")
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_read_note",
-    new_callable=AsyncMock,
-    return_value={"title": "", "permalink": "", "content": "", "frontmatter": {}},
-)
-def test_read_note_plain_empty_content(mock_mcp):
-    """read-note --plain prints nothing for an empty note (no placeholder)."""
-    result = _tty_runner(["tool", "read-note", "empty-note", "--plain"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert result.output == ""
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_build_context",
-    new_callable=AsyncMock,
-    return_value=BUILD_CONTEXT_RESULT,
-)
-def test_build_context_plain_output(mock_mcp):
-    """build-context --plain emits an ASCII-indented outline with observations."""
-    result = _tty_runner(["tool", "build-context", "memory://notes/test-note", "--plain"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "Context: notes/test-note" in result.output
-    assert "Test Note" in result.output
-    # Observation rendered with literal bracketed category, indented
-    assert "  [fact] This is a key fact about the test note" in result.output
-    # Related item with relation type, indented
-    assert "Related Note" in result.output
-    assert "references" in result.output
-    # Not JSON
-    with pytest.raises((json.JSONDecodeError, ValueError)):
-        json.loads(result.output)
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_build_context",
-    new_callable=AsyncMock,
-    return_value=BUILD_CONTEXT_BRACKETED_OBS,
-)
-def test_build_context_plain_bracket_survives(mock_mcp):
-    """Observation category [fact] must appear literally in plain build-context output."""
-    result = _tty_runner(["tool", "build-context", "memory://people/joanna", "--plain"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "[fact]" in result.output
-    assert "Joanna lives in Austin" in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_build_context",
-    new_callable=AsyncMock,
-    return_value=BUILD_CONTEXT_EMPTY,
-)
-def test_build_context_plain_empty(mock_mcp):
-    """build-context --plain handles empty results."""
-    result = _tty_runner(["tool", "build-context", "memory://notes/test-note", "--plain"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "No related content found." in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_recent_activity",
-    new_callable=AsyncMock,
-    return_value=RECENT_ACTIVITY_RESULT,
-)
-def test_recent_activity_plain_output(mock_mcp):
-    """recent-activity --plain emits "- title (type) permalink updated" lines."""
-    result = _tty_runner(["tool", "recent-activity", "--plain"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "- Note A (entity) notes/note-a" in result.output
-    assert "- Note B (entity) notes/note-b" in result.output
-    # Not JSON
-    with pytest.raises((json.JSONDecodeError, ValueError)):
-        json.loads(result.output)
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_recent_activity",
-    new_callable=AsyncMock,
-    return_value=[],
-)
-def test_recent_activity_plain_empty(mock_mcp):
-    """recent-activity --plain handles empty results."""
-    result = _tty_runner(["tool", "recent-activity", "--plain"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    assert "No recent activity." in result.output
-
-
-# ---------------------------------------------------------------------------
-# Precedence matrix: --json > --plain > non-TTY (JSON) > TTY (config style)
-# ---------------------------------------------------------------------------
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT,
-)
-def test_json_beats_plain(mock_mcp):
-    """--json wins over --plain... when they are not used together, --json alone is JSON.
-
-    The contradictory combination is tested separately (must error); here we
-    confirm --json on its own produces JSON even in a TTY.
-    """
-    result = _tty_runner(["tool", "search-notes", "test", "--json"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    data = json.loads(result.output)
-    assert data["total"] == 2
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT,
-)
-def test_json_and_plain_together_errors(mock_mcp):
-    """Passing both --json and --plain is a clear, non-zero error."""
-    result = _tty_runner(["tool", "search-notes", "test", "--json", "--plain"])
-
-    assert result.exit_code != 0
-    assert "mutually exclusive" in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_read_note",
-    new_callable=AsyncMock,
-    return_value=READ_NOTE_RESULT,
-)
-def test_read_note_json_and_plain_together_errors(mock_mcp):
-    """read-note also rejects --json --plain together."""
-    result = _tty_runner(["tool", "read-note", "test-note", "--json", "--plain"])
-
-    assert result.exit_code != 0
-    assert "mutually exclusive" in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT,
-)
-def test_plain_forces_plain_when_piped(mock_mcp):
-    """--plain forces plain output even when stdout is NOT a TTY (piped)."""
-    # No _use_rich patch: the default CliRunner stdout is not a TTY, so absent
-    # --plain this would be JSON.  --plain must override that into plain text.
-    result = runner.invoke(cli_app, ["tool", "search-notes", "test", "--plain"])
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    with pytest.raises((json.JSONDecodeError, ValueError)):
-        json.loads(result.output)
-    assert "1. Test Note" in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT,
-)
-def test_tty_config_rich_renders_rich(mock_mcp):
-    """TTY + cli_output_style=rich → Rich output (box-drawing present)."""
-    result = _tty_runner_with_style(["tool", "search-notes", "test"], style="rich")
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    with pytest.raises((json.JSONDecodeError, ValueError)):
-        json.loads(result.output)
-    # Rich draws a Panel border
-    assert "─" in result.output or "│" in result.output
-
-
-@patch(
-    "basic_memory.cli.commands.tool.mcp_search",
-    new_callable=AsyncMock,
-    return_value=SEARCH_RESULT,
-)
-def test_tty_config_plain_renders_plain(mock_mcp):
-    """TTY + cli_output_style=plain → plain output (no box-drawing)."""
-    result = _tty_runner_with_style(["tool", "search-notes", "test"], style="plain")
-
-    assert result.exit_code == 0, f"CLI failed: {result.output}"
-    with pytest.raises((json.JSONDecodeError, ValueError)):
-        json.loads(result.output)
-    assert "1. Test Note" in result.output
-    assert "─" not in result.output
-    assert "│" not in result.output

tests/cli/test_share_commands.py

@@ -0,0 +1,999 @@
+"""Tests for cloud share CLI commands.
+
+Issue #880: Tests for share create, list, update, revoke commands that surface
+the cloud /api/shares endpoints.
+"""
+
+from unittest.mock import AsyncMock, Mock, patch
+
+import httpx
+from typer.testing import CliRunner
+
+from basic_memory.cli.app import app
+from basic_memory.cli.commands.cloud.api_client import (
+    CloudAPIError,
+    SubscriptionRequiredError,
+)
+from basic_memory.schemas.cloud import WorkspaceInfo
+
+# A real workspace/tenant UUID is forwarded verbatim as X-Workspace-ID with no
+# workspace lookup; the cloud's resolver only accepts this UUID form.
+TENANT_UUID = "5ccbae40-ca03-43a2-b23d-9931eb130e22"
+
+
+def _workspace(slug: str, tenant_id: str, name: str) -> WorkspaceInfo:
+    """Build a WorkspaceInfo for workspace-resolution tests."""
+    return WorkspaceInfo(
+        tenant_id=tenant_id,
+        workspace_type="organization",
+        slug=slug,
+        name=name,
+        role="owner",
+        is_default=False,
+    )
+
+
+def _patch_available_workspaces(workspaces):
+    """Patch the workspace list fetch used when --workspace is a slug/name.
+
+    Asserts can wrap the returned mock to confirm the lookup was (or was not)
+    performed for a given invocation.
+    """
+    return patch(
+        "basic_memory.mcp.project_context.get_available_workspaces",
+        new=AsyncMock(return_value=workspaces),
+    )
+
+
+SHARE_RESPONSE = {
+    "id": "11111111-1111-1111-1111-111111111111",
+    "token": "abc123",
+    "project_name": "my-project",
+    "note_permalink": "notes/my-idea",
+    "note_external_id": "ext-1",
+    "enabled": True,
+    "expires_at": None,
+    "share_url": "https://share.example.com/abc123",
+    "view_count": 0,
+    "last_viewed_at": None,
+    "created_at": "2025-01-18T12:00:00Z",
+}
+
+
+def _mock_config_manager():
+    mock_config = Mock()
+    mock_config.cloud_host = "https://cloud.example.com"
+    mock_config_manager = Mock()
+    mock_config_manager.config = mock_config
+    return mock_config_manager
+
+
+def _patch_workspace(resolved):
+    """Patch the workspace resolver used by the share commands.
+
+    Returns whatever ``resolved`` is for every lookup, so tests can assert the
+    X-Workspace-ID header is built (and routed) the way the cloud expects
+    without depending on real config files.
+    """
+    return patch(
+        "basic_memory.cli.commands.cloud.shares.resolve_configured_workspace",
+        return_value=resolved,
+    )
+
+
+class TestShareCreateCommand:
+    """Tests for 'bm cloud share create' command."""
+
+    def test_create_share_success(self):
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = SHARE_RESPONSE
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["json_data"] = kwargs.get("json_data")
+            captured["headers"] = kwargs.get("headers")
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with _patch_workspace(TENANT_UUID):
+                    with _patch_available_workspaces([]) as fetch:
+                        result = runner.invoke(
+                            app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
+                        )
+
+                assert result.exit_code == 0
+                assert "Share link created successfully" in result.stdout
+                assert "abc123" in result.stdout
+                assert "https://share.example.com/abc123" in result.stdout
+                # Payload should match the cloud CreateShareRequest contract.
+                assert captured["json_data"] == {
+                    "project_name": "my-project",
+                    "note_permalink": "notes/my-idea",
+                }
+                # Workspace routing: a resolved tenant UUID travels verbatim as the
+                # X-Workspace-ID header so team-workspace projects aren't evaluated
+                # against the caller's default tenant.
+                assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
+                # A UUID needs no resolution: the workspace list is never fetched.
+                fetch.assert_not_called()
+
+    def test_create_share_slug_resolves_to_tenant_uuid(self):
+        """A --workspace slug is resolved to the tenant UUID before routing."""
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = SHARE_RESPONSE
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["headers"] = kwargs.get("headers")
+            return mock_response
+
+        seen = {}
+
+        def fake_resolve(*, project_name=None, workspace=None):
+            seen["project_name"] = project_name
+            seen["workspace"] = workspace
+            return workspace
+
+        workspaces = [
+            _workspace("basic-memory-7020de4e925843c68c9056c60d101d9e", TENANT_UUID, "Acme Org"),
+            _workspace("other-slug", "11111111-1111-1111-1111-111111111111", "Other"),
+        ]
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with patch(
+                    "basic_memory.cli.commands.cloud.shares.resolve_configured_workspace",
+                    side_effect=fake_resolve,
+                ):
+                    with _patch_available_workspaces(workspaces) as fetch:
+                        result = runner.invoke(
+                            app,
+                            [
+                                "cloud",
+                                "share",
+                                "create",
+                                "my-project",
+                                "notes/my-idea",
+                                "--workspace",
+                                "basic-memory-7020de4e925843c68c9056c60d101d9e",
+                            ],
+                        )
+
+                assert result.exit_code == 0
+                assert seen == {
+                    "project_name": "my-project",
+                    "workspace": "basic-memory-7020de4e925843c68c9056c60d101d9e",
+                }
+                # The slug was mapped to the workspace's tenant UUID.
+                assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
+                fetch.assert_awaited_once()
+
+    def test_create_share_display_name_resolves_case_insensitively(self):
+        """A --workspace display name resolves case-insensitively to the tenant UUID."""
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = SHARE_RESPONSE
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["headers"] = kwargs.get("headers")
+            return mock_response
+
+        workspaces = [_workspace("acme-slug", TENANT_UUID, "Acme Org")]
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with _patch_workspace("acme org"):
+                    with _patch_available_workspaces(workspaces):
+                        result = runner.invoke(
+                            app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
+                        )
+
+                assert result.exit_code == 0
+                assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
+
+    def test_create_share_tenant_id_input_passthrough(self):
+        """A tenant UUID resolved from config is forwarded without a lookup."""
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = SHARE_RESPONSE
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["headers"] = kwargs.get("headers")
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with _patch_workspace(TENANT_UUID):
+                    with _patch_available_workspaces([]) as fetch:
+                        result = runner.invoke(
+                            app,
+                            [
+                                "cloud",
+                                "share",
+                                "create",
+                                "my-project",
+                                "notes/my-idea",
+                                "--workspace",
+                                TENANT_UUID,
+                            ],
+                        )
+
+                assert result.exit_code == 0
+                assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
+                fetch.assert_not_called()
+
+    def test_create_share_ambiguous_name_errors_with_candidate_slugs(self):
+        """A display name matching multiple workspaces errors and lists candidates."""
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):  # pragma: no cover
+            raise AssertionError("API should not be called when workspace is ambiguous")
+
+        workspaces = [
+            _workspace("acme-prod", TENANT_UUID, "Acme"),
+            _workspace("acme-staging", "11111111-1111-1111-1111-111111111111", "Acme"),
+        ]
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with _patch_workspace("Acme"):
+                    with _patch_available_workspaces(workspaces):
+                        result = runner.invoke(
+                            app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
+                        )
+
+                assert result.exit_code == 1
+                assert "ambiguous" in result.stdout
+                # Candidate slugs are listed so the user can disambiguate.
+                assert "acme-prod" in result.stdout
+                assert "acme-staging" in result.stdout
+                # The typer.Exit must not be re-wrapped by the broad handler.
+                assert "Unexpected error" not in result.stdout
+
+    def test_create_share_unknown_workspace_errors_with_available_slugs(self):
+        """An unknown identifier errors and lists the available workspace slugs."""
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):  # pragma: no cover
+            raise AssertionError("API should not be called for an unknown workspace")
+
+        workspaces = [
+            _workspace("acme-prod", TENANT_UUID, "Acme"),
+            _workspace("widget-co", "11111111-1111-1111-1111-111111111111", "Widget Co"),
+        ]
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with _patch_workspace("does-not-exist"):
+                    with _patch_available_workspaces(workspaces):
+                        result = runner.invoke(
+                            app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
+                        )
+
+                assert result.exit_code == 1
+                assert "was not found" in result.stdout
+                assert "acme-prod" in result.stdout
+                assert "widget-co" in result.stdout
+                assert "Unexpected error" not in result.stdout
+
+    def test_create_share_no_workspace_sends_no_header(self):
+        """When nothing resolves, no routing header is added (default tenant)."""
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = SHARE_RESPONSE
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["headers"] = kwargs.get("headers")
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with _patch_workspace(None):
+                    result = runner.invoke(
+                        app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
+                    )
+
+                assert result.exit_code == 0
+                assert captured["headers"] == {}
+
+    def test_create_share_with_expires_at(self):
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = SHARE_RESPONSE
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["json_data"] = kwargs.get("json_data")
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(
+                    app,
+                    [
+                        "cloud",
+                        "share",
+                        "create",
+                        "my-project",
+                        "notes/my-idea",
+                        "--expires-at",
+                        "2025-12-31",
+                    ],
+                )
+
+                assert result.exit_code == 0
+                assert captured["json_data"]["expires_at"].startswith("2025-12-31")
+
+    def test_create_share_invalid_expires_at(self):
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):  # pragma: no cover
+            raise AssertionError("API should not be called on invalid input")
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(
+                    app,
+                    [
+                        "cloud",
+                        "share",
+                        "create",
+                        "my-project",
+                        "notes/my-idea",
+                        "--expires-at",
+                        "not-a-date",
+                    ],
+                )
+
+                assert result.exit_code == 1
+                assert "Invalid --expires-at" in result.stdout
+                # A parse error must produce a single clean message, not a
+                # spurious "Unexpected error: 1" from the broad handler
+                # re-catching typer.Exit. See issue #880 review.
+                assert "Unexpected error" not in result.stdout
+
+    def test_create_share_note_not_found(self):
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):
+            raise CloudAPIError("Not found", status_code=404)
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(
+                    app, ["cloud", "share", "create", "my-project", "notes/missing"]
+                )
+
+                assert result.exit_code == 1
+                assert "Note not found" in result.stdout
+
+    def test_create_share_subscription_required(self):
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):
+            raise SubscriptionRequiredError(
+                message="Active subscription required",
+                subscribe_url="https://basicmemory.com/subscribe",
+            )
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(
+                    app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
+                )
+
+                assert result.exit_code == 1
+                assert "Subscription Required" in result.stdout
+
+    def test_create_share_api_error(self):
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):
+            raise CloudAPIError("Server error", status_code=500)
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(
+                    app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
+                )
+
+                assert result.exit_code == 1
+                assert "Failed to create share link" in result.stdout
+
+
+class TestShareListCommand:
+    """Tests for 'bm cloud share list' command."""
+
+    def test_list_shares_success(self):
+        # Wide terminal so the rich table doesn't truncate cell contents.
+        runner = CliRunner(env={"COLUMNS": "200"})
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "shares": [
+                SHARE_RESPONSE,
+                {
+                    **SHARE_RESPONSE,
+                    "token": "def456",
+                    "note_permalink": "notes/second",
+                    "enabled": False,
+                    "expires_at": "2025-12-31T00:00:00Z",
+                    "view_count": 7,
+                },
+            ],
+            "total": 2,
+        }
+
+        async def mock_make_api_request(*args, **kwargs):
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(app, ["cloud", "share", "list"])
+
+                assert result.exit_code == 0
+                assert "abc123" in result.stdout
+                assert "def456" in result.stdout
+                assert "notes/second" in result.stdout
+
+    def test_list_shares_empty(self):
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = {"shares": [], "total": 0}
+
+        async def mock_make_api_request(*args, **kwargs):
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(app, ["cloud", "share", "list"])
+
+                assert result.exit_code == 0
+                assert "No share links found" in result.stdout
+
+    def test_list_shares_with_project_filter(self):
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = {"shares": [SHARE_RESPONSE], "total": 1}
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["url"] = kwargs.get("url", args[1] if len(args) > 1 else "")
+            captured["headers"] = kwargs.get("headers")
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with _patch_workspace(TENANT_UUID):
+                    result = runner.invoke(
+                        app, ["cloud", "share", "list", "--project", "my-project"]
+                    )
+
+                assert result.exit_code == 0
+                assert "project_name=my-project" in captured["url"]
+                assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
+
+    def test_list_shares_unknown_workspace_errors_without_double_error(self):
+        """An unknown --workspace on list errors cleanly (no 'Unexpected error').
+
+        Exercises the list handler's typer.Exit re-raise: workspace resolution
+        raises typer.Exit, which must not be re-wrapped by the broad handler.
+        """
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):  # pragma: no cover
+            raise AssertionError("API should not be called for an unknown workspace")
+
+        workspaces = [_workspace("acme-prod", TENANT_UUID, "Acme")]
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with _patch_workspace("does-not-exist"):
+                    with _patch_available_workspaces(workspaces):
+                        result = runner.invoke(app, ["cloud", "share", "list"])
+
+                assert result.exit_code == 1
+                assert "was not found" in result.stdout
+                assert "acme-prod" in result.stdout
+                assert "Unexpected error" not in result.stdout
+
+    def test_list_shares_ambiguous_slug_errors_with_candidates(self):
+        """A slug colliding across workspaces errors and lists candidates.
+
+        Exercises the slug tier of _match_workspace_identifier raising on >1 match.
+        """
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):  # pragma: no cover
+            raise AssertionError("API should not be called when the slug is ambiguous")
+
+        workspaces = [
+            _workspace("shared-slug", TENANT_UUID, "Acme"),
+            _workspace("shared-slug", "11111111-1111-1111-1111-111111111111", "Widget"),
+        ]
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with _patch_workspace("shared-slug"):
+                    with _patch_available_workspaces(workspaces):
+                        result = runner.invoke(app, ["cloud", "share", "list"])
+
+                assert result.exit_code == 1
+                assert "ambiguous" in result.stdout
+                assert TENANT_UUID in result.stdout
+                assert "Unexpected error" not in result.stdout
+
+    def test_list_shares_project_filter_url_encoded(self):
+        """Project names with query-reserved chars must be percent-encoded.
+
+        A name like "R&D+notes #1" interpolated raw would split into bogus
+        query params (project_name=R, plus a stray "D+notes #1" key); encoding
+        keeps it a single faithful project_name value.
+        """
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = {"shares": [SHARE_RESPONSE], "total": 1}
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["url"] = kwargs.get("url", args[1] if len(args) > 1 else "")
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with _patch_workspace(None):
+                    result = runner.invoke(
+                        app, ["cloud", "share", "list", "--project", "R&D+notes #1"]
+                    )
+
+                assert result.exit_code == 0
+                # Reserved characters are percent-encoded into a single value.
+                assert "project_name=R%26D%2Bnotes+%231" in captured["url"]
+                # And the raw, ambiguous form never reaches the wire.
+                assert "project_name=R&D" not in captured["url"]
+
+    def test_list_shares_api_error(self):
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):
+            raise CloudAPIError("Server error", status_code=500)
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(app, ["cloud", "share", "list"])
+
+                assert result.exit_code == 1
+                assert "Failed to list share links" in result.stdout
+
+
+class TestShareUpdateCommand:
+    """Tests for 'bm cloud share update' command."""
+
+    def test_update_disable(self):
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = {**SHARE_RESPONSE, "enabled": False}
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["json_data"] = kwargs.get("json_data")
+            captured["method"] = kwargs.get("method")
+            captured["headers"] = kwargs.get("headers")
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with _patch_workspace(TENANT_UUID):
+                    result = runner.invoke(
+                        app,
+                        ["cloud", "share", "update", "abc123", "--disable", "--workspace", "acme"],
+                    )
+
+                assert result.exit_code == 0
+                assert "updated successfully" in result.stdout
+                assert captured["method"] == "PATCH"
+                assert captured["json_data"] == {"enabled": False}
+                assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
+
+    def test_update_enable(self):
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = SHARE_RESPONSE
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["json_data"] = kwargs.get("json_data")
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(app, ["cloud", "share", "update", "abc123", "--enable"])
+
+                assert result.exit_code == 0
+                assert captured["json_data"] == {"enabled": True}
+
+    def test_update_expires_at(self):
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = SHARE_RESPONSE
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["json_data"] = kwargs.get("json_data")
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(
+                    app,
+                    ["cloud", "share", "update", "abc123", "--expires-at", "2026-01-01"],
+                )
+
+                assert result.exit_code == 0
+                assert captured["json_data"]["expires_at"].startswith("2026-01-01")
+
+    def test_update_clear_expires_at(self):
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 200
+        mock_response.json.return_value = SHARE_RESPONSE
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["json_data"] = kwargs.get("json_data")
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(
+                    app, ["cloud", "share", "update", "abc123", "--expires-at", "none"]
+                )
+
+                assert result.exit_code == 0
+                assert captured["json_data"] == {"expires_at": None}
+
+    def test_update_enable_and_disable_conflict(self):
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):  # pragma: no cover
+            raise AssertionError("API should not be called on conflicting flags")
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(
+                    app,
+                    ["cloud", "share", "update", "abc123", "--enable", "--disable"],
+                )
+
+                assert result.exit_code == 1
+                assert "Cannot use --enable and --disable together" in result.stdout
+
+    def test_update_nothing_to_change(self):
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):  # pragma: no cover
+            raise AssertionError("API should not be called with empty update")
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(app, ["cloud", "share", "update", "abc123"])
+
+                assert result.exit_code == 1
+                assert "Nothing to update" in result.stdout
+
+    def test_update_not_found(self):
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):
+            raise CloudAPIError("Not found", status_code=404)
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(app, ["cloud", "share", "update", "missing", "--disable"])
+
+                assert result.exit_code == 1
+                assert "Share not found" in result.stdout
+
+
+class TestShareRevokeCommand:
+    """Tests for 'bm cloud share revoke' command."""
+
+    def test_revoke_success_with_force(self):
+        runner = CliRunner()
+
+        mock_response = Mock(spec=httpx.Response)
+        mock_response.status_code = 204
+        mock_response.json.return_value = {}
+
+        captured = {}
+
+        async def mock_make_api_request(*args, **kwargs):
+            captured["method"] = kwargs.get("method")
+            captured["url"] = kwargs.get("url")
+            captured["headers"] = kwargs.get("headers")
+            return mock_response
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                with _patch_workspace(TENANT_UUID):
+                    result = runner.invoke(
+                        app,
+                        ["cloud", "share", "revoke", "abc123", "--force", "--workspace", "acme"],
+                    )
+
+                assert result.exit_code == 0
+                assert "revoked successfully" in result.stdout
+                assert captured["method"] == "DELETE"
+                assert captured["url"].endswith("/api/shares/abc123")
+                assert captured["headers"] == {"X-Workspace-ID": TENANT_UUID}
+
+    def test_revoke_cancelled(self):
+        runner = CliRunner()
+
+        call_count = 0
+
+        async def mock_make_api_request(*args, **kwargs):
+            nonlocal call_count
+            call_count += 1
+            return Mock(spec=httpx.Response)
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(app, ["cloud", "share", "revoke", "abc123"], input="n\n")
+
+                assert result.exit_code == 0
+                assert "cancelled" in result.stdout
+                assert call_count == 0
+
+    def test_revoke_not_found(self):
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):
+            raise CloudAPIError("Not found", status_code=404)
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(app, ["cloud", "share", "revoke", "missing", "--force"])
+
+                assert result.exit_code == 1
+                assert "Share not found" in result.stdout
+
+    def test_revoke_subscription_required(self):
+        runner = CliRunner()
+
+        async def mock_make_api_request(*args, **kwargs):
+            raise SubscriptionRequiredError(
+                message="Active subscription required",
+                subscribe_url="https://basicmemory.com/subscribe",
+            )
+
+        with patch(
+            "basic_memory.cli.commands.cloud.shares.make_api_request",
+            side_effect=mock_make_api_request,
+        ):
+            with patch(
+                "basic_memory.cli.commands.cloud.shares.ConfigManager",
+                return_value=_mock_config_manager(),
+            ):
+                result = runner.invoke(app, ["cloud", "share", "revoke", "abc123", "--force"])
+
+                assert result.exit_code == 1
+                assert "Subscription Required" in result.stdout