feat(cli): rename read-note --include-frontmatter to --frontmatter

Keep --include-frontmatter as a deprecated alias for back-compat (the old
name shipped in 0.22.0). The MCP tool parameter include_frontmatter is
unchanged.

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

.claude-plugin/marketplace.json

@@ -6,14 +6,14 @@
   },
   "metadata": {
     "description": "Official Basic Memory plugins from the canonical basic-memory repository",
-    "version": "0.21.6"
+    "version": "0.22.0"
   },
   "plugins": [
     {
       "name": "basic-memory",
       "source": "./plugins/claude-code",
       "description": "The bridge between Claude's working memory and Basic Memory's durable knowledge graph \u2014 session briefings, pre-compaction checkpoints, and capture reflexes",
-      "version": "0.21.6",
+      "version": "0.22.0",
       "author": {
         "name": "Basic Machines"
       },

.github/workflows/bm-bossbot.yml

@@ -1,193 +0,0 @@
-name: BM Bossbot
-
-"on":
-  workflow_run:
-    workflows:
-      - Tests
-    types:
-      - completed
-  workflow_dispatch:
-    inputs:
-      pr_number:
-        description: Pull request number to review
-        required: true
-  # Review-thread activity re-evaluates the approval status without re-running
-  # the full LLM review: new feedback flips the gate to failure, resolving the
-  # last thread restores a previously earned approval for the same head SHA.
-  pull_request_review:
-    types:
-      - submitted
-  pull_request_review_comment:
-    types:
-      - created
-  pull_request_review_thread:
-    types:
-      - resolved
-      - unresolved
-
-permissions:
-  contents: read
-  pull-requests: write
-  statuses: write
-  issues: read
-
-env:
-  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
-  BM_BOSSBOT_STATUS_CONTEXT: "BM Bossbot Approval"
-
-jobs:
-  review:
-    name: BM Bossbot Review
-    # Job-level concurrency (not workflow-level): thread-recheck events for the
-    # same PR must never cancel an in-flight review run, and vice versa.
-    concurrency:
-      group: bm-bossbot-review-${{ github.event.workflow_run.pull_requests[0].number || inputs.pr_number }}
-      cancel-in-progress: true
-    if: |
-      github.event_name == 'workflow_dispatch' ||
-      (
-        github.event_name == 'workflow_run' &&
-        github.event.workflow_run.conclusion == 'success' &&
-        github.event.workflow_run.pull_requests[0].number != ''
-      )
-    runs-on: ubuntu-latest
-    outputs:
-      pr_number: ${{ steps.pr.outputs.pr_number }}
-      should_review: ${{ steps.pr.outputs.should_review }}
-
-    steps:
-      - name: Checkout trusted base ref
-        uses: actions/checkout@v6
-        with:
-          ref: ${{ github.event.repository.default_branch }}
-          fetch-depth: 1
-
-      - name: Set up uv
-        uses: astral-sh/setup-uv@v3
-
-      - name: Normalize PR event
-        id: pr
-        env:
-          GH_TOKEN: ${{ github.token }}
-        run: |
-          set -euo pipefail
-          event_file="${RUNNER_TEMP}/bm-bossbot-event.json"
-          should_review=true
-
-          if [ "${GITHUB_EVENT_NAME}" = "workflow_dispatch" ]; then
-            pr_number="${{ inputs.pr_number }}"
-            tested_sha=""
-          else
-            pr_number="$(jq -r '.workflow_run.pull_requests[0].number // ""' "${GITHUB_EVENT_PATH}")"
-            tested_sha="$(jq -r '.workflow_run.head_sha // ""' "${GITHUB_EVENT_PATH}")"
-          fi
-
-          gh api "repos/${GITHUB_REPOSITORY}/pulls/${pr_number}" > "${RUNNER_TEMP}/pull.json"
-          current_head_sha="$(jq -r '.head.sha' "${RUNNER_TEMP}/pull.json")"
-          draft="$(jq -r '.draft' "${RUNNER_TEMP}/pull.json")"
-
-          if [ "${GITHUB_EVENT_NAME}" = "workflow_dispatch" ]; then
-            tests_run_id="$(
-              gh api -X GET "repos/${GITHUB_REPOSITORY}/actions/workflows/test.yml/runs" \
-                -f event=push \
-                -f head_sha="${current_head_sha}" \
-                -f status=completed \
-                --jq '[.workflow_runs[] | select(.conclusion == "success")][0].id // ""'
-            )"
-            if [ -z "${tests_run_id}" ]; then
-              should_review=false
-              echo "BM Bossbot skipped PR ${pr_number}: no successful Tests workflow for ${current_head_sha}."
-            fi
-          fi
-
-          if [ "${draft}" = "true" ]; then
-            should_review=false
-            echo "BM Bossbot skipped PR ${pr_number}: draft pull request."
-          fi
-
-          if [ -n "${tested_sha}" ] && [ "${tested_sha}" != "${current_head_sha}" ]; then
-            should_review=false
-            echo "BM Bossbot skipped PR ${pr_number}: Tests passed for ${tested_sha}, but current head is ${current_head_sha}."
-          fi
-
-          jq --arg repo "${GITHUB_REPOSITORY}" \
-            '{repository:{full_name:$repo}, pull_request:{number:.number,title:.title,body:(.body // ""),html_url:.html_url,head:{sha:.head.sha,ref:.head.ref},base:{ref:.base.ref,sha:.base.sha},author_association:.author_association,draft:.draft}}' \
-            "${RUNNER_TEMP}/pull.json" > "${event_file}"
-
-          echo "event_file=${event_file}" >> "${GITHUB_OUTPUT}"
-          echo "pr_number=$(jq -r '.pull_request.number' "${event_file}")" >> "${GITHUB_OUTPUT}"
-          echo "head_sha=$(jq -r '.pull_request.head.sha' "${event_file}")" >> "${GITHUB_OUTPUT}"
-          echo "head_ref=$(jq -r '.pull_request.head.ref' "${event_file}")" >> "${GITHUB_OUTPUT}"
-          echo "author_association=$(jq -r '.pull_request.author_association // ""' "${event_file}")" >> "${GITHUB_OUTPUT}"
-          echo "tested_sha=${tested_sha}" >> "${GITHUB_OUTPUT}"
-          echo "should_review=${should_review}" >> "${GITHUB_OUTPUT}"
-
-      - name: Classify PR author
-        id: trust
-        if: steps.pr.outputs.should_review == 'true'
-        env:
-          AUTHOR_ASSOCIATION: ${{ steps.pr.outputs.author_association }}
-        run: |
-          set -euo pipefail
-          case "${AUTHOR_ASSOCIATION}" in
-            OWNER|MEMBER|COLLABORATOR)
-              trusted_author=true
-              ;;
-            *)
-              trusted_author=false
-              ;;
-          esac
-          echo "trusted_author=${trusted_author}" >> "${GITHUB_OUTPUT}"
-          echo "author_association=${AUTHOR_ASSOCIATION}" >> "${GITHUB_OUTPUT}"
-
-      - name: Mark BM Bossbot approval pending
-        if: steps.pr.outputs.should_review == 'true'
-        env:
-          GITHUB_TOKEN: ${{ github.token }}
-        run: |
-          uv run --script scripts/bm_bossbot_status.py pending \
-            --event "${{ steps.pr.outputs.event_file }}" \
-            --repo "${GITHUB_REPOSITORY}" \
-            --run-url "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}"
-
-      - name: Finalize BM Bossbot approval
-        if: always() && steps.pr.outputs.should_review == 'true'
-        env:
-          GITHUB_TOKEN: ${{ github.token }}
-        run: |
-          uv run --script scripts/bm_bossbot_status.py finalize \
-            --event "${{ steps.pr.outputs.event_file }}" \
-            --trusted "${{ steps.trust.outputs.trusted_author }}" \
-            --repo "${GITHUB_REPOSITORY}" \
-            --run-url "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}"
-
-  recheck:
-    name: BM Bossbot Thread Recheck
-    if: |
-      github.event_name == 'pull_request_review' ||
-      github.event_name == 'pull_request_review_comment' ||
-      github.event_name == 'pull_request_review_thread'
-    runs-on: ubuntu-latest
-    # Job-level concurrency: collapse bursts of thread events for one PR while
-    # staying isolated from the review job's group so neither cancels the other.
-    concurrency:
-      group: bm-bossbot-recheck-${{ github.event.pull_request.number }}
-      cancel-in-progress: true
-    steps:
-      - name: Checkout trusted base ref
-        uses: actions/checkout@v6
-        with:
-          ref: ${{ github.event.repository.default_branch }}
-          fetch-depth: 1
-
-      - name: Set up uv
-        uses: astral-sh/setup-uv@v3
-
-      - name: Re-evaluate approval from review-thread state
-        env:
-          GITHUB_TOKEN: ${{ github.token }}
-        run: |
-          uv run --script scripts/bm_bossbot_status.py recheck \
-            --pr-number "${{ github.event.pull_request.number }}" \
-            --repo "${GITHUB_REPOSITORY}" \
-            --run-url "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}"

.github/workflows/test.yml

@@ -13,15 +13,46 @@ on:
 
 env:
   FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
-  # Required CI records pytest-testmon data but still runs the full selected suite.
-  # The selective mode stays on explicit developer flows such as `just testmon`.
-  BASIC_MEMORY_TESTMON_FLAGS: "--testmon-noselect"
+  # Branch builds (PRs arrive as push events — this workflow has no
+  # pull_request trigger) select only impacted tests from the cached testmon
+  # baseline (branch cache falling back to main's full-run recording). Pushes
+  # to main run the full suite with --testmon-noselect to refresh the baseline.
+  BASIC_MEMORY_TESTMON_FLAGS: ${{ github.ref_name == 'main' && '--testmon-noselect' || '--testmon --testmon-forceselect' }}
 
 jobs:
+  changes:
+    # Docs/workflow-only changes skip the entire test matrix while the workflow
+    # still concludes successfully, so the BM Bossbot gate (workflow_run on
+    # Tests success) keeps firing and the PR stays mergeable.
+    name: Detect code changes
+    runs-on: ubuntu-latest
+    outputs:
+      code: ${{ steps.filter.outputs.code }}
+    steps:
+      - uses: actions/checkout@v6
+      - id: filter
+        uses: dorny/paths-filter@v3
+        with:
+          # Tests only runs on push events; for branch pushes compare against
+          # main (merge-base), for main pushes dorny diffs the push range.
+          base: main
+          filters: |
+            code:
+              - 'src/**'
+              - 'tests/**'
+              - 'test-int/**'
+              - 'alembic/**'
+              - 'pyproject.toml'
+              - 'uv.lock'
+              - 'justfile'
+              - '.github/workflows/test.yml'
+
   static-checks:
+    needs: changes
+    if: needs.changes.outputs.code == 'true'
     name: Static Checks (Python 3.12)
     timeout-minutes: 20
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
 
     steps:
       - uses: actions/checkout@v6
@@ -57,15 +88,17 @@ jobs:
           just lint
 
   test-sqlite-unit:
+    needs: changes
+    if: needs.changes.outputs.code == 'true'
     name: Test SQLite Unit (${{ matrix.os }}, Python ${{ matrix.python-version }})
     timeout-minutes: 45
     strategy:
       fail-fast: false
       matrix:
         include:
-          - os: depot-ubuntu-24.04
+          - os: ubuntu-latest
             python-version: "3.12"
-          - os: depot-ubuntu-24.04
+          - os: ubuntu-latest
             python-version: "3.13"
           # Python 3.14 unit tests are the longest full-suite slice; keep this
           # one on GitHub-hosted runners after Depot terminated it mid-suite.
@@ -118,17 +151,19 @@ jobs:
           just test-unit-sqlite
 
   test-sqlite-integration:
+    needs: changes
+    if: needs.changes.outputs.code == 'true'
     name: Test SQLite Integration (${{ matrix.os }}, Python ${{ matrix.python-version }})
     timeout-minutes: 45
     strategy:
       fail-fast: false
       matrix:
         include:
-          - os: depot-ubuntu-24.04
+          - os: ubuntu-latest
             python-version: "3.12"
-          - os: depot-ubuntu-24.04
+          - os: ubuntu-latest
             python-version: "3.13"
-          - os: depot-ubuntu-24.04
+          - os: ubuntu-latest
             python-version: "3.14"
           - os: windows-latest
             python-version: "3.12"
@@ -177,21 +212,20 @@ jobs:
           just test-int-sqlite
 
   test-postgres-unit:
-    name: Test Postgres Unit (Python ${{ matrix.python-version }})
+    needs: changes
+    if: needs.changes.outputs.code == 'true'
+    name: Test Postgres Unit (Python ${{ matrix.python-version }}, shard ${{ matrix.group }}/3)
     timeout-minutes: 60
     strategy:
       fail-fast: false
       matrix:
-        include:
-          - python-version: "3.12"
-            os: depot-ubuntu-24.04
-          - python-version: "3.13"
-            os: depot-ubuntu-24.04
-          # Match the SQLite unit slice: this full Python 3.14 path outlived
-          # the Depot runner and was terminated mid-suite.
-          - python-version: "3.14"
-            os: ubuntu-latest
-    runs-on: ${{ matrix.os }}
+        # Shard the largest suite across parallel jobs: each shard is a full job
+        # with its own Postgres service running 1/3 of the collection.
+        # Postgres runs on the latest Python only — the SQLite matrix carries
+        # Python-version coverage; Postgres carries backend coverage.
+        group: [1, 2, 3]
+        python-version: ["3.14"]
+    runs-on: ubuntu-latest
     services:
       postgres:
         image: pgvector/pgvector:pg16
@@ -233,9 +267,10 @@ jobs:
             .testmondata
             .testmondata-shm
             .testmondata-wal
-          key: ${{ runner.os }}-testmon-postgres-unit-py${{ matrix.python-version }}-${{ github.ref_name }}-${{ github.run_id }}
+          key: ${{ runner.os }}-testmon-postgres-unit-py${{ matrix.python-version }}-g${{ matrix.group }}-${{ github.ref_name }}-${{ github.run_id }}
           restore-keys: |
-            ${{ runner.os }}-testmon-postgres-unit-py${{ matrix.python-version }}-${{ github.ref_name }}-
+            ${{ runner.os }}-testmon-postgres-unit-py${{ matrix.python-version }}-g${{ matrix.group }}-${{ github.ref_name }}-
+            ${{ runner.os }}-testmon-postgres-unit-py${{ matrix.python-version }}-g${{ matrix.group }}-main-
             ${{ runner.os }}-testmon-postgres-unit-py${{ matrix.python-version }}-main-
             ${{ runner.os }}-testmon-postgres-unit-py${{ matrix.python-version }}-
 
@@ -249,19 +284,20 @@ jobs:
 
       - name: Run tests
         run: |
-          just test-unit-postgres
+          BASIC_MEMORY_PYTEST_SPLIT_FLAGS="--splits 3 --group ${{ matrix.group }}" just test-unit-postgres
 
   test-postgres-integration:
+    needs: changes
+    if: needs.changes.outputs.code == 'true'
     name: Test Postgres Integration (Python ${{ matrix.python-version }})
     timeout-minutes: 45
     strategy:
       fail-fast: false
       matrix:
-        include:
-          - python-version: "3.12"
-          - python-version: "3.13"
-          - python-version: "3.14"
-    runs-on: depot-ubuntu-24.04
+        # Latest Python only: SQLite carries version coverage, Postgres carries
+        # backend coverage.
+        python-version: ["3.14"]
+    runs-on: ubuntu-latest
     services:
       postgres:
         image: pgvector/pgvector:pg16
@@ -322,9 +358,11 @@ jobs:
           just test-int-postgres
 
   test-semantic:
+    needs: changes
+    if: needs.changes.outputs.code == 'true'
     name: Test Semantic (Python 3.12)
     timeout-minutes: 45
-    runs-on: depot-ubuntu-24.04
+    runs-on: ubuntu-latest
 
     steps:
       - uses: actions/checkout@v6

CHANGELOG.md

@@ -1,5 +1,70 @@
 # CHANGELOG
 
+## v0.22.0 (2026-06-11)
+
+Team-safe cloud sync. New additive `bm cloud push` and `bm cloud pull`
+commands work safely on shared Team workspaces, while the destructive mirror
+commands are gated to Personal workspaces. Also: a large batch of MCP tool
+fixes, search improvements, and embedding reliability work.
+
+### Features
+
+- **#917**: Added Team-safe `bm cloud push` / `bm cloud pull`. Both are
+  additive (they never delete on the destination) and abort on conflicts by
+  default, git-style, with `--on-conflict {fail|keep-local|keep-cloud|keep-both}`.
+  The destructive `bm cloud sync` / `bm cloud bisync` mirrors are now gated
+  to Personal workspaces.
+- **#920**: Team push/pull uses per-workspace rclone remotes, so remotes and
+  credentials stay scoped to each workspace.
+- **#908**: Search supports an observation category filter.
+- **#809**: Added an experimental LiteLLM embedding provider for semantic
+  search (marked experimental, see **#899**).
+- **#907**: `bm tool write-note` accepts `--type`.
+- **#906**: `bm status` accepts `--wait` and `--timeout`.
+- Added the `bm tool delete-note` command.
+- **#905**: Improved workspace and cloud bisync command discoverability.
+
+### Bug Fixes
+
+- **#931 / #946**: Truncated observation permalinks are disambiguated to
+  prevent search-index collisions, and `build_context` resolves observations
+  by the same permalink the search index uses (**#909**, **#929**).
+- **#934**: `edit_note` recovers when the file exists on disk but is not yet
+  indexed (**#581**).
+- **#911 / #932 / #941**: Comma-separated tags are split consistently in
+  `parse_tags` and the `search_notes` tags parameter, with input normalized
+  for direct callers (**#910**).
+- **#933**: `read_note` accepts `page`/`page_size` for parity with sibling
+  tools (**#883**).
+- **#914 / #904 / #916**: `move_note` resolves `memory://` URLs, stops
+  falsely rejecting same-project moves as cross-project, no longer reports
+  false success across project boundaries, and mismatch guidance points at
+  the landing path.
+- **#915**: Navigation pagination is validated, and `recent_activity` shows
+  the correct project.
+- **#913**: `bm tool` commands align with MCP behavior (error exit codes,
+  overwrite handling, category support, defaults).
+- **#923**: `bm cloud setup` no longer overwrites an existing rclone remote
+  (**#922**).
+- **#912**: The `note_types` search filter is case-insensitive.
+- `build_context` allows cross-project context traversal, `write_note`
+  resolves overwrite conflicts, and sync uses strict deferred relation
+  resolution.
+- Embedding reliability: FastEmbed vectors are L2-normalized (**#843**),
+  corrupt FastEmbed model caches self-heal (**#900**), a single embedding
+  provider is reused per process (**#903**), `sqlite-vec` loads for the
+  embedding-status query (**#901**), and engine disposal no longer crashes
+  on the Postgres backend (**#902**).
+
+### Maintenance
+
+- Leaner, faster CI: testmon-selected branch builds, sharded Postgres jobs,
+  and faster default test fixtures (**#928**, **#938**, **#945**).
+- Documented personal-vs-team cloud sync semantics (**#947**, closes
+  **#851**).
+- Fixed npx skill install docs (**#927**).
+- Added `glama.json` to claim the Glama MCP directory listing (**#953**).
+
 ## v0.21.6 (2026-06-04)
 
 Monorepo consolidation plus a redesigned Claude Code plugin. The satellite

docs/cloud-cli.md

@@ -318,7 +318,7 @@ For MCP stdio, routing is always local.
 | `bm cloud push` | local → cloud | Personal + Team | Upload local changes, additively (git-style) |
 | `bm cloud sync` | local → cloud | Personal only | One-way mirror (cloud becomes identical to local) |
 | `bm cloud bisync` | local ↔ cloud | Personal only | Two-way mirror (recommended for solo use) |
-| `bm cloud check` | — | Personal + Team | Verify files match (no changes) |
+| `bm cloud check` | — | Personal only | Verify mirror integrity (no changes) |
 
 If you collaborate on a shared Team workspace, use **`push`/`pull`** (see [Team Workspaces](#team-workspaces-push--pull-additive-git-style)). If you are the only writer (a Personal workspace), the mirror commands `sync`/`bisync` give you a single source of truth.
 
@@ -459,10 +459,12 @@ bm cloud bisync --name research
 - You edit in multiple places
 - You want automatic conflict resolution
 
-### Verify Sync Integrity
+### Verify Sync Integrity (Personal only)
 
 **Use case:** Check if local and cloud match without making changes.
 
+> **Personal workspaces only.** `check` compares against the Personal workspace mirror remote, like `sync`/`bisync`. On Team workspaces use `bm cloud pull --dry-run` / `bm cloud push --dry-run` to preview differences instead.
+
 ```bash
 bm cloud check --name research
 ```
@@ -999,7 +1001,7 @@ bm cloud bisync --name <project> --resync # First time / force baseline
 bm cloud bisync --name <project> --dry-run
 bm cloud bisync --name <project> --verbose
 
-# Integrity check
+# Integrity check - Personal workspaces only
 bm cloud check --name <project>
 bm cloud check --name <project> --one-way
 

glama.json

@@ -0,0 +1,7 @@
+{
+  "$schema": "https://glama.ai/mcp/schemas/server.json",
+  "maintainers": [
+    "phernandez",
+    "groksrc"
+  ]
+}

integrations/hermes/__init__.py

@@ -38,7 +38,7 @@ from typing import Any, Callable
 from agent.memory_provider import MemoryProvider
 from tools.registry import tool_error
 
-__version__ = "0.21.6"
+__version__ = "0.22.0"
 
 logger = logging.getLogger("hermes.memory.basic-memory")
 

integrations/hermes/plugin.yaml

@@ -1,5 +1,5 @@
 name: basic-memory
-version: 0.21.6
+version: 0.22.0
 description: "Basic Memory — persistent knowledge graph backed by the basic-memory MCP server"
 pip_dependencies:
   - mcp

integrations/openclaw/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@basicmemory/openclaw-basic-memory",
-  "version": "0.21.6",
+  "version": "0.22.0",
   "type": "module",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",

justfile

@@ -3,6 +3,9 @@
 TESTMON_FLAGS := env_var_or_default("BASIC_MEMORY_TESTMON_FLAGS", "--testmon-noselect")
 TESTMON_SELECT_FLAGS := env_var_or_default("BASIC_MEMORY_TESTMON_SELECT_FLAGS", "--testmon --testmon-forceselect")
 TESTMON_REFRESH_FLAGS := env_var_or_default("BASIC_MEMORY_TESTMON_REFRESH_FLAGS", "--testmon-noselect")
+# CI shards the Postgres unit suite across parallel jobs via pytest-split
+# (e.g. "--splits 3 --group 2"). Empty locally.
+PYTEST_SPLIT_FLAGS := env_var_or_default("BASIC_MEMORY_PYTEST_SPLIT_FLAGS", "")
 
 # Install dependencies
 install:
@@ -43,8 +46,12 @@ test-unit-sqlite: testmon-seed
     BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov {{TESTMON_FLAGS}} --testmon-env=unit-sqlite tests
 
 # Run unit tests against Postgres
+# Exit code 5 (no tests collected) is success: a testmon-selected PR build can
+# leave a pytest-split shard empty.
 test-unit-postgres: testmon-seed
-    BASIC_MEMORY_ENV=test BASIC_MEMORY_TEST_POSTGRES=1 uv run pytest -p pytest_mock -v --no-cov {{TESTMON_FLAGS}} --testmon-env=unit-postgres tests
+    #!/usr/bin/env bash
+    set -euo pipefail
+    BASIC_MEMORY_ENV=test BASIC_MEMORY_TEST_POSTGRES=1 uv run pytest -p pytest_mock -v --no-cov {{TESTMON_FLAGS}} {{PYTEST_SPLIT_FLAGS}} --testmon-env=unit-postgres tests || test $? -eq 5
 
 # Run integration tests against SQLite (excludes semantic tests and on-demand benchmarks —
 # use just test-semantic / run benchmark files explicitly)

plugins/claude-code/.claude-plugin/marketplace.json

@@ -6,14 +6,14 @@
   },
   "metadata": {
     "description": "Official plugins from Basic Machines for knowledge management and AI-assisted development",
-    "version": "0.21.6"
+    "version": "0.22.0"
   },
   "plugins": [
     {
       "name": "basic-memory",
       "source": "./",
       "description": "The bridge between Claude's working memory and Basic Memory's durable knowledge graph \u2014 session briefings, pre-compaction checkpoints, and capture reflexes",
-      "version": "0.21.6",
+      "version": "0.22.0",
       "author": {
         "name": "Basic Machines"
       },

plugins/claude-code/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "basic-memory",
   "description": "The bridge between Claude's working memory and Basic Memory's durable knowledge graph \u2014 session briefings, pre-compaction checkpoints, and capture reflexes",
-  "version": "0.21.6",
+  "version": "0.22.0",
   "author": {
     "name": "Basic Machines"
   },

plugins/codex/.codex-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "codex",
-  "version": "0.21.6",
+  "version": "0.22.0",
   "description": "A Codex-native bridge to Basic Memory for durable engineering context, decisions, and resumable checkpoints.",
   "author": {
     "name": "Basic Machines",

pyproject.toml

@@ -120,6 +120,7 @@ dev = [
     "cst-lsp>=0.1.3",
     "libcst>=1.8.6",
     "pytest-timeout>=2.4.0",
+    "pytest-split>=0.11.0",
 ]
 
 [tool.hatch.version]

scripts/bm_bossbot_status.py

@@ -1,534 +0,0 @@
-#!/usr/bin/env -S uv run --script
-# /// script
-# requires-python = ">=3.12"
-# dependencies = [
-#   "typer>=0.9.0",
-# ]
-# ///
-"""BM Bossbot status and PR-body helpers.
-
-BM Bossbot is a deterministic merge gate — no LLM review. It approves a head
-SHA only when the Tests workflow succeeded for it (enforced by the workflow
-trigger), the PR is not a draft, the author is trusted, and every review
-thread is resolved. Code review itself comes from the Codex connector and
-human reviewers; this gate just refuses to let unaddressed feedback merge.
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import re
-import urllib.error
-import urllib.request
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Annotated, Any, Mapping
-
-import typer
-
-
-STATUS_CONTEXT = "BM Bossbot Approval"
-SUMMARY_START = "<!-- BM_BOSSBOT_SUMMARY:start -->"
-SUMMARY_END = "<!-- BM_BOSSBOT_SUMMARY:end -->"
-APPROVED_DESCRIPTION = "BM Bossbot approved this head SHA"
-PENDING_DESCRIPTION = "BM Bossbot is reviewing this head SHA"
-app = typer.Typer(
-    add_completion=False,
-    help="Manage deterministic BM Bossbot PR approval statuses.",
-    no_args_is_help=True,
-)
-
-
-@dataclass(frozen=True)
-class ApprovalResult:
-    approved: bool
-    state: str
-    description: str
-
-
-@dataclass(frozen=True)
-class PullRequestEvent:
-    repo: str
-    number: int
-    head_sha: str
-    body: str
-
-
-def read_json(path: Path) -> Any:
-    try:
-        return json.loads(path.read_text(encoding="utf-8"))
-    except FileNotFoundError:
-        raise SystemExit(f"Missing JSON file: {path}") from None
-    except json.JSONDecodeError as exc:
-        raise SystemExit(f"{path}: invalid JSON: {exc}") from None
-
-
-def pull_request_event(
-    payload: Mapping[str, Any], repo_override: str | None = None
-) -> PullRequestEvent:
-    pr = payload.get("pull_request")
-    if not isinstance(pr, Mapping):
-        raise SystemExit("GitHub event payload is missing pull_request")
-
-    repo = repo_override
-    if repo is None:
-        repository = payload.get("repository")
-        if isinstance(repository, Mapping):
-            repo = _string(repository.get("full_name"))
-    if not repo:
-        raise SystemExit("Could not determine GitHub repository")
-
-    number = pr.get("number")
-    if not isinstance(number, int):
-        raise SystemExit("GitHub event payload is missing pull_request.number")
-
-    head = pr.get("head")
-    head_sha = (
-        _string(head.get("sha")) if isinstance(head, Mapping) else _string(pr.get("head_sha"))
-    )
-    if not head_sha:
-        raise SystemExit("GitHub event payload is missing pull_request.head.sha")
-
-    return PullRequestEvent(
-        repo=repo,
-        number=number,
-        head_sha=head_sha,
-        body=_string(pr.get("body")),
-    )
-
-
-def count_unresolved_review_threads(*, token: str, repo: str, number: int) -> int:
-    """Count unresolved review threads (e.g. open Codex inline comments) on a PR.
-
-    Review threads are the canonical 'outstanding feedback' signal: bot reviewers
-    submit COMMENTED reviews that never flip reviewDecision, so thread resolution
-    is the only deterministic way to know feedback was addressed.
-    """
-    owner, _, name = repo.partition("/")
-    if not owner or not name:
-        raise SystemExit(f"Invalid repository: {repo}")
-
-    query = """
-    query($owner: String!, $name: String!, $number: Int!, $cursor: String) {
-      repository(owner: $owner, name: $name) {
-        pullRequest(number: $number) {
-          reviewThreads(first: 100, after: $cursor) {
-            pageInfo { hasNextPage endCursor }
-            nodes { isResolved }
-          }
-        }
-      }
-    }
-    """
-
-    unresolved = 0
-    cursor: str | None = None
-    while True:
-        response = _github_request(
-            method="POST",
-            path="/graphql",
-            token=token,
-            payload={
-                "query": query,
-                "variables": {"owner": owner, "name": name, "number": number, "cursor": cursor},
-            },
-        )
-        if not isinstance(response, Mapping) or response.get("errors"):
-            raise SystemExit(f"GitHub GraphQL reviewThreads query failed: {response}")
-        try:
-            threads = response["data"]["repository"]["pullRequest"]["reviewThreads"]
-            nodes = threads["nodes"]
-            page_info = threads["pageInfo"]
-        except (KeyError, TypeError):
-            raise SystemExit(
-                "GitHub GraphQL reviewThreads response was missing expected fields"
-            ) from None
-        unresolved += sum(1 for node in nodes if not node.get("isResolved"))
-        if not page_info.get("hasNextPage"):
-            return unresolved
-        cursor = page_info.get("endCursor")
-
-
-def unresolved_threads_result(count: int) -> ApprovalResult:
-    return ApprovalResult(
-        False,
-        "failure",
-        f"BM Bossbot found {count} unresolved review thread(s)",
-    )
-
-
-def evaluate_gate(
-    *,
-    token: str,
-    repo: str,
-    number: int,
-    trusted: bool,
-) -> tuple[ApprovalResult, int]:
-    """Deterministic approval decision: trusted author + zero unresolved threads.
-
-    Tests-passed-for-this-head and non-draft are enforced upstream by the
-    workflow trigger and the normalize step (should_review). Returns the
-    result plus the unresolved-thread count for the PR-body summary.
-    """
-    if not trusted:
-        return (
-            ApprovalResult(
-                False,
-                "failure",
-                "BM Bossbot only gates owner/member/collaborator PRs",
-            ),
-            0,
-        )
-    unresolved = count_unresolved_review_threads(token=token, repo=repo, number=number)
-    if unresolved > 0:
-        return unresolved_threads_result(unresolved), unresolved
-    return ApprovalResult(True, "success", APPROVED_DESCRIPTION), 0
-
-
-def build_status_payload(*, state: str, description: str, target_url: str) -> dict[str, str]:
-    return {
-        "state": state,
-        "context": STATUS_CONTEXT,
-        "description": description,
-        "target_url": target_url,
-    }
-
-
-def render_summary(
-    *,
-    head_sha: str,
-    result: ApprovalResult,
-    trusted: bool,
-    unresolved_threads: int,
-) -> str:
-    return "\n".join(
-        [
-            f"Reviewed SHA: `{head_sha}`",
-            "Gate: deterministic (tests, draft, author trust, review threads)",
-            f"Status: `{result.state}` - {result.description}",
-            "",
-            f"- Trusted author: {'yes' if trusted else 'no'}",
-            f"- Unresolved review threads: {unresolved_threads}",
-            "",
-            "Code review comes from the Codex connector and human reviewers;",
-            "resolve every review thread to (re)gain approval for this head SHA.",
-        ]
-    )
-
-
-def upsert_summary_block(body: str, summary: str) -> str:
-    block = f"{SUMMARY_START}\n{summary.rstrip()}\n{SUMMARY_END}"
-    pattern = re.compile(
-        rf"{re.escape(SUMMARY_START)}.*?{re.escape(SUMMARY_END)}",
-        flags=re.DOTALL,
-    )
-    if pattern.search(body):
-        return pattern.sub(block, body, count=1)
-    if body.strip():
-        return f"{body.rstrip()}\n\n{block}\n"
-    return f"{block}\n"
-
-
-def set_commit_status(*, token: str, repo: str, sha: str, payload: Mapping[str, str]) -> None:
-    _github_request(
-        method="POST",
-        path=f"/repos/{repo}/statuses/{sha}",
-        token=token,
-        payload=payload,
-    )
-
-
-def update_pull_request_body(*, token: str, repo: str, number: int, body: str) -> None:
-    _github_request(
-        method="PATCH",
-        path=f"/repos/{repo}/pulls/{number}",
-        token=token,
-        payload={"body": body},
-    )
-
-
-def get_pull_request_body(*, token: str, repo: str, number: int) -> str:
-    response = _github_request(
-        method="GET",
-        path=f"/repos/{repo}/pulls/{number}",
-        token=token,
-    )
-    if not isinstance(response, Mapping):
-        raise SystemExit("GitHub API response for pull request was invalid")
-    return _string(response.get("body"))
-
-
-def mark_pending(
-    *,
-    event_path: Path,
-    repo: str | None,
-    run_url: str,
-    token_env: str,
-) -> None:
-    event = pull_request_event(read_json(event_path), repo_override=repo)
-    set_commit_status(
-        token=_token(token_env),
-        repo=event.repo,
-        sha=event.head_sha,
-        payload=build_status_payload(
-            state="pending",
-            description=PENDING_DESCRIPTION,
-            target_url=run_url,
-        ),
-    )
-    typer.echo(f"Marked {STATUS_CONTEXT} pending for {event.head_sha}")
-
-
-def finalize_review(
-    *,
-    event_path: Path,
-    trusted: bool,
-    repo: str | None,
-    run_url: str,
-    token_env: str,
-) -> ApprovalResult:
-    event = pull_request_event(read_json(event_path), repo_override=repo)
-    token = _token(token_env)
-
-    result, unresolved = evaluate_gate(
-        token=token, repo=event.repo, number=event.number, trusted=trusted
-    )
-    current_body = get_pull_request_body(token=token, repo=event.repo, number=event.number)
-    updated_body = upsert_summary_block(
-        current_body,
-        render_summary(
-            head_sha=event.head_sha,
-            result=result,
-            trusted=trusted,
-            unresolved_threads=unresolved,
-        ),
-    )
-    update_pull_request_body(token=token, repo=event.repo, number=event.number, body=updated_body)
-    set_commit_status(
-        token=token,
-        repo=event.repo,
-        sha=event.head_sha,
-        payload=build_status_payload(
-            state=result.state,
-            description=result.description,
-            target_url=run_url,
-        ),
-    )
-    typer.echo(f"Marked {STATUS_CONTEXT} {result.state} for {event.head_sha}")
-    return result
-
-
-def get_pull_request_head_sha(*, token: str, repo: str, number: int) -> str:
-    response = _github_request(
-        method="GET",
-        path=f"/repos/{repo}/pulls/{number}",
-        token=token,
-    )
-    if not isinstance(response, Mapping):
-        raise SystemExit("GitHub API response for pull request was invalid")
-    head = response.get("head")
-    head_sha = _string(head.get("sha")) if isinstance(head, Mapping) else ""
-    if not head_sha:
-        raise SystemExit("GitHub API response was missing pull request head SHA")
-    return head_sha
-
-
-def head_sha_was_approved(*, token: str, repo: str, sha: str) -> bool:
-    """Return whether a full BM Bossbot review previously approved this head SHA.
-
-    Commit statuses are append-only history, so the approval record survives a
-    later thread-failure status for the same SHA. The recheck path can post a
-    new status on every review-thread event, so a busy PR can accumulate more
-    than one page of statuses — page through all of them or the approval
-    record falls off page one and a valid approval is never restored.
-    """
-    page = 1
-    while True:
-        response = _github_request(
-            method="GET",
-            path=f"/repos/{repo}/commits/{sha}/statuses?per_page=100&page={page}",
-            token=token,
-        )
-        if not isinstance(response, list):
-            raise SystemExit("GitHub API response for commit statuses was invalid")
-        if not response:
-            return False
-        if any(
-            isinstance(status, Mapping)
-            and status.get("context") == STATUS_CONTEXT
-            and status.get("state") == "success"
-            and status.get("description") == APPROVED_DESCRIPTION
-            for status in response
-        ):
-            return True
-        page += 1
-
-
-def recheck_threads(
-    *,
-    repo: str,
-    number: int,
-    run_url: str,
-    token_env: str,
-) -> None:
-    """Re-evaluate the approval status when review threads change.
-
-    Trigger: pull_request_review / review_comment / review_thread events.
-    Why: the full review runs once per head SHA after Tests; feedback that
-         arrives later (or gets resolved later) must move the gate without
-         re-running the LLM review.
-    Outcome: unresolved threads flip the status to failure; once every thread
-         is resolved, a previously earned approval for the same head SHA is
-         restored. Without a prior approval the status is left untouched so a
-         pending/failed review cannot be upgraded by thread resolution alone.
-    """
-    token = _token(token_env)
-    head_sha = get_pull_request_head_sha(token=token, repo=repo, number=number)
-    unresolved = count_unresolved_review_threads(token=token, repo=repo, number=number)
-
-    if unresolved > 0:
-        result = unresolved_threads_result(unresolved)
-    elif head_sha_was_approved(token=token, repo=repo, sha=head_sha):
-        result = ApprovalResult(True, "success", APPROVED_DESCRIPTION)
-    else:
-        typer.echo(
-            f"All review threads resolved but no prior approval exists for {head_sha}; "
-            "leaving status unchanged"
-        )
-        return
-
-    set_commit_status(
-        token=token,
-        repo=repo,
-        sha=head_sha,
-        payload=build_status_payload(
-            state=result.state,
-            description=result.description,
-            target_url=run_url,
-        ),
-    )
-    typer.echo(f"Marked {STATUS_CONTEXT} {result.state} for {head_sha} ({result.description})")
-
-
-def _github_request(
-    *,
-    method: str,
-    path: str,
-    token: str,
-    payload: Mapping[str, Any] | None = None,
-) -> Any:
-    data = None if payload is None else json.dumps(payload).encode("utf-8")
-    request = urllib.request.Request(
-        f"https://api.github.com{path}",
-        data=data,
-        method=method,
-        headers={
-            "Accept": "application/vnd.github+json",
-            "Authorization": f"Bearer {token}",
-            "Content-Type": "application/json",
-            "User-Agent": "basic-memory-bm-bossbot",
-            "X-GitHub-Api-Version": "2022-11-28",
-        },
-    )
-    try:
-        with urllib.request.urlopen(request, timeout=30) as response:
-            response_body = response.read().decode("utf-8")
-    except urllib.error.HTTPError as exc:
-        detail = exc.read().decode("utf-8", errors="replace")
-        raise SystemExit(f"GitHub API request failed: {exc.code} {detail}") from None
-    return json.loads(response_body) if response_body else None
-
-
-def _string(value: object) -> str:
-    return value if isinstance(value, str) else ""
-
-
-def _token(env_name: str) -> str:
-    token = os.environ.get(env_name)
-    if not token:
-        raise SystemExit(f"Missing required token environment variable: {env_name}")
-    return token
-
-
-@app.command("pending")
-def pending(
-    event: Annotated[
-        Path,
-        typer.Option(
-            "--event",
-            exists=True,
-            dir_okay=False,
-            readable=True,
-            help="GitHub event payload JSON.",
-        ),
-    ],
-    run_url: Annotated[str, typer.Option("--run-url", help="Workflow run URL.")],
-    repo: Annotated[str | None, typer.Option("--repo", help="owner/name repository.")] = None,
-    token_env: Annotated[
-        str,
-        typer.Option("--token-env", help="Environment variable containing a GitHub token."),
-    ] = "GITHUB_TOKEN",
-) -> None:
-    """Set BM Bossbot Approval pending on the PR head SHA."""
-    mark_pending(event_path=event, repo=repo, run_url=run_url, token_env=token_env)
-
-
-@app.command("finalize")
-def finalize(
-    event: Annotated[
-        Path,
-        typer.Option(
-            "--event",
-            exists=True,
-            dir_okay=False,
-            readable=True,
-            help="GitHub event payload JSON.",
-        ),
-    ],
-    trusted: Annotated[
-        str,
-        typer.Option(
-            "--trusted",
-            help="Whether the PR author is trusted (true/false from the classify step).",
-        ),
-    ],
-    run_url: Annotated[str, typer.Option("--run-url", help="Workflow run URL.")],
-    repo: Annotated[str | None, typer.Option("--repo", help="owner/name repository.")] = None,
-    token_env: Annotated[
-        str,
-        typer.Option("--token-env", help="Environment variable containing a GitHub token."),
-    ] = "GITHUB_TOKEN",
-) -> None:
-    """Finalize BM Bossbot Approval from the deterministic gate."""
-    result = finalize_review(
-        event_path=event,
-        trusted=trusted.strip().lower() == "true",
-        repo=repo,
-        run_url=run_url,
-        token_env=token_env,
-    )
-    if not result.approved:
-        raise typer.Exit(1)
-
-
-@app.command("recheck")
-def recheck(
-    pr_number: Annotated[int, typer.Option("--pr-number", min=1, help="Pull request number.")],
-    run_url: Annotated[str, typer.Option("--run-url", help="Workflow run URL.")],
-    repo: Annotated[str, typer.Option("--repo", help="owner/name repository.")],
-    token_env: Annotated[
-        str,
-        typer.Option("--token-env", help="Environment variable containing a GitHub token."),
-    ] = "GITHUB_TOKEN",
-) -> None:
-    """Re-evaluate BM Bossbot Approval from current review-thread state."""
-    recheck_threads(repo=repo, number=pr_number, run_url=run_url, token_env=token_env)
-
-
-def main() -> None:
-    app()
-
-
-if __name__ == "__main__":
-    main()

scripts/generate_infographic.py

@@ -1,166 +0,0 @@
-#!/usr/bin/env -S uv run --script
-# /// script
-# requires-python = ">=3.12"
-# dependencies = [
-#   "openai>=1.100.2",
-#   "python-dotenv>=1.1.0",
-#   "typer>=0.9.0",
-# ]
-# ///
-"""Generate a BM Bossbot infographic with the OpenAI Images API."""
-
-from __future__ import annotations
-
-import base64
-import time
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Annotated, Any
-
-import typer
-from dotenv import load_dotenv
-from openai import OpenAI
-
-
-DEFAULT_MODEL = "gpt-image-2"
-DEFAULT_SIZE = "1536x1024"
-DEFAULT_QUALITY = "high"
-DEFAULT_FORMAT = "webp"
-DEFAULT_COMPRESSION = 90
-app = typer.Typer(
-    add_completion=False,
-    help="Generate Basic Memory infographics with the OpenAI Images API.",
-    no_args_is_help=True,
-)
-
-
-@dataclass(frozen=True)
-class GeneratedImage:
-    path: Path
-    revised_prompt: str | None
-
-
-def validate_output_path(path: Path, *, repo_root: Path | None = None) -> Path:
-    root = (repo_root or Path.cwd()).resolve()
-    output = path.resolve()
-    allowed_root = (root / "docs" / "assets" / "infographics").resolve()
-    if not output.is_relative_to(allowed_root):
-        allowed_path = allowed_root.relative_to(root).as_posix()
-        raise ValueError(f"Output path must be under {allowed_path}")
-    if output.suffix != ".webp":
-        raise ValueError("Output path must end with .webp")
-    return output
-
-
-def generate_image_result(
-    *,
-    prompt: str,
-    output_path: Path,
-    model: str = DEFAULT_MODEL,
-    size: str = DEFAULT_SIZE,
-    quality: str = DEFAULT_QUALITY,
-    output_format: str = DEFAULT_FORMAT,
-    output_compression: int = DEFAULT_COMPRESSION,
-    client: Any | None = None,
-    retries: int = 2,
-) -> GeneratedImage:
-    output = validate_output_path(output_path)
-    output.parent.mkdir(parents=True, exist_ok=True)
-    load_dotenv()
-    openai_client = client or OpenAI()
-
-    for attempt in range(retries + 1):
-        try:
-            response = openai_client.images.generate(
-                model=model,
-                prompt=prompt,
-                size=size,
-                quality=quality,
-                output_format=output_format,
-                output_compression=output_compression,
-            )
-            image = response.data[0]
-            image_b64 = image.b64_json
-            if not image_b64:
-                raise RuntimeError("OpenAI image response did not include b64_json")
-            output.write_bytes(base64.b64decode(image_b64))
-            return GeneratedImage(path=output, revised_prompt=image.revised_prompt)
-        except Exception:
-            if attempt >= retries:
-                raise
-            time.sleep(2**attempt)
-
-    raise RuntimeError("Image generation retry loop exited unexpectedly")
-
-
-def generate_image(
-    *,
-    prompt: str,
-    output_path: Path,
-    model: str = DEFAULT_MODEL,
-    size: str = DEFAULT_SIZE,
-    quality: str = DEFAULT_QUALITY,
-    output_format: str = DEFAULT_FORMAT,
-    output_compression: int = DEFAULT_COMPRESSION,
-    client: Any | None = None,
-    retries: int = 2,
-) -> Path:
-    return generate_image_result(
-        prompt=prompt,
-        output_path=output_path,
-        model=model,
-        size=size,
-        quality=quality,
-        output_format=output_format,
-        output_compression=output_compression,
-        client=client,
-        retries=retries,
-    ).path
-
-
-@app.command()
-def generate(
-    prompt_file: Annotated[
-        Path,
-        typer.Option(
-            "--prompt-file",
-            exists=True,
-            dir_okay=False,
-            readable=True,
-            help="Markdown/text prompt file to send to the image model.",
-        ),
-    ],
-    output: Annotated[Path, typer.Option("--output", help="Output .webp path.")],
-    model: Annotated[str, typer.Option("--model", help="OpenAI image model.")] = DEFAULT_MODEL,
-    size: Annotated[str, typer.Option("--size", help="Image size.")] = DEFAULT_SIZE,
-    quality: Annotated[str, typer.Option("--quality", help="Image quality.")] = DEFAULT_QUALITY,
-    output_compression: Annotated[
-        int,
-        typer.Option(
-            "--output-compression",
-            min=0,
-            max=100,
-            help="WebP output compression.",
-        ),
-    ] = DEFAULT_COMPRESSION,
-    retries: Annotated[int, typer.Option("--retries", min=0, help="Retry attempts.")] = 2,
-) -> None:
-    """Generate an infographic from a prompt file."""
-    output = generate_image(
-        prompt=prompt_file.read_text(encoding="utf-8"),
-        output_path=output,
-        model=model,
-        size=size,
-        quality=quality,
-        output_compression=output_compression,
-        retries=retries,
-    )
-    typer.echo(output)
-
-
-def main() -> None:
-    app()
-
-
-if __name__ == "__main__":
-    main()

scripts/generate_pr_infographic.py

@@ -1,438 +0,0 @@
-#!/usr/bin/env -S uv run --script
-# /// script
-# requires-python = ">=3.12"
-# dependencies = [
-#   "openai>=1.100.2",
-#   "python-dotenv>=1.1.0",
-#   "typer>=0.9.0",
-# ]
-# ///
-"""Build and generate a non-gating BM Bossbot PR image."""
-
-from __future__ import annotations
-
-import hashlib
-import html
-import json
-import re
-from dataclasses import dataclass
-from enum import StrEnum
-from pathlib import Path
-from typing import Annotated, Any, Mapping
-
-import typer
-
-if __package__:
-    from .generate_infographic import (
-        DEFAULT_MODEL,
-        DEFAULT_QUALITY,
-        DEFAULT_SIZE,
-        generate_image_result,
-    )
-else:
-    from generate_infographic import (
-        DEFAULT_MODEL,
-        DEFAULT_QUALITY,
-        DEFAULT_SIZE,
-        generate_image_result,
-    )
-
-
-SUMMARY_START = "<!-- BM_BOSSBOT_SUMMARY:start -->"
-SUMMARY_END = "<!-- BM_BOSSBOT_SUMMARY:end -->"
-THEME_START = "<!-- BM_INFOGRAPHIC_THEME:start -->"
-THEME_END = "<!-- BM_INFOGRAPHIC_THEME:end -->"
-PROVENANCE_START = "<!-- BM_INFOGRAPHIC_PROVENANCE:start -->"
-PROVENANCE_END = "<!-- BM_INFOGRAPHIC_PROVENANCE:end -->"
-IMAGE_START = "<!-- pr-infographic:start -->"
-IMAGE_END = "<!-- pr-infographic:end -->"
-# Managed blocks are bot-written artifacts (review verdict, image embed,
-# provenance). They must never feed the image: sourcing the review summary is
-# what made every image an "APPROVED" stamp instead of depicting the change.
-MANAGED_BLOCKS = (
-    (SUMMARY_START, SUMMARY_END),
-    (THEME_START, THEME_END),
-    (PROVENANCE_START, PROVENANCE_END),
-    (IMAGE_START, IMAGE_END),
-)
-app = typer.Typer(
-    add_completion=False,
-    help="Generate a non-gating BM Bossbot PR image.",
-    no_args_is_help=True,
-)
-
-
-class ThemeSource(StrEnum):
-    AUTO = "auto"
-    CLI = "cli"
-    PR_BODY = "pr-body"
-
-
-@dataclass(frozen=True)
-class ThemeSelection:
-    theme: str
-    source: ThemeSource
-
-
-BM_IMAGE_THEME_POOL = (
-    "computer science college textbook: SICP-style diagrams, automata, compiler "
-    "pipelines, type theory, and annotated chalkboard rigor",
-    "classic literature: sea voyages, gothic manors, Dickensian streets, library "
-    "marginalia, and travel-journal artifacts",
-    "fantasy quest ledger: original guild maps, spellbooks, dungeon keys, tavern "
-    "notices, and artifact inventories with no copyrighted settings",
-    "heavy music editorial: metal, hard rock, punk, techno, soul, or reggae "
-    "tour-poster energy with no direct band logos or likenesses",
-    "knockoff space opera: fleet routes, mission consoles, contraband manifests, "
-    "and practical starship drama with no named fictional universes",
-    "sword-and-sorcery: ruined temples, desert roads, battle standards, ancient "
-    "maps, and heroic silhouettes with no named character likenesses",
-    "comic book cover: original splash-page composition, caption boxes, clean "
-    "halftone texture, and bold issue-cover drama",
-    "French new wave movie poster: stark typography, city streets, jump-cut "
-    "composition, and high-contrast editorial photography cues",
-    "WWII public-information poster: home-front logistics, mobilization arrows, "
-    "bold simplified figures, and no real-world party symbols or hate imagery",
-    "Italian movie poster: hand-painted drama, expressive color, credit-block "
-    "energy, and 1960s or 1970s cinema composition with no actor likenesses",
-    "Shakespearean stage: acts and scenes, court intrigue, stage blocking, "
-    "dramatis personae, backstage cue sheets, and theatrical light",
-    "Greek mythology: temple steps, oracle tablets, constellations, labyrinths, "
-    "ship routes, and original heroic allegory",
-    "noir detective photography: case files, typed evidence labels, civic "
-    "infrastructure, streetlight shadows, and newsroom archive grit",
-    "space exploration and astronomy: celestial atlases, observatory charts, "
-    "orbital mechanics, planetary survey routes, and deep-space mission drama",
-    "editorial painting: abstract, classical landscape, western action, "
-    "chiaroscuro, historical mural, stormy seascape, or allegorical canvas",
-    "classic black-and-white photography: documentary field report, contact "
-    "sheet, street photography, civic infrastructure, and darkroom contrast",
-    "80's action movie poster: smoky backlit warehouses, neon streets, practical "
-    "explosions, mission dossiers, countdowns, and no actor likenesses",
-    "alchemy manuscript: transformation diagrams, annotated symbols, recipe-like "
-    "process artifacts, and illuminated margins",
-    "brutalist civic planning: concrete signage, zoning blocks, transit diagrams, "
-    "infrastructure maps, and stern public-service clarity",
-)
-
-
-def build_change_shape(
-    context: Mapping[str, Any],
-    *,
-    max_commits: int = 10,
-    max_files: int = 10,
-) -> str:
-    """Render a compact factual digest of the PR: labels, linked issues, commits, files.
-
-    Mirrors the delivery context the Basic Memory CI capture flow collects
-    (ProjectUpdateContext): the goal is to ground the image in the theme of the
-    WHOLE change — what it touches and why — not just the title/description.
-    """
-    lines: list[str] = []
-
-    labels = [str(label.get("name", "")) for label in context.get("labels") or []]
-    labels = [label for label in labels if label]
-    if labels:
-        lines.append(f"Labels: {', '.join(labels)}")
-
-    issues = context.get("closingIssuesReferences") or []
-    if issues:
-        lines.append("Linked issues:")
-        for issue in issues:
-            number = issue.get("number")
-            title = str(issue.get("title") or "").strip()
-            lines.append(f"- #{number}: {title}" if title else f"- #{number}")
-
-    commits = context.get("commits") or []
-    subjects = [str(commit.get("messageHeadline") or "").strip() for commit in commits]
-    # Merge commits carry no thematic signal — they're branch bookkeeping.
-    subjects = [
-        subject
-        for subject in subjects
-        if subject and not subject.startswith(("Merge branch ", "Merge pull request "))
-    ]
-    if subjects:
-        lines.append(f"Commit subjects ({len(subjects)} total):")
-        lines.extend(f"- {subject}" for subject in subjects[:max_commits])
-        if len(subjects) > max_commits:
-            lines.append(f"- ... and {len(subjects) - max_commits} more")
-
-    files = context.get("files") or []
-    if files:
-        additions = sum(int(item.get("additions") or 0) for item in files)
-        deletions = sum(int(item.get("deletions") or 0) for item in files)
-        lines.append(f"Files changed ({len(files)} total, +{additions}/-{deletions}):")
-        ranked = sorted(
-            files,
-            key=lambda item: int(item.get("additions") or 0) + int(item.get("deletions") or 0),
-            reverse=True,
-        )
-        for item in ranked[:max_files]:
-            path = str(item.get("path") or "")
-            lines.append(f"- {path} (+{item.get('additions') or 0}/-{item.get('deletions') or 0})")
-        if len(files) > max_files:
-            lines.append(f"- ... and {len(files) - max_files} more files")
-
-    return "\n".join(lines) if lines else "(no additional change context available)"
-
-
-def extract_pr_content(pr_body: str) -> str:
-    """Return the author's own PR description with all managed bot blocks removed."""
-    content = pr_body
-    for start, end in MANAGED_BLOCKS:
-        content = re.sub(
-            rf"{re.escape(start)}.*?{re.escape(end)}",
-            "",
-            content,
-            flags=re.DOTALL,
-        )
-    return content.strip()
-
-
-def extract_infographic_theme(pr_body: str) -> str | None:
-    pattern = re.compile(
-        rf"{re.escape(THEME_START)}\s*(.*?)\s*{re.escape(THEME_END)}",
-        flags=re.DOTALL,
-    )
-    match = pattern.search(pr_body)
-    if not match:
-        return None
-    theme = match.group(1).strip()
-    return theme or None
-
-
-def select_image_theme(
-    *,
-    pr_number: int,
-    pr_title: str,
-    pr_body: str,
-    theme_override: str | None,
-) -> ThemeSelection:
-    if theme_override:
-        return ThemeSelection(theme=theme_override, source=ThemeSource.CLI)
-    body_theme = extract_infographic_theme(pr_body)
-    if body_theme:
-        return ThemeSelection(theme=body_theme, source=ThemeSource.PR_BODY)
-    # Seed on author-owned PR identity, not the review summary, so the pick is
-    # stable across re-reviews of the same PR.
-    seed = f"{pr_number}\n{pr_title}".encode("utf-8")
-    index = int.from_bytes(hashlib.sha256(seed).digest()[:2], byteorder="big") % len(
-        BM_IMAGE_THEME_POOL
-    )
-    return ThemeSelection(theme=BM_IMAGE_THEME_POOL[index], source=ThemeSource.AUTO)
-
-
-def _preformatted(value: str) -> str:
-    return f"<pre><code>{html.escape(value, quote=False)}</code></pre>"
-
-
-def build_infographic_provenance_block(
-    *,
-    pr_number: int,
-    output_path: Path,
-    model: str,
-    size: str,
-    quality: str,
-    theme: str,
-    theme_source: ThemeSource,
-) -> str:
-    return f"""
-{PROVENANCE_START}
-<details>
-<summary>BM Bossbot image choices</summary>
-
-- Pull request: `#{pr_number}`
-- Generated asset: `{output_path.as_posix()}`
-- Image model: `{model}`
-- Size: `{size}`
-- Quality: `{quality}`
-- Image mode: `editorial-image`
-- Theme source: `{theme_source.value}`
-
-Theme / visual direction:
-{_preformatted(theme)}
-
-</details>
-{PROVENANCE_END}
-""".strip()
-
-
-def upsert_managed_block(body: str, *, block: str, start: str, end: str) -> str:
-    pattern = re.compile(rf"{re.escape(start)}.*?{re.escape(end)}", flags=re.DOTALL)
-    if pattern.search(body):
-        return pattern.sub(block, body, count=1)
-    if body.strip():
-        return f"{body.rstrip()}\n\n{block}\n"
-    return f"{block}\n"
-
-
-def build_infographic_prompt(
-    *,
-    pr_number: int,
-    pr_title: str,
-    pr_content: str,
-    change_shape: str,
-    theme: str,
-    theme_source: ThemeSource,
-) -> str:
-    theme_label = (
-        "Selected BM visual direction"
-        if theme_source == ThemeSource.AUTO
-        else "User-supplied visual direction"
-    )
-
-    return f"""
-Create a polished landscape WebP editorial image for Basic Memory PR #{pr_number}.
-
-Your subject is the CONTENT of the pull request — what the change does and why
-it matters — described in the title, description, and change shape below.
-Express the theme of the whole change as a visual story.
-
-This image is decoration for the PR conversation. It is NOT a review artifact:
-do not depict review verdicts, approval, or process. Never render approval
-stamps, "APPROVED"/"SUCCESS"/"VERDICT" wording, rubber stamps, wax seals of
-approval, badges, checkmarks, checklists, status lines, SHA strings, or
-BM Bossbot itself. If the composition needs text, draw it from the change's
-subject matter only.
-
-Pull request title:
-{pr_title}
-
-Pull request description:
-{pr_content}
-
-Change shape — factual delivery context (labels, linked issues, commit
-subjects, changed files). Use it to understand what the whole PR touches and
-let that steer the imagery. It is context, NOT captions: never render file
-paths, diff stats, issue numbers, or commit subjects verbatim in the image.
-{change_shape}
-
-{theme_label}:
-{theme}
-
-Treat the visual direction as style inspiration only. Do not let it override
-facts, readability, source material, or the prohibition on review imagery.
-
-Use image-first composition: create a scene, movie poster, editorial painting,
-classic photograph, cover image, symbolic tableau, staged artifact, or another
-visual moment that expresses the PR intent.
-
-Make the selected direction shape the subject, lighting, composition, props,
-environment, and mood. Use one strong focal point. Prefer visual metaphor over
-explanatory UI.
-
-Use at most a short title and zero to three short labels if text helps. Any text
-that appears must be high-contrast, smooth, anti-aliased, and readable.
-
-Do not render an infographic, dashboard, flowchart, timeline strip, checklist,
-bullet-list panel, data panel, or dense explanatory diagram.
-
-Avoid fake screenshots, code blocks, invented claims, copyrighted characters,
-logos, named fictional universes, direct band logos, album art, celebrity
-likenesses, or decorations that obscure content.
-""".strip()
-
-
-@app.command()
-def generate(
-    pr_number: Annotated[
-        int,
-        typer.Option("--pr-number", min=1, help="Pull request number."),
-    ],
-    pr_context_file: Annotated[
-        Path,
-        typer.Option(
-            "--pr-context-file",
-            exists=True,
-            dir_okay=False,
-            readable=True,
-            help=(
-                "JSON from `gh pr view --json "
-                "title,body,labels,files,commits,closingIssuesReferences`."
-            ),
-        ),
-    ],
-    output: Annotated[Path, typer.Option("--output", help="Output .webp path.")],
-    model: Annotated[str, typer.Option("--model", help="OpenAI image model.")] = DEFAULT_MODEL,
-    size: Annotated[str, typer.Option("--size", help="Image size.")] = DEFAULT_SIZE,
-    quality: Annotated[str, typer.Option("--quality", help="Image quality.")] = DEFAULT_QUALITY,
-    retries: Annotated[int, typer.Option("--retries", min=0, help="Retry attempts.")] = 2,
-    theme: Annotated[
-        str | None,
-        typer.Option("--theme", help="Optional visual theme preference."),
-    ] = None,
-    provenance_output: Annotated[
-        Path | None,
-        typer.Option(
-            "--provenance-output",
-            dir_okay=False,
-            help="Optional file to write the managed PR-body provenance block.",
-        ),
-    ] = None,
-    print_prompt: Annotated[
-        bool,
-        typer.Option(
-            "--print-prompt",
-            "--dry-run",
-            help="Print the generated prompt and exit without calling OpenAI. Alias: --dry-run.",
-        ),
-    ] = False,
-) -> None:
-    """Generate the canonical PR image from the PR's title, description, and change shape."""
-    context = json.loads(pr_context_file.read_text(encoding="utf-8"))
-    if not isinstance(context, Mapping):
-        raise typer.BadParameter("PR context file must contain a JSON object")
-    pr_title = str(context.get("title") or "")
-    pr_body = str(context.get("body") or "")
-    pr_content = extract_pr_content(pr_body)
-    change_shape = build_change_shape(context)
-    theme_selection = select_image_theme(
-        pr_number=pr_number,
-        pr_title=pr_title,
-        pr_body=pr_body,
-        theme_override=theme,
-    )
-    prompt = build_infographic_prompt(
-        pr_number=pr_number,
-        pr_title=pr_title,
-        pr_content=pr_content,
-        change_shape=change_shape,
-        theme=theme_selection.theme,
-        theme_source=theme_selection.source,
-    )
-    if print_prompt:
-        typer.echo(prompt)
-        raise typer.Exit()
-
-    image_result = generate_image_result(
-        prompt=prompt,
-        output_path=output,
-        model=model,
-        size=size,
-        quality=quality,
-        retries=retries,
-    )
-    output_path = image_result.path
-    if provenance_output:
-        provenance_output.parent.mkdir(parents=True, exist_ok=True)
-        provenance_output.write_text(
-            build_infographic_provenance_block(
-                pr_number=pr_number,
-                output_path=output_path,
-                model=model,
-                size=size,
-                quality=quality,
-                theme=theme_selection.theme,
-                theme_source=theme_selection.source,
-            ),
-            encoding="utf-8",
-        )
-    typer.echo(output_path)
-
-
-def main() -> None:
-    app()
-
-
-if __name__ == "__main__":
-    main()

server.json

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

src/basic_memory/__init__.py

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

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

@@ -10,10 +10,18 @@ Key improvements:
 - Simplified caching strategies
 """
 
+import os
+import pathlib
+
 from fastapi import APIRouter, HTTPException, Response, Path
 from loguru import logger
 
 import logfire
+from basic_memory.ignore_utils import (
+    IGNORED_PATH_REJECTION_DETAIL,
+    load_gitignore_patterns,
+    should_ignore_path,
+)
 from basic_memory.deps import (
     EntityServiceV2ExternalDep,
     SearchServiceV2ExternalDep,
@@ -24,6 +32,7 @@ from basic_memory.deps import (
     EntityRepositoryV2ExternalDep,
     RelationRepositoryV2ExternalDep,
     ProjectExternalIdPathDep,
+    SyncServiceV2ExternalDep,
     TaskSchedulerDep,
 )
 from basic_memory.schemas import DeleteEntitiesResponse
@@ -40,8 +49,10 @@ from basic_memory.schemas.v2 import (
     MoveDirectoryRequestV2,
     DeleteDirectoryRequestV2,
     OrphanEntitiesResponse,
+    SyncFileRequest,
 )
 from basic_memory.schemas.response import DirectoryMoveResult, DirectoryDeleteResult
+from basic_memory.utils import validate_project_path
 
 router = APIRouter(prefix="/knowledge", tags=["knowledge-v2"])
 
@@ -236,6 +247,187 @@ async def resolve_identifier(
         return result
 
 
+## Single-file sync endpoint
+
+
+def _canonical_file_path(home: pathlib.Path, segments: list[str]) -> str | None:
+    """Resolve the actual on-disk casing of a file path under the project home.
+
+    Trigger: case-insensitive filesystems (macOS/Windows) pass existence checks for
+        wrong-cased paths like 'notes/Disk-Note.md' when the file is 'notes/disk-note.md'.
+    Why: indexing the caller-supplied casing misses the existing DB row keyed by the
+        on-disk path and inserts a duplicate entity under the wrong-cased path.
+    Outcome: each segment is matched against real directory entries — exact name first
+        (so distinct case-variant files on case-sensitive filesystems stay distinct),
+        then a unique case-insensitive match. Returns None when any segment cannot be
+        matched to exactly one entry, including missing files. Traversal stops at the
+        project boundary: a directory whose resolved path escapes the project home is
+        never scanned.
+    """
+    resolved_home = home.resolve()
+    current = home
+    canonical_segments: list[str] = []
+    for segment in segments:
+        # Trigger: a previously matched segment may be a symlink whose target lies
+        #     outside the project root (e.g. wrong-cased 'LINK' matched the on-disk
+        #     'link' -> /tmp/outside on a case-sensitive filesystem).
+        # Why: os.scandir follows symlinked directories, so continuing would read
+        #     directory contents outside the project boundary even though the
+        #     post-canonicalization containment check rejects the request later.
+        # Outcome: bail before scanning the moment resolution escapes the home.
+        if not current.resolve().is_relative_to(resolved_home):
+            return None
+        try:
+            with os.scandir(current) as entries_iter:
+                entries = [entry.name for entry in entries_iter]
+        except OSError:
+            # A parent segment resolved to a non-directory (or vanished): no canonical
+            # path exists for the remaining segments.
+            return None
+        if segment in entries:
+            matched = segment
+        else:
+            matches = [entry for entry in entries if entry.lower() == segment.lower()]
+            if len(matches) != 1:
+                return None
+            matched = matches[0]
+        canonical_segments.append(matched)
+        current = current / matched
+    return "/".join(canonical_segments)
+
+
+@router.post("/sync-file", response_model=EntityResponseV2)
+async def sync_file(
+    data: SyncFileRequest,
+    project_id: ProjectExternalIdPathDep,
+    sync_service: SyncServiceV2ExternalDep,
+    project_config: ProjectConfigV2ExternalDep,
+    search_service: SearchServiceV2ExternalDep,
+    app_config: AppConfigDep,
+) -> EntityResponseV2:
+    """Index a single markdown file that exists on disk but is not indexed yet.
+
+    Recovery path for files written directly to disk before the watcher indexed
+    them (#581): callers such as edit_note can index the exact file and retry
+    identifier resolution without running a full project sync.
+
+    Args:
+        data: Request containing the markdown file path relative to project root
+
+    Returns:
+        The indexed entity
+
+    Raises:
+        HTTPException: 400 if the path escapes the project root, contains
+            non-normalized segments, matches the project ignore rules, or is
+            not markdown, 404 if the file does not exist on disk
+    """
+    with logfire.span(
+        "api.request.knowledge.sync_file",
+        entrypoint="api",
+        domain="knowledge",
+        action="sync_file",
+    ):
+        logger.info(f"API v2 request: sync_file file_path='{data.file_path}'")
+
+        if not validate_project_path(data.file_path, project_config.home):
+            raise HTTPException(
+                status_code=400,
+                detail=f"File path '{data.file_path}' is not allowed - "
+                "paths must stay within project boundaries",
+            )
+
+        # Trigger: segments like './' or '//' survive the traversal check above
+        # Why: a non-normalized path would index under a non-canonical DB key
+        # Outcome: reject fail-fast instead of guessing the canonical form
+        segments = data.file_path.replace("\\", "/").split("/")
+        if any(segment in ("", ".") for segment in segments):
+            raise HTTPException(
+                status_code=400,
+                detail=f"File path '{data.file_path}' is not normalized - "
+                "segments like './' or '//' are not allowed",
+            )
+
+        # Canonicalize to the actual on-disk casing so the DB lookup below hits the
+        # row keyed by the real path instead of inserting a wrong-cased duplicate.
+        file_path = _canonical_file_path(project_config.home, segments)
+        if file_path is None:
+            raise HTTPException(
+                status_code=404, detail=f"File not found on disk: '{data.file_path}'"
+            )
+        # Trigger: canonicalization rewrote a segment to its on-disk form, and that
+        #     segment may be a symlink. The pre-check above validated the ORIGINAL
+        #     request path — on a case-sensitive filesystem 'LINK/secret.md' does not
+        #     exist, so resolve() cannot follow the real 'link' symlink and the check
+        #     passes even when 'link' points outside the project root.
+        # Why: indexing through an escaping symlink would read and index content
+        #     outside the project boundary — and even an is_file() existence probe on
+        #     the joined path would follow the symlink and stat its target, so
+        #     containment must hold BEFORE any filesystem probe that follows symlinks.
+        #     Path.resolve() only walks symlink names (readlink); it never opens or
+        #     stats the final target, so it is safe to run pre-containment.
+        # Outcome: the canonical path is re-validated and the fully-resolved absolute
+        #     target must stay inside the resolved project home; escapes get a 400
+        #     before the file-existence probe below ever touches the target.
+        resolved_target = (project_config.home / file_path).resolve()
+        if not validate_project_path(file_path, project_config.home) or not (
+            resolved_target.is_relative_to(project_config.home.resolve())
+        ):
+            raise HTTPException(
+                status_code=400,
+                detail=f"File path '{data.file_path}' is not allowed - "
+                "paths must stay within project boundaries",
+            )
+        # Containment holds, so probing the resolved target cannot leave the project.
+        if not resolved_target.is_file():
+            raise HTTPException(
+                status_code=404, detail=f"File not found on disk: '{data.file_path}'"
+            )
+        # Trigger: the canonical path matches the .bmignore / project .gitignore rules
+        # Why: scan and watch flows filter ignored files before they ever reach the
+        #      indexer; indexing one here would bypass the ignored-file contract and
+        #      make hidden or gitignored content searchable
+        # Outcome: the same should_ignore_path() rules apply to single-file sync
+        ignore_patterns = load_gitignore_patterns(project_config.home)
+        if should_ignore_path(
+            project_config.home / file_path, project_config.home, ignore_patterns
+        ):
+            raise HTTPException(
+                status_code=400,
+                detail=f"File path '{data.file_path}' {IGNORED_PATH_REJECTION_DETAIL} "
+                "and cannot be indexed",
+            )
+        if not sync_service.file_service.is_markdown(file_path):
+            raise HTTPException(
+                status_code=400,
+                detail=f"Only markdown files can be indexed: '{data.file_path}'",
+            )
+
+        # Trigger: the file may already have a DB row (e.g. modified on disk after indexing)
+        # Why: the indexer needs to know whether to insert or update the entity
+        # Outcome: new is computed from the database instead of assumed by the caller
+        existing = await sync_service.entity_repository.get_by_file_path(file_path)
+        synced = await sync_service.sync_one_markdown_file(
+            file_path, new=existing is None, index_search=True
+        )
+
+        # Trigger: semantic search is enabled and the entity index was just refreshed
+        # Why: the project sync flow awaits sync_entity_vectors_batch() inline after
+        #      indexing changed files (SyncService.sync); without the single-entity
+        #      equivalent, a note recovered via sync-file stays missing or stale in
+        #      semantic search until a later edit or full project sync
+        # Outcome: vectors refresh synchronously before the response returns,
+        #          mirroring the sync flow instead of the out-of-band scheduler
+        if app_config.semantic_search_enabled:
+            await search_service.sync_entity_vectors_batch([synced.entity.id])
+
+        result = EntityResponseV2.model_validate(synced.entity)
+        logger.info(
+            f"API v2 response: sync_file file_path='{file_path}' external_id={result.external_id}"
+        )
+        return result
+
+
 ## Read endpoints
 
 

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

@@ -512,7 +512,11 @@ def bisync_project_command(
     resync: bool = typer.Option(False, "--resync", help="Force new baseline"),
     verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
 ) -> None:
-    """Two-way sync: local <-> cloud (bidirectional sync).
+    """Two-way mirror: local <-> cloud (bidirectional sync).
+
+    Personal workspaces only. This mirror can delete and overwrite files on both
+    sides, so on Team workspaces use `bm cloud pull` (fetch) / `bm cloud push`
+    (additive upload) instead.
 
     Examples:
       bm cloud bisync --name research --resync  # First time
@@ -576,7 +580,11 @@ def check_project_command(
     name: str = typer.Option(..., "--name", "--project", help="Project name to check"),
     one_way: bool = typer.Option(False, "--one-way", help="Check one direction only (faster)"),
 ) -> None:
-    """Verify file integrity between local and cloud.
+    """Verify file integrity between local and cloud (no changes made).
+
+    Personal workspaces only: check compares against the Personal workspace
+    mirror remote. On Team workspaces use `bm cloud pull --dry-run` /
+    `bm cloud push --dry-run` to preview differences instead.
 
     Example:
       bm cloud check --name research
@@ -624,6 +632,9 @@ def bisync_reset(
 ) -> None:
     """Clear bisync state for a project.
 
+    Personal workspaces only (bisync is a Personal-workspace mirror; on Team
+    workspaces use `bm cloud pull` / `bm cloud push` instead).
+
     This removes the bisync metadata files, forcing a fresh --resync on next bisync.
     Useful when bisync gets into an inconsistent state or when remote path changes.
     """

src/basic_memory/cli/commands/tool.py

@@ -1,18 +1,42 @@
 """CLI tool commands for Basic Memory.
 
 Every command calls its MCP tool with output_format="json" and prints the result.
-No text formatting, no separate code paths, no duplicate data fetching.
+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``).
 """
 
 import json
 import sys
-from typing import Annotated, Any, Dict, List, Optional
+from typing import Annotated, Any, Dict, List, Literal, 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
@@ -32,15 +56,410 @@ 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")
@@ -181,7 +600,16 @@ def write_note(
 def read_note(
     identifier: str,
     include_frontmatter: bool = typer.Option(
-        False, "--include-frontmatter", help="Include YAML frontmatter in output"
+        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"
     ),
     project: Annotated[
         Optional[str],
@@ -201,13 +629,21 @@ 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 --include-frontmatter
+    bm tool read-note my-note --frontmatter
+    bm tool read-note my-note --plain
+    bm tool read-note my-note --json
     """
     try:
         validate_routing_flags(local, cloud)
+        _validate_output_flags(json_output, plain)
 
         with force_routing(local=local, cloud=cloud):
             result = run_with_cleanup(
@@ -232,7 +668,15 @@ def read_note(
             _print_json(result)
             raise typer.Exit(1)
 
-        _print_json(result)
+        # 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)
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)
@@ -390,6 +834,12 @@ 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."),
@@ -408,13 +858,21 @@ 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(
@@ -430,7 +888,15 @@ def build_context(
                     output_format="json",
                 )
             )
-        _print_json(result)
+
+        # 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)
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)
@@ -452,6 +918,12 @@ 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."),
@@ -470,14 +942,22 @@ 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(
@@ -492,7 +972,15 @@ def recent_activity(
                     output_format="json",
                 )
             )
-        _print_json(result)
+
+        # 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)
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)
@@ -556,6 +1044,12 @@ 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."),
@@ -574,6 +1068,11 @@ 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"
@@ -581,9 +1080,12 @@ 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
@@ -658,7 +1160,13 @@ def search_notes(
             typer.echo(result, err=True)
             raise typer.Exit(1)
 
-        _print_json(result)
+        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 "")
     except ValueError as e:
         typer.echo(f"Error: {e}", err=True)
         raise typer.Exit(1)

src/basic_memory/config.py

@@ -442,6 +442,18 @@ 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.",

src/basic_memory/ignore_utils.py

@@ -7,6 +7,15 @@ from typing import Set
 from basic_memory.config import resolve_data_dir
 
 
+# Marker shared by the API ignored-path rejection detail and MCP-side error handling.
+# The sync-file endpoint embeds it in its 400 detail and edit_note's disk recovery
+# matches on it, so "exists but ignored" stays distinguishable from generic rejections
+# without duplicating message text across layers.
+IGNORED_PATH_REJECTION_DETAIL = (
+    "matches Basic Memory ignore rules (.bmignore or project .gitignore)"
+)
+
+
 # Common directories and patterns to ignore by default
 # These are used as fallback if .bmignore doesn't exist
 DEFAULT_IGNORE_PATTERNS = {

src/basic_memory/mcp/clients/knowledge.py

@@ -276,6 +276,35 @@ class KnowledgeClient:
             )
         return DirectoryDeleteResult.model_validate(response.json())
 
+    # --- Single-file sync ---
+
+    async def sync_file(self, file_path: str) -> EntityResponse:
+        """Index a markdown file that exists on disk but is not indexed yet.
+
+        Args:
+            file_path: Markdown file path relative to the project root
+
+        Returns:
+            EntityResponse for the indexed entity
+
+        Raises:
+            ToolError: If the file does not exist on disk or indexing fails
+        """
+        with logfire.span(
+            "mcp.client.knowledge.sync_file",
+            client_name="knowledge",
+            operation="sync_file",
+        ):
+            response = await call_post(
+                self.http_client,
+                f"{self._base_path}/sync-file",
+                json={"file_path": file_path},
+                client_name="knowledge",
+                operation="sync_file",
+                path_template="/v2/projects/{project_id}/knowledge/sync-file",
+            )
+        return EntityResponse.model_validate(response.json())
+
     # --- Orphan detection ---
 
     async def get_orphans(self) -> list[GraphNode]:

src/basic_memory/mcp/tools/edit_note.py

@@ -1,13 +1,19 @@
 """Edit note tool for Basic Memory MCP server."""
 
-from typing import Annotated, Optional, Literal
+from typing import TYPE_CHECKING, Annotated, Optional, Literal
 
 import logfire
+from httpx import HTTPStatusError
 from loguru import logger
 from fastmcp import Context
+from mcp.server.fastmcp.exceptions import ToolError
 from pydantic import AliasChoices, Field
 
+if TYPE_CHECKING:  # pragma: no cover
+    from basic_memory.mcp.clients import KnowledgeClient
+
 from basic_memory.config import ConfigManager
+from basic_memory.ignore_utils import IGNORED_PATH_REJECTION_DETAIL
 from basic_memory.mcp.project_context import (
     _workspace_identifier_discovery_available,
     detect_project_from_memory_url_prefix,
@@ -16,6 +22,7 @@ from basic_memory.mcp.project_context import (
     resolve_project_and_path,
 )
 from basic_memory.mcp.server import mcp
+from basic_memory.mcp.tools.utils import _extract_response_data, _response_detail_text
 from basic_memory.schemas.base import Entity
 from basic_memory.schemas.response import EntityResponse
 from basic_memory.services.link_resolver import (
@@ -52,6 +59,79 @@ def _parse_identifier_to_title_and_directory(identifier: str) -> tuple[str, str]
     return title, directory
 
 
+# Suffixes mimetypes maps to text/markdown (extension matching is case-insensitive),
+# mirroring FileService.is_markdown which gates the sync-file endpoint server-side.
+_MARKDOWN_SUFFIXES = (".md", ".markdown")
+
+
+async def _resolve_after_disk_recovery(
+    knowledge_client: "KnowledgeClient",
+    identifier: str,
+) -> Optional[str]:
+    """Recover from a resolution miss when the note exists on disk but is not indexed.
+
+    Trigger: identifier resolution failed with "not found", but the identifier may map
+        to a markdown file written directly to disk before the watcher indexed it (#581).
+    Why: editing an on-disk note should not require a manual full sync or watcher restart.
+    Outcome: the single file is indexed server-side and resolution is retried exactly
+        once. Returns None when the identifier does not map to an indexable file on
+        disk, so the caller keeps its existing not-found handling.
+    """
+    # Try the identifier as-is first so existing .markdown/.MD files are found; only
+    # fall back to appending markdown suffixes (".md" first, then ".markdown") when
+    # the identifier does not already carry one, so 'notes/foo.markdown' never becomes
+    # 'notes/foo.markdown.md' and a stem identifier still reaches 'notes/foo.markdown'.
+    candidates = [identifier]
+    if not identifier.lower().endswith(_MARKDOWN_SUFFIXES):
+        candidates.extend(f"{identifier}{suffix}" for suffix in _MARKDOWN_SUFFIXES)
+
+    for candidate in candidates:
+        try:
+            synced = await knowledge_client.sync_file(candidate)
+        except ToolError as sync_error:
+            # Trigger: the sync-file request failed
+            # Why: 400/404 are the expected "nothing to recover" rejections (missing
+            #      file, traversal, non-markdown) — except the ignored-path 400, which
+            #      means the file exists on disk but the ignore rules forbid indexing
+            #      it, so falling through to auto-create would silently shadow the
+            #      file. Anything else — auth, server, transport-level failures — is a
+            #      real error that must not be masked as a not-found miss.
+            # Outcome: ignored-path rejections raise a clear ToolError; other expected
+            #      rejections try the next candidate or fall through to the caller's
+            #      existing not-found behavior; unexpected failures propagate.
+            cause = sync_error.__cause__
+            candidate_rejected = isinstance(
+                cause, HTTPStatusError
+            ) and cause.response.status_code in (400, 404)
+            if not candidate_rejected:
+                raise
+            detail = _response_detail_text(_extract_response_data(cause.response)) or ""
+            if IGNORED_PATH_REJECTION_DETAIL in detail:
+                raise ToolError(
+                    f"Note file '{candidate}' exists on disk but {IGNORED_PATH_REJECTION_DETAIL} "
+                    "and will not be edited"
+                ) from sync_error
+            logger.debug(f"edit_note disk recovery skipped for '{candidate}': {sync_error}")
+            continue
+
+        # Trigger: sync-file succeeded and returned the indexed entity.
+        # Why: the server may have canonicalized the path casing (notes/Disk-Note ->
+        #      notes/disk-note.md), so strictly re-resolving the raw identifier can
+        #      still miss the entity we just indexed.
+        # Outcome: use the entity identity from the sync-file response directly; only
+        #      fall back to a strict re-resolve when an older server omits external_id,
+        #      and let that re-resolve fail loudly instead of guessing.
+        if synced.external_id:
+            logger.info(
+                f"edit_note indexed unindexed file '{candidate}' as entity {synced.external_id}"
+            )
+            return synced.external_id
+        logger.info(f"edit_note indexed unindexed file '{candidate}'; retrying resolution")
+        return await knowledge_client.resolve_entity(identifier, strict=True)
+
+    return None
+
+
 def _compose_workspace_project_route(
     *,
     workspace: Optional[str],
@@ -126,7 +206,8 @@ The note with identifier '{identifier}' could not be found. The `find_replace` a
 ## Suggestions to try:
 1. **Use append/prepend instead**: These operations will create the note automatically if it doesn't exist
 2. **Search for the note first**: Use `search_notes("{project or "project-name"}", "{identifier.split("/")[-1]}")` to find similar notes with exact identifiers
-3. **Try different exact identifier formats**:
+3. **File exists on disk but is not indexed yet?**: edit_note indexes the file automatically when the identifier matches its path (e.g. 'folder/note' for 'folder/note.md'). If your identifier is a title or differs from the file path, run a sync (`basic-memory sync`) or wait for the file watcher, then retry
+4. **Try different exact identifier formats**:
    - If you used a permalink like "folder/note-title", try the exact title: "{identifier.split("/")[-1].replace("-", " ").title()}"
    - If you used a title, try the exact permalink format: "{identifier.lower().replace(" ", "-")}"
    - Use `read_note("{project or "project-name"}", "{identifier}")` first to verify the note exists and get the exact identifier
@@ -343,7 +424,9 @@ async def edit_note(
 
     Note:
         Edit operations require exact identifier matches. If unsure, use read_note() or
-        search_notes() first to find the correct identifier. The tool provides detailed
+        search_notes() first to find the correct identifier. When the identifier looks
+        like a file path and the file exists on disk but is not indexed yet, edit_note
+        indexes that file automatically and retries the edit. The tool provides detailed
         error messages with suggestions if operations fail.
     """
     # Resolve effective default: allow MCP clients to send null for optional int field
@@ -465,14 +548,27 @@ async def edit_note(
                         strict=True,
                     )
                 except Exception as resolve_error:
-                    # Trigger: entity does not exist yet
-                    # Why: append/prepend can meaningfully create a new note from the content,
-                    #      while find_replace/replace_section require existing content to modify
-                    # Outcome: note is created via the same path as write_note
                     error_msg = str(resolve_error).lower()
                     is_not_found = "entity not found" in error_msg or "not found" in error_msg
 
-                    if is_not_found and operation in ("append", "prepend"):
+                    # Trigger: resolution missed but the file may already exist on disk
+                    # Why: files written directly to disk are invisible to identifier
+                    #      resolution until indexed; editing them should just work (#581)
+                    # Outcome: the single file is indexed and resolution retried once
+                    recovered_entity_id: str | None = None
+                    if is_not_found:
+                        recovered_entity_id = await _resolve_after_disk_recovery(
+                            knowledge_client, entity_identifier
+                        )
+
+                    if recovered_entity_id is not None:
+                        entity_id = recovered_entity_id
+                    elif is_not_found and operation in ("append", "prepend"):
+                        # Trigger: entity does not exist yet (on disk or in the index)
+                        # Why: append/prepend can meaningfully create a new note from the
+                        #      content, while find_replace/replace_section require existing
+                        #      content to modify
+                        # Outcome: note is created via the same path as write_note
                         title, directory = _parse_identifier_to_title_and_directory(identifier)
 
                         # Validate directory path (same security check as write_note)

src/basic_memory/mcp/tools/search.py

@@ -11,7 +11,13 @@ from fastmcp import Context
 from pydantic import AliasChoices, BeforeValidator, Field
 
 from basic_memory.config import ConfigManager, has_cloud_credentials
-from basic_memory.utils import build_canonical_permalink, coerce_dict, coerce_list, parse_tags
+from basic_memory.utils import (
+    build_canonical_permalink,
+    coerce_dict,
+    coerce_list,
+    parse_tags,
+    strict_search_tags,
+)
 from basic_memory.mcp.async_client import (
     _explicit_routing,
     _force_local_mode,
@@ -676,13 +682,15 @@ async def search_notes(
         Dict[str, Any] | None,
         BeforeValidator(coerce_dict),
     ] = None,
-    # parse_tags, not coerce_list: tags="a,b" must split into ["a", "b"] to match the
-    # tag: query shorthand below and write_note's documented tags convention (#910).
-    # coerce_list would wrap the comma string as the single literal tag ["a,b"],
-    # which matches nothing.
+    # strict_search_tags, not coerce_list: tags="a,b" must split into ["a", "b"] to
+    # match the tag: query shorthand below and write_note's documented tags convention
+    # (#910). coerce_list would wrap the comma string as the single literal tag
+    # ["a,b"], which matches nothing. Unlike bare parse_tags, the strict wrapper only
+    # splits str/list/None and lets Pydantic reject other types (42, {"a": 1}) with a
+    # clear validation error instead of stringifying them into junk tags.
     tags: Annotated[
         List[str] | None,
-        BeforeValidator(parse_tags),
+        BeforeValidator(strict_search_tags),
     ] = None,
     status: Optional[str] = None,
     min_similarity: Annotated[
@@ -893,6 +901,15 @@ async def search_notes(
     # so preserve their original casing (unlike the lowercased note_types).
     categories = categories or []
 
+    # Trigger: tags arrived via a direct function call instead of the MCP layer.
+    # Why: the BeforeValidator above only runs through MCP/Pydantic validation; direct
+    #      callers (e.g. `bm tool search-notes --tag a,b` in cli/commands/tool.py, which
+    #      Typer collects as the one-element list ["a,b"]) would otherwise forward the
+    #      comma string as one literal tag that matches nothing (#910).
+    # Outcome: comma-split/list normalization applies on every path; parse_tags is
+    #          idempotent, so MCP-validated input passes through unchanged.
+    tags = parse_tags(tags) or None
+
     # Parse tag:<value> shorthand at tool level so it works with all search modes.
     # Handles "tag:security", "tag:coffee tag:brewing", "tag:coffee AND tag:brewing".
     # Without this, hybrid/vector modes fail because they require non-empty text,

src/basic_memory/schemas/v2/__init__.py

@@ -9,6 +9,7 @@ from basic_memory.schemas.v2.entity import (
     DeleteDirectoryRequestV2,
     ProjectResolveRequest,
     ProjectResolveResponse,
+    SyncFileRequest,
 )
 from basic_memory.schemas.v2.graph import (
     GraphEdge,
@@ -31,6 +32,7 @@ __all__ = [
     "DeleteDirectoryRequestV2",
     "ProjectResolveRequest",
     "ProjectResolveResponse",
+    "SyncFileRequest",
     "GraphEdge",
     "GraphNode",
     "GraphResponse",

src/basic_memory/schemas/v2/entity.py

@@ -54,6 +54,21 @@ class EntityResolveResponse(BaseModel):
     )
 
 
+class SyncFileRequest(BaseModel):
+    """Request to index a single markdown file that exists on disk.
+
+    Used as a recovery path when an identifier fails resolution but maps to a
+    file written directly to disk that the watcher has not indexed yet (#581).
+    """
+
+    file_path: str = Field(
+        ...,
+        description="Markdown file path to index (relative to project root)",
+        min_length=1,
+        max_length=500,
+    )
+
+
 class MoveEntityRequestV2(BaseModel):
     """V2 request schema for moving an entity to a new file location.
 

src/basic_memory/services/context_service.py

@@ -245,9 +245,12 @@ class ContextService:
                                     type="observation",
                                     id=obs.id,
                                     title=f"{obs.category}: {obs.content[:50]}...",
-                                    permalink=generate_permalink(
-                                        f"{primary_item.permalink}/observations/{obs.category}/{obs.content}"
-                                    ),
+                                    # Observation.permalink is the single definition of the
+                                    # synthetic permalink format (200-char truncation plus
+                                    # content digest); rebuilding it inline diverged from the
+                                    # search index for long observations (#929). The parent
+                                    # entity is eager-loaded by ObservationRepository.
+                                    permalink=obs.permalink,
                                     file_path=primary_item.file_path,
                                     content=obs.content,
                                     category=obs.category,

src/basic_memory/utils.py

@@ -568,6 +568,38 @@ def parse_tags(tags: Union[List[str], str, None]) -> List[str]:
         return []
 
 
+def strict_search_tags(v: Any) -> Any:
+    """Strictly coerce tag input at the search_notes tool boundary.
+
+    parse_tags stringifies anything (42 -> ["42"], {"a": 1} -> junk tags), which would
+    turn caller type mistakes into silent no-result searches. At the tool boundary only
+    str, all-string lists, and None are valid tag inputs; everything else — including
+    lists with non-string elements like [42] — passes through unchanged so Pydantic
+    rejects it with a clear validation error.
+
+    JSON array strings (the MCP clients-serialize-arrays-as-strings path) get the same
+    all-string check: '[42]' or '["ok", 42]' would otherwise be stringified by
+    parse_tags' recursive JSON handling before Pydantic ever sees the bad elements.
+    """
+    if isinstance(v, list) and not all(isinstance(item, str) for item in v):
+        return v
+    # Trigger: a str that looks like a JSON array, mirroring parse_tags' detection.
+    # Why: parse_tags recursively parses JSON arrays, stringifying non-string elements
+    #   ('[42]' -> ["42"]) and hiding the type error from Pydantic.
+    # Outcome: malformed arrays pass through unchanged so Pydantic rejects them; valid
+    #   all-string arrays and plain comma strings still delegate to parse_tags.
+    if isinstance(v, str) and v.strip().startswith("[") and v.strip().endswith("]"):
+        try:
+            parsed = json.loads(v)
+        except json.JSONDecodeError:
+            parsed = None
+        if isinstance(parsed, list) and not all(isinstance(item, str) for item in parsed):
+            return v
+    if v is None or isinstance(v, (str, list)):
+        return parse_tags(v)
+    return v
+
+
 def coerce_list(v: Any) -> Any:
     """Coerce string input to list for MCP clients that serialize lists as strings."""
     if v is None:

test-int/mcp/test_edit_note_integration.py

@@ -4,6 +4,8 @@ Integration tests for edit_note MCP tool.
 Tests the complete edit note workflow: MCP client -> MCP server -> FastAPI -> database
 """
 
+from pathlib import Path
+
 import pytest
 from fastmcp import Client
 
@@ -788,3 +790,39 @@ async def test_edit_note_append_autocreate_does_not_fuzzy_match(mcp_server, app,
 
         error_text = edit_result2.content[0].text
         assert "Edit Failed" in error_text
+
+
+@pytest.mark.asyncio
+async def test_edit_note_recovers_file_on_disk_not_indexed(mcp_server, app, test_project):
+    """edit_note should index and edit a markdown file written directly to disk (#581).
+
+    Common flow: a file is written straight to the project directory and edit_note is
+    called before the watcher indexes it. The tool must recover by indexing the single
+    file and retrying resolution instead of failing with "Entity not found".
+    """
+    note_path = Path(test_project.path) / "direct" / "disk-note.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Disk Note\n\nstatus: draft\n", encoding="utf-8")
+
+    async with Client(mcp_server) as client:
+        edit_result = await client.call_tool(
+            "edit_note",
+            {
+                "project": test_project.name,
+                "identifier": "direct/disk-note",
+                "operation": "find_replace",
+                "content": "status: final",
+                "find_text": "status: draft",
+            },
+        )
+
+        edit_text = edit_result.content[0].text
+        assert "Edited note (find_replace)" in edit_text
+
+        read_result = await client.call_tool(
+            "read_note",
+            {"project": test_project.name, "identifier": "direct/disk-note"},
+        )
+        content = read_result.content[0].text
+        assert "status: final" in content
+        assert "status: draft" not in content

test-int/mcp/test_observation_permalink_collision_integration.py

@@ -0,0 +1,141 @@
+"""Regression tests for https://github.com/basicmachines-co/basic-memory/issues/909.
+
+Observation content is truncated to 200 chars when building the synthetic
+permalink (Postgres btree index limit), so distinct observations sharing a
+category and a 200-char content prefix used to collide and the second one was
+silently dropped from the search index, making it unfindable. #931 fixed this
+by appending a content digest to the truncated permalink.
+
+These tests pin the end-to-end behavior independent of the disambiguation
+mechanism: every observation stays searchable, and deleting the note removes
+every index row — including rows whose permalinks needed disambiguation.
+"""
+
+import json
+from typing import Any
+
+import pytest
+from fastmcp import Client
+
+
+def _json_content(tool_result) -> Any:
+    """Parse a FastMCP tool result content block into JSON."""
+    assert len(tool_result.content) == 1
+    assert tool_result.content[0].type == "text"
+    return json.loads(tool_result.content[0].text)  # pyright: ignore [reportAttributeAccessIssue]
+
+
+@pytest.mark.asyncio
+async def test_duplicate_category_content_observations_both_searchable(
+    mcp_server, app, test_project
+):
+    """Both observations must be indexed even when their synthetic permalinks collide."""
+    async with Client(mcp_server) as client:
+        prefix = "x" * 210  # > 200 so the truncated permalink prefixes are identical
+        content = (
+            "# Dup Obs Note\n\n"
+            "## Observations\n"
+            f"- [note] {prefix} ALPHA_UNIQUE_MARKER\n"
+            f"- [note] {prefix} BETA_UNIQUE_MARKER\n"
+        )
+        await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Dup Obs Note",
+                "directory": "dup",
+                "content": content,
+            },
+        )
+
+        # Both observations should be independently findable by their unique suffix
+        for marker in ("ALPHA_UNIQUE_MARKER", "BETA_UNIQUE_MARKER"):
+            result = await client.call_tool(
+                "search_notes",
+                {
+                    "project": test_project.name,
+                    "query": marker,
+                    "search_type": "text",
+                    "entity_types": ["observation"],
+                    "output_format": "json",
+                },
+            )
+            data = _json_content(result)
+            snippets = [r.get("content") or "" for r in data["results"]]
+            assert any(marker in s for s in snippets), (
+                f"observation containing {marker} was dropped from the search index "
+                "due to a synthetic-permalink collision (content truncated to 200 chars). "
+                "results=" + json.dumps(data, default=str)[:800]
+            )
+
+
+@pytest.mark.asyncio
+async def test_delete_note_with_colliding_observations_leaves_no_ghost_rows(
+    mcp_server, app, test_project, search_service
+):
+    """Deleting a note must clean up disambiguated observation index rows.
+
+    search_index has no FK cascade from entity, so any index-time permalink
+    disambiguation must be matched by delete-time cleanup or the extra
+    observation survives in the search index as a ghost row pointing at the
+    deleted file.
+
+    The post-delete assertions inspect the index through ``search_service``
+    rather than the ``search_notes`` tool: the MCP search pipeline happens to
+    hide rows whose entity is gone, which would mask the orphan.
+    """
+    from basic_memory.schemas.search import SearchQuery
+
+    async with Client(mcp_server) as client:
+        prefix = "y" * 210  # > 200 so the truncated permalink prefixes are identical
+        content = (
+            "# Ghost Obs Note\n\n"
+            "## Observations\n"
+            f"- [note] {prefix} GHOST_ALPHA_MARKER\n"
+            f"- [note] {prefix} GHOST_BETA_MARKER\n"
+        )
+        await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Ghost Obs Note",
+                "directory": "ghost",
+                "content": content,
+            },
+        )
+
+        # Both observations are searchable before deletion
+        for marker in ("GHOST_ALPHA_MARKER", "GHOST_BETA_MARKER"):
+            result = await client.call_tool(
+                "search_notes",
+                {
+                    "project": test_project.name,
+                    "query": marker,
+                    "search_type": "text",
+                    "entity_types": ["observation"],
+                    "output_format": "json",
+                },
+            )
+            data = _json_content(result)
+            assert any(marker in (r.get("content") or "") for r in data["results"])
+
+        # Both rows exist in the search index itself under distinct permalinks
+        index_rows = await search_service.search(SearchQuery(text="GHOST_BETA_MARKER"))
+        assert any(r.type == "observation" for r in index_rows)
+
+        delete_result = await client.call_tool(
+            "delete_note",
+            {
+                "project": test_project.name,
+                "identifier": "Ghost Obs Note",
+            },
+        )
+        assert "true" in delete_result.content[0].text.lower()  # pyright: ignore [reportAttributeAccessIssue]
+
+        # No ghost rows remain in the index - including any disambiguated row
+        for marker in ("GHOST_ALPHA_MARKER", "GHOST_BETA_MARKER"):
+            index_rows = await search_service.search(SearchQuery(text=marker))
+            assert index_rows == [], (
+                f"search index row containing {marker} survived note deletion as a ghost: "
+                f"{[(r.type, r.permalink) for r in index_rows]}"
+            )

test-int/mcp/test_search_integration.py

@@ -497,3 +497,55 @@ async def test_search_case_insensitive(mcp_server, app, test_project):
 
             result_text = search_result.content[0].text
             assert "Machine Learning Guide" in result_text, f"Failed for search term: {search_term}"
+
+
+@pytest.mark.asyncio
+async def test_tags_param_vs_tag_query_comma_consistency(mcp_server, app, test_project):
+    """The `tags=` parameter must split comma-separated strings like the `tag:` shorthand.
+
+    Regression test for #910: `search_notes(tags="alpha,beta")` previously coerced the
+    bare string into the single literal tag `["alpha,beta"]` (matching nothing), while
+    the `tag:alpha,beta` query shorthand splits on commas. Both paths must agree.
+    """
+
+    async with Client(mcp_server) as client:
+        # Note tagged alpha + beta
+        await client.call_tool(
+            "write_note",
+            {
+                "project": test_project.name,
+                "title": "Tag Shorthand Note",
+                "directory": "tag-shorthand",
+                "content": "# Tag Shorthand Note\n\nTagShorthandToken body",
+                "tags": ["alpha", "beta"],
+            },
+        )
+
+        # Path A: tag: query shorthand with comma list -> splits, matches
+        via_query = await client.call_tool(
+            "search_notes",
+            {
+                "project": test_project.name,
+                "query": "tag:alpha,beta",
+                "search_type": "text",
+            },
+        )
+        query_hit = "Tag Shorthand Note" in via_query.content[0].text
+
+        # Path B: tags= parameter with the SAME comma string
+        via_param = await client.call_tool(
+            "search_notes",
+            {
+                "project": test_project.name,
+                "query": "TagShorthandToken",
+                "search_type": "text",
+                "tags": "alpha,beta",
+            },
+        )
+        param_hit = "Tag Shorthand Note" in via_param.content[0].text
+
+        assert query_hit, "tag: query shorthand should match (sanity)"
+        assert param_hit == query_hit, (
+            "tags='alpha,beta' param must behave like the tag: shorthand "
+            f"(both split commas). query_hit={query_hit} param_hit={param_hit}"
+        )

tests/api/v2/test_knowledge_router.py

@@ -1,17 +1,23 @@
 """Tests for V2 knowledge graph API routes (ID-based endpoints)."""
 
+import os
 from datetime import datetime, timezone
+from pathlib import Path
 import uuid
 
 import pytest
 from httpx import AsyncClient
 
+from basic_memory.api.v2.routers.knowledge_router import _canonical_file_path
+from basic_memory.ignore_utils import get_bmignore_path
 from basic_memory.models import Entity as EntityModel, Project
 from basic_memory.repository.entity_repository import EntityRepository
 from basic_memory.repository.project_repository import ProjectRepository
+from basic_memory.repository.search_repository_base import VectorSyncBatchResult
 from basic_memory.schemas import DeleteEntitiesResponse
 from basic_memory.schemas.response import DirectoryMoveResult, DirectoryDeleteResult
 from basic_memory.schemas.v2 import EntityResponseV2, EntityResolveResponse
+from basic_memory.services.search_service import SearchService
 
 
 @pytest.mark.asyncio
@@ -955,3 +961,512 @@ async def test_entity_response_includes_user_tracking_fields(client: AsyncClient
     assert "last_updated_by" in body
     assert body["created_by"] is None
     assert body["last_updated_by"] is None
+
+
+## Single-file sync endpoint tests
+
+
+@pytest.mark.asyncio
+async def test_sync_file_indexes_file_on_disk(
+    client: AsyncClient, v2_project_url, test_project: Project
+):
+    """A markdown file written directly to disk becomes resolvable after sync-file (#581)."""
+    note_path = Path(test_project.path) / "incoming" / "disk-note.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Disk Note\n\nWritten directly to disk.\n", encoding="utf-8")
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "incoming/disk-note.md"},
+    )
+    assert response.status_code == 200
+    entity = EntityResponseV2.model_validate(response.json())
+    assert entity.file_path == "incoming/disk-note.md"
+
+    # The file is now resolvable by path, which is what edit_note retries with
+    resolve_response = await client.post(
+        f"{v2_project_url}/knowledge/resolve",
+        json={"identifier": "incoming/disk-note.md", "strict": True},
+    )
+    assert resolve_response.status_code == 200
+    resolved = EntityResolveResponse.model_validate(resolve_response.json())
+    assert resolved.external_id == entity.external_id
+
+
+@pytest.mark.asyncio
+async def test_sync_file_already_indexed_is_idempotent(
+    client: AsyncClient, v2_project_url, test_project: Project
+):
+    """sync-file on an already indexed, unchanged file returns the existing entity."""
+    entity_data = {
+        "title": "AlreadyIndexed",
+        "directory": "test",
+        "content": "Already indexed content",
+    }
+    create_response = await client.post(f"{v2_project_url}/knowledge/entities", json=entity_data)
+    assert create_response.status_code == 200
+    created = EntityResponseV2.model_validate(create_response.json())
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": created.file_path},
+    )
+    assert response.status_code == 200
+    synced = EntityResponseV2.model_validate(response.json())
+    assert synced.external_id == created.external_id
+    assert synced.file_path == created.file_path
+
+
+@pytest.mark.asyncio
+async def test_sync_file_syncs_vectors_when_semantic_enabled(
+    client: AsyncClient,
+    v2_project_url,
+    test_project: Project,
+    app_config,
+    monkeypatch: pytest.MonkeyPatch,
+):
+    """sync-file refreshes semantic vectors for the synced entity.
+
+    Mirrors the inline sync_entity_vectors_batch() pass the project sync flow runs
+    after indexing changed files (SyncService.sync); without it, a note recovered
+    via sync-file stays missing from semantic search until a later edit or full
+    sync. Fixtures run with semantic search disabled, so enable it here and stub
+    the service-level vector batch (like test_search_service.py::test_reindex_vectors
+    stubs the repository batch) to exercise the wiring without the embedding stack.
+    """
+    app_config.semantic_search_enabled = True
+
+    synced_batches: list[list[int]] = []
+
+    async def stub_sync_entity_vectors_batch(
+        self, entity_ids: list[int], progress_callback=None
+    ) -> VectorSyncBatchResult:
+        synced_batches.append(list(entity_ids))
+        return VectorSyncBatchResult(
+            entities_total=len(entity_ids),
+            entities_synced=len(entity_ids),
+            entities_failed=0,
+        )
+
+    # The router builds its SearchService per request, so patch the class method
+    # rather than a fixture instance.
+    monkeypatch.setattr(SearchService, "sync_entity_vectors_batch", stub_sync_entity_vectors_batch)
+
+    note_path = Path(test_project.path) / "incoming" / "semantic-note.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Semantic Note\n\nNeeds vectors after recovery.\n", encoding="utf-8")
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "incoming/semantic-note.md"},
+    )
+    assert response.status_code == 200
+    entity = EntityResponseV2.model_validate(response.json())
+
+    assert synced_batches == [[entity.id]]
+
+
+@pytest.mark.asyncio
+async def test_sync_file_skips_vector_sync_when_semantic_disabled(
+    client: AsyncClient,
+    v2_project_url,
+    test_project: Project,
+    app_config,
+    monkeypatch: pytest.MonkeyPatch,
+):
+    """sync-file does not touch the vector pipeline when semantic search is disabled."""
+    assert app_config.semantic_search_enabled is False
+
+    synced_batches: list[list[int]] = []
+
+    async def stub_sync_entity_vectors_batch(
+        self, entity_ids: list[int], progress_callback=None
+    ) -> VectorSyncBatchResult:
+        synced_batches.append(list(entity_ids))
+        return VectorSyncBatchResult(
+            entities_total=len(entity_ids),
+            entities_synced=len(entity_ids),
+            entities_failed=0,
+        )
+
+    monkeypatch.setattr(SearchService, "sync_entity_vectors_batch", stub_sync_entity_vectors_batch)
+
+    note_path = Path(test_project.path) / "incoming" / "plain-note.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Plain Note\n\nNo vectors needed.\n", encoding="utf-8")
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "incoming/plain-note.md"},
+    )
+    assert response.status_code == 200
+    assert synced_batches == []
+
+
+@pytest.mark.asyncio
+async def test_sync_file_missing_file_returns_404(client: AsyncClient, v2_project_url):
+    """sync-file fails fast when the file does not exist on disk."""
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "missing/never-written.md"},
+    )
+    assert response.status_code == 404
+    assert "File not found on disk" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_sync_file_rejects_path_traversal(client: AsyncClient, v2_project_url):
+    """sync-file rejects paths that escape the project root."""
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "../outside-project.md"},
+    )
+    assert response.status_code == 400
+    assert "project boundaries" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_sync_file_rejects_symlink_escape(
+    client: AsyncClient, v2_project_url, test_project: Project, entity_repository
+):
+    """sync-file rejects paths whose canonical target escapes the project via symlink.
+
+    The exact-cased request ('link/secret.md') is rejected by the pre-canonicalization
+    boundary check: the path exists, so resolve() follows the symlink and detects the
+    escape. The wrong-cased request ('LINK/secret.md') is the regression case — on a
+    case-sensitive filesystem that path does not exist, the pre-check resolves it
+    lexically and passes; canonicalization then matches the real 'link' segment but
+    stops at the project boundary and reports the path as not found (404). On
+    case-insensitive filesystems the pre-check catches both with 400.
+    """
+    project_path = Path(test_project.path)
+    outside_dir = project_path.parent / "sync-file-outside"
+    outside_dir.mkdir(parents=True, exist_ok=True)
+    (outside_dir / "secret.md").write_text(
+        "# Outside\n\nMust never be indexed.\n", encoding="utf-8"
+    )
+    (project_path / "link").symlink_to(outside_dir, target_is_directory=True)
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "link/secret.md"},
+    )
+    assert response.status_code == 400
+    assert "project boundaries" in response.json()["detail"]
+
+    # 400 (pre-check, case-insensitive FS) or 404 (canonicalization stops at the
+    # boundary, case-sensitive FS) — either way the escape is rejected.
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "LINK/secret.md"},
+    )
+    assert response.status_code in (400, 404)
+
+    # Nothing outside the project root was indexed
+    assert await entity_repository.find_all() == []
+
+
+@pytest.mark.asyncio
+async def test_sync_file_symlink_escape_never_scans_outside_directory(
+    client: AsyncClient,
+    v2_project_url,
+    test_project: Project,
+    entity_repository,
+    monkeypatch: pytest.MonkeyPatch,
+):
+    """Canonicalization never scans directories outside the project root.
+
+    Rejecting the request is not enough: before this fix, the wrong-cased request
+    ('LINK/secret.md') passed the pre-check on a case-sensitive filesystem, and
+    canonicalization followed the escaping 'link' symlink with os.scandir before the
+    post-canonicalization containment check fired — an information touch outside the
+    boundary. We spy on os.scandir and assert the outside directory is never scanned.
+    The spy is what makes this test meaningful on case-insensitive macOS too, where
+    the request is already rejected by the pre-check: the assertion proves no layer
+    scanned past the boundary either way.
+    """
+    project_path = Path(test_project.path)
+    outside_dir = (project_path.parent / "sync-file-outside-scan").resolve()
+    outside_dir.mkdir(parents=True, exist_ok=True)
+    (outside_dir / "secret.md").write_text(
+        "# Outside\n\nMust never be scanned.\n", encoding="utf-8"
+    )
+    (project_path / "link").symlink_to(outside_dir, target_is_directory=True)
+
+    real_scandir = os.scandir
+    scanned: list[Path] = []
+
+    def recording_scandir(path=".", *args, **kwargs):
+        # Record the resolved path so a scandir on the symlinked 'link' directory
+        # shows up as the outside directory it actually reads.
+        scanned.append(Path(path).resolve())
+        return real_scandir(path, *args, **kwargs)
+
+    monkeypatch.setattr(os, "scandir", recording_scandir)
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "LINK/secret.md"},
+    )
+    assert response.status_code in (400, 404)
+    assert outside_dir not in scanned
+
+    assert await entity_repository.find_all() == []
+
+
+@pytest.mark.asyncio
+async def test_sync_file_symlink_escape_never_probes_outside_target(
+    client: AsyncClient,
+    v2_project_url,
+    test_project: Project,
+    entity_repository,
+    monkeypatch: pytest.MonkeyPatch,
+):
+    """The containment check runs before any filesystem probe that follows symlinks.
+
+    Regression: a wrong-cased request ('SECRET.md') canonicalizes onto an in-project
+    FILE symlink ('secret.md' -> outside target) on a case-sensitive filesystem.
+    Before the fix, the endpoint's is_file() existence probe ran before the
+    resolved-containment check, so it followed the symlink and stat'ed the target
+    outside the project boundary (and the response flipped between 404 and 400
+    depending on whether the external target existed). We spy on Path.is_file and
+    assert no probe ever resolves to the outside target; the escape is always
+    rejected with 400 — by the pre-check on case-insensitive filesystems, by the
+    post-canonicalization containment check on case-sensitive ones.
+    """
+    project_path = Path(test_project.path)
+    outside_dir = (project_path.parent / "sync-file-outside-probe").resolve()
+    outside_dir.mkdir(parents=True, exist_ok=True)
+    outside_target = outside_dir / "secret.md"
+    outside_target.write_text("# Outside\n\nMust never be probed.\n", encoding="utf-8")
+    (project_path / "secret.md").symlink_to(outside_target)
+
+    real_is_file = Path.is_file
+    probed: list[Path] = []
+
+    def recording_is_file(self, *args, **kwargs):
+        # Record the resolved path so a probe on the in-project symlink name shows
+        # up as the outside target it would actually stat.
+        probed.append(self.resolve())
+        return real_is_file(self, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "is_file", recording_is_file)
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "SECRET.md"},
+    )
+    assert response.status_code == 400
+    assert "project boundaries" in response.json()["detail"]
+    assert outside_target not in probed
+
+    assert await entity_repository.find_all() == []
+
+
+def test_canonical_file_path_stops_at_project_boundary(tmp_path: Path):
+    """_canonical_file_path bails before descending past the resolved project home.
+
+    Exercised directly (not via the endpoint) so the boundary bail is covered on
+    case-insensitive filesystems too, where the endpoint pre-check rejects the
+    request before canonicalization runs.
+    """
+    home = tmp_path / "project"
+    home.mkdir()
+    outside_dir = tmp_path / "outside"
+    outside_dir.mkdir()
+    (outside_dir / "secret.md").write_text("# Outside\n", encoding="utf-8")
+    (home / "link").symlink_to(outside_dir, target_is_directory=True)
+
+    assert _canonical_file_path(home, ["link", "secret.md"]) is None
+
+    # Symlinks that stay inside the project keep canonicalizing as before.
+    real_dir = home / "real"
+    real_dir.mkdir()
+    (real_dir / "inside.md").write_text("# Inside\n", encoding="utf-8")
+    (home / "alias").symlink_to(real_dir, target_is_directory=True)
+
+    assert _canonical_file_path(home, ["alias", "inside.md"]) == "alias/inside.md"
+
+
+@pytest.mark.asyncio
+async def test_sync_file_symlink_inside_project_still_indexes(
+    client: AsyncClient, v2_project_url, test_project: Project
+):
+    """A symlinked directory that stays inside the project is still accepted.
+
+    Pre-existing behavior we preserve: the containment check follows the symlink,
+    sees the resolved target inside the project root, and indexes the entity under
+    the requested (symlinked) path — only escapes outside the root are rejected.
+    """
+    project_path = Path(test_project.path)
+    real_dir = project_path / "real"
+    real_dir.mkdir(parents=True, exist_ok=True)
+    (real_dir / "inside.md").write_text("# Inside\n\nReachable via alias.\n", encoding="utf-8")
+    (project_path / "alias").symlink_to(real_dir, target_is_directory=True)
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "alias/inside.md"},
+    )
+    assert response.status_code == 200
+    entity = EntityResponseV2.model_validate(response.json())
+    assert entity.file_path == "alias/inside.md"
+
+
+@pytest.mark.asyncio
+async def test_sync_file_wrong_cased_path_does_not_create_duplicate(
+    client: AsyncClient, v2_project_url, test_project: Project, entity_repository
+):
+    """A wrong-cased path resolves to the canonical on-disk file without duplicating it.
+
+    On case-insensitive filesystems (macOS/Windows) a wrong-cased path passes existence
+    checks; without canonicalization the indexer would insert a second entity keyed by
+    the wrong-cased path. The endpoint matches real directory entries, so the request
+    behaves identically on case-sensitive and case-insensitive filesystems.
+    """
+    note_path = Path(test_project.path) / "notes" / "disk-note.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Disk Note\n\nWritten directly to disk.\n", encoding="utf-8")
+
+    first = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "notes/disk-note.md"},
+    )
+    assert first.status_code == 200
+    canonical = EntityResponseV2.model_validate(first.json())
+    assert canonical.file_path == "notes/disk-note.md"
+
+    second = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "notes/Disk-Note.md"},
+    )
+    assert second.status_code == 200
+    synced = EntityResponseV2.model_validate(second.json())
+    assert synced.file_path == "notes/disk-note.md"
+    assert synced.external_id == canonical.external_id
+
+    entities = await entity_repository.find_all()
+    assert [entity.file_path for entity in entities] == ["notes/disk-note.md"]
+
+
+@pytest.mark.asyncio
+async def test_sync_file_rejects_non_normalized_segments(
+    client: AsyncClient, v2_project_url, test_project: Project
+):
+    """sync-file rejects './' and '//' style segments instead of indexing them verbatim."""
+    note_path = Path(test_project.path) / "notes" / "disk-note.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Disk Note\n", encoding="utf-8")
+
+    for non_normalized in ("./notes/disk-note.md", "notes//disk-note.md"):
+        response = await client.post(
+            f"{v2_project_url}/knowledge/sync-file",
+            json={"file_path": non_normalized},
+        )
+        assert response.status_code == 400
+        assert "not normalized" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_sync_file_directory_returns_404(
+    client: AsyncClient, v2_project_url, test_project: Project
+):
+    """sync-file refuses a path that canonicalizes to a directory instead of a file."""
+    (Path(test_project.path) / "just-a-directory").mkdir(parents=True, exist_ok=True)
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "just-a-directory"},
+    )
+    assert response.status_code == 404
+    assert "File not found on disk" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_sync_file_path_through_file_returns_404(
+    client: AsyncClient, v2_project_url, test_project: Project
+):
+    """sync-file fails fast when a parent segment resolves to a file, not a directory."""
+    note_path = Path(test_project.path) / "notes" / "disk-note.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Disk Note\n", encoding="utf-8")
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "notes/disk-note.md/child.md"},
+    )
+    assert response.status_code == 404
+    assert "File not found on disk" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_sync_file_rejects_hidden_file(
+    client: AsyncClient, v2_project_url, test_project: Project
+):
+    """sync-file refuses hidden files, matching the default '.*' ignore pattern."""
+    hidden_path = Path(test_project.path) / ".secrets.md"
+    hidden_path.write_text("# Hidden\n\nShould never be indexed.\n", encoding="utf-8")
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": ".secrets.md"},
+    )
+    assert response.status_code == 400
+    assert "ignore rules" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_sync_file_rejects_gitignored_file(
+    client: AsyncClient, v2_project_url, test_project: Project
+):
+    """sync-file honors the project .gitignore, matching scan/watch filtering."""
+    project_path = Path(test_project.path)
+    (project_path / ".gitignore").write_text("private/\n", encoding="utf-8")
+    note_path = project_path / "private" / "secret.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Secret\n\nGitignored content.\n", encoding="utf-8")
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "private/secret.md"},
+    )
+    assert response.status_code == 400
+    assert "ignore rules" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_sync_file_rejects_bmignored_file(
+    client: AsyncClient, v2_project_url, test_project: Project
+):
+    """sync-file honors user .bmignore patterns, matching scan/watch filtering."""
+    bmignore_path = get_bmignore_path()
+    bmignore_path.parent.mkdir(parents=True, exist_ok=True)
+    bmignore_path.write_text("drafts-wip\n", encoding="utf-8")
+
+    note_path = Path(test_project.path) / "drafts-wip" / "scratch.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Scratch\n\nBmignored content.\n", encoding="utf-8")
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "drafts-wip/scratch.md"},
+    )
+    assert response.status_code == 400
+    assert "ignore rules" in response.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_sync_file_rejects_non_markdown(
+    client: AsyncClient, v2_project_url, test_project: Project
+):
+    """sync-file only indexes markdown notes."""
+    file_path = Path(test_project.path) / "data" / "records.csv"
+    file_path.parent.mkdir(parents=True, exist_ok=True)
+    file_path.write_text("a,b,c\n", encoding="utf-8")
+
+    response = await client.post(
+        f"{v2_project_url}/knowledge/sync-file",
+        json={"file_path": "data/records.csv"},
+    )
+    assert response.status_code == 400
+    assert "Only markdown files" in response.json()["detail"]

tests/ci/test_bm_bossbot_workflow.py

@@ -1,138 +0,0 @@
-import re
-from pathlib import Path
-
-import yaml
-
-
-WORKFLOW_PATH = Path(".github/workflows/bm-bossbot.yml")
-
-
-def _workflow() -> dict:
-    return yaml.safe_load(WORKFLOW_PATH.read_text(encoding="utf-8"))
-
-
-def test_bm_bossbot_runs_after_successful_tests_workflow() -> None:
-    workflow = _workflow()
-    review_job = workflow["jobs"]["review"]
-
-    assert workflow["name"] == "BM Bossbot"
-    assert "pull_request_target" not in workflow["on"]
-    assert workflow["on"]["workflow_run"]["workflows"] == ["Tests"]
-    assert workflow["on"]["workflow_run"]["types"] == ["completed"]
-    assert "workflow_dispatch" in workflow["on"]
-    assert "github.event.workflow_run.conclusion == 'success'" in review_job["if"]
-    assert "github.event.workflow_run.pull_requests[0].number != ''" in review_job["if"]
-    assert review_job["outputs"]["should_review"] == "${{ steps.pr.outputs.should_review }}"
-
-    permissions = workflow["permissions"]
-    assert permissions["contents"] == "read"
-    assert permissions["pull-requests"] == "write"
-    assert permissions["statuses"] == "write"
-
-    assert "assets" not in workflow["jobs"]
-
-
-def test_bm_bossbot_workflow_never_checks_out_untrusted_head() -> None:
-    workflow = _workflow()
-    checkout_steps = [
-        step
-        for job in workflow["jobs"].values()
-        for step in job["steps"]
-        if step.get("uses") == "actions/checkout@v6"
-    ]
-
-    assert checkout_steps
-    for checkout_step in checkout_steps:
-        assert checkout_step["with"]["ref"] == "${{ github.event.repository.default_branch }}"
-        assert "${{ github.event.pull_request.head.sha }}" not in str(checkout_step)
-    workflow_text = WORKFLOW_PATH.read_text(encoding="utf-8")
-    # The danger is consuming UNTRUSTED PR data: checking out the PR head, or
-    # interpolating attacker-controlled strings (title/body/branch names) into
-    # run scripts. The numeric PR id is safe and the recheck job needs it, so
-    # allow exactly `.number` and nothing else from the pull_request payload.
-    pr_event_fields = set(re.findall(r"github\.event\.pull_request\.[a-zA-Z_.]+", workflow_text))
-    assert pr_event_fields <= {"github.event.pull_request.number"}
-    assert "github.event.pull_request.head" not in workflow_text
-    assert "cancel-in-progress: true" in workflow_text
-
-
-def test_bm_bossbot_workflow_has_deterministic_status_steps() -> None:
-    workflow = _workflow()
-    steps = workflow["jobs"]["review"]["steps"]
-    names = [step["name"] for step in steps]
-
-    assert "Set up uv" in names
-    assert "Mark BM Bossbot approval pending" in names
-    assert "Finalize BM Bossbot approval" in names
-    # The gate is deterministic: no LLM review step and no image generation.
-    assert "Run BM Bossbot review with Codex" not in names
-    assert "Collect sanitized PR context" not in names
-
-    workflow_text = WORKFLOW_PATH.read_text(encoding="utf-8")
-    assert "openai/codex-action" not in workflow_text
-    assert "OPENAI_API_KEY" not in workflow_text
-
-    pending = next(step for step in steps if step["name"] == "Mark BM Bossbot approval pending")
-    assert pending["if"] == "steps.pr.outputs.should_review == 'true'"
-    finalize = next(step for step in steps if step["name"] == "Finalize BM Bossbot approval")
-    assert finalize["if"] == "always() && steps.pr.outputs.should_review == 'true'"
-    assert '--trusted "${{ steps.trust.outputs.trusted_author }}"' in finalize["run"]
-    assert "BM Bossbot Approval" in workflow_text
-    assert "uv run --script scripts/bm_bossbot_status.py pending" in workflow_text
-    assert "uv run --script scripts/bm_bossbot_status.py finalize" in workflow_text
-
-
-def test_bm_bossbot_rejects_stale_successful_test_runs_before_finalize() -> None:
-    workflow_text = WORKFLOW_PATH.read_text(encoding="utf-8")
-    workflow = _workflow()
-    steps = workflow["jobs"]["review"]["steps"]
-    normalize = next(step for step in steps if step["name"] == "Normalize PR event")
-    classify = next(step for step in steps if step["name"] == "Classify PR author")
-
-    assert "tested_sha" in normalize["run"]
-    assert "current_head_sha" in normalize["run"]
-    assert "actions/workflows/test.yml/runs" in normalize["run"]
-    assert "-f event=push" in normalize["run"]
-    assert "-f event=pull_request" not in normalize["run"]
-    assert '-f head_sha="${current_head_sha}"' in normalize["run"]
-    assert 'select(.conclusion == "success")' in normalize["run"]
-    assert "no successful Tests workflow for ${current_head_sha}" in workflow_text
-    stale_sha_guard = '[ -n "${tested_sha}" ] && [ "${tested_sha}" != "${current_head_sha}" ]'
-    assert stale_sha_guard in normalize["run"]
-    assert "should_review=false" in normalize["run"]
-    assert (
-        "Tests passed for ${tested_sha}, but current head is ${current_head_sha}" in workflow_text
-    )
-    assert classify["if"] == "steps.pr.outputs.should_review == 'true'"
-
-
-def test_bm_bossbot_has_no_image_generation() -> None:
-    """The per-PR image job was removed: it spent OpenAI tokens on every run."""
-    workflow = _workflow()
-    workflow_text = WORKFLOW_PATH.read_text(encoding="utf-8")
-
-    assert "assets" not in workflow["jobs"]
-    assert "generate_pr_infographic" not in workflow_text
-    assert "pr-assets/" not in workflow_text
-
-
-def test_bm_bossbot_classifies_authors_and_gates_untrusted_deterministically() -> None:
-    workflow = _workflow()
-    steps = workflow["jobs"]["review"]["steps"]
-
-    classify = next(step for step in steps if step["name"] == "Classify PR author")
-    finalize = next(step for step in steps if step["name"] == "Finalize BM Bossbot approval")
-
-    assert "OWNER|MEMBER|COLLABORATOR" in classify["run"]
-    assert classify["if"] == "steps.pr.outputs.should_review == 'true'"
-    assert '--trusted "${{ steps.trust.outputs.trusted_author }}"' in finalize["run"]
-
-
-def test_claude_code_review_is_manual_advisory_only() -> None:
-    workflow = yaml.safe_load(
-        Path(".github/workflows/claude-code-review.yml").read_text(encoding="utf-8")
-    )
-
-    assert "pull_request" not in workflow["on"]
-    assert "workflow_dispatch" in workflow["on"]
-    assert workflow["on"]["workflow_dispatch"]["inputs"]["pr_number"]["required"] is True

tests/ci/test_project_updates.py

@@ -1,718 +0,0 @@
-import json
-from pathlib import Path
-
-import pytest
-import yaml
-from pydantic import ValidationError
-
-from basic_memory.ci.project_updates import (
-    AgentSynthesis,
-    ProjectUpdateConfig,
-    ProjectUpdateContext,
-    build_project_update_note,
-    collect_project_update_context,
-    detect_github_repo,
-    load_project_update_config,
-    parse_github_remote,
-    render_agent_synthesis_schema,
-    render_capture_prompt,
-    render_soul_template,
-    render_workflow,
-    schema_seed_specs,
-)
-from basic_memory.ci import project_updates
-
-
-def _write_json(path: Path, payload: dict) -> Path:
-    path.write_text(json.dumps(payload), encoding="utf-8")
-    return path
-
-
-def _pr_payload(*, merged: bool = True) -> dict:
-    return {
-        "action": "closed",
-        "repository": {
-            "full_name": "basicmachines-co/basic-memory",
-            "html_url": "https://github.com/basicmachines-co/basic-memory",
-        },
-        "pull_request": {
-            "number": 123,
-            "title": "Remember project updates",
-            "body": "Adds Auto BM capture.\n\nCloses #77",
-            "html_url": "https://github.com/basicmachines-co/basic-memory/pull/123",
-            "merged": merged,
-            "merged_at": "2026-06-04T18:42:00Z" if merged else None,
-            "merge_commit_sha": "abc123",
-            "changed_files": 4,
-            "labels": [{"name": "feature"}, {"name": "ci"}],
-            "user": {"login": "octocat"},
-        },
-    }
-
-
-def _synthesis_payload(**overrides: object) -> dict[str, object]:
-    payload: dict[str, object] = {
-        "summary": "Auto BM now records project updates.",
-        "story": (
-            "GitHub delivery events were losing their useful narrative after merge. "
-            "Auto BM collects source facts, lets the agent explain the change, and "
-            "publishes the result as durable project memory."
-        ),
-        "problem_addressed": "Project delivery context was not preserved after GitHub events.",
-        "solution": "Collect GitHub facts and publish an idempotent Basic Memory note.",
-        "system_impact": "Future humans and agents can recover the delivery narrative.",
-        "why_it_matters": "Future agents can recover project context.",
-        "components_changed": ["basic_memory.ci.project_updates"],
-        "complexity_introduced": [],
-        "refactors_or_removals": [],
-        "user_facing_changes": [],
-        "internal_changes": [],
-        "verification": [],
-        "follow_ups": [],
-        "decision_candidates": [],
-        "task_candidates": [],
-    }
-    payload.update(overrides)
-    return payload
-
-
-def test_collect_merged_pull_request_context(tmp_path: Path) -> None:
-    event_path = _write_json(tmp_path / "event.json", _pr_payload())
-
-    context = collect_project_update_context(
-        event_name="pull_request",
-        event_path=event_path,
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-
-    assert context.eligible is True
-    assert context.source_event == "pull_request_merged"
-    assert context.repo == "basicmachines-co/basic-memory"
-    assert context.idempotency_key == "github:basicmachines-co/basic-memory:pull_request_merged:123"
-    assert context.pr_number == 123
-    assert context.sha == "abc123"
-    assert context.labels == ["feature", "ci"]
-    assert context.linked_issues == ["#77"]
-    assert context.source_url == "https://github.com/basicmachines-co/basic-memory/pull/123"
-
-
-def test_collect_enriches_pull_request_context_from_github_api(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    def fake_github_api_get(path: str, token: str) -> list[dict] | dict:
-        assert token == "github-token"
-        if path.startswith("/repos/basicmachines-co/basic-memory/pulls/123/files"):
-            return [
-                {
-                    "filename": "src/basic_memory/ci/project_updates.py",
-                    "status": "modified",
-                    "additions": 42,
-                    "deletions": 7,
-                    "changes": 49,
-                }
-            ]
-        if path.startswith("/repos/basicmachines-co/basic-memory/pulls/123/commits"):
-            return [
-                {
-                    "sha": "abc123def456",
-                    "commit": {
-                        "message": "fix ci synthesis schema\n\nRequire all fields.",
-                        "author": {"name": "Pat"},
-                    },
-                }
-            ]
-        if path == "/repos/basicmachines-co/basic-memory/issues/77":
-            return {
-                "number": 77,
-                "title": "Codex structured output rejects optional schema fields",
-                "body": "Auto BM failed before publish when optional fields were omitted.",
-                "html_url": "https://github.com/basicmachines-co/basic-memory/issues/77",
-                "state": "closed",
-            }
-        raise AssertionError(f"unexpected GitHub API path: {path}")
-
-    monkeypatch.setenv("GITHUB_TOKEN", "github-token")
-    monkeypatch.setattr(project_updates, "_github_api_get", fake_github_api_get, raising=False)
-    event_path = _write_json(tmp_path / "event.json", _pr_payload())
-
-    context = collect_project_update_context(
-        event_name="pull_request",
-        event_path=event_path,
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-
-    assert context.changed_files[0].filename == "src/basic_memory/ci/project_updates.py"
-    assert context.changed_files[0].status == "modified"
-    assert context.commits[0].message == "fix ci synthesis schema\n\nRequire all fields."
-    assert context.linked_issue_details[0].number == 77
-    assert (
-        context.linked_issue_details[0].title
-        == "Codex structured output rejects optional schema fields"
-    )
-
-
-def test_github_api_get_list_fetches_multiple_pages(monkeypatch: pytest.MonkeyPatch) -> None:
-    calls: list[str] = []
-
-    def fake_github_api_get(path: str, token: str) -> list[dict]:
-        assert token == "github-token"
-        calls.append(path)
-        if path.endswith("page=1"):
-            return [{"filename": f"file-{index}.py"} for index in range(100)]
-        if path.endswith("page=2"):
-            return [{"filename": "file-100.py"}]
-        raise AssertionError(f"unexpected GitHub API path: {path}")
-
-    monkeypatch.setattr(project_updates, "_github_api_get", fake_github_api_get, raising=False)
-
-    files = project_updates._github_api_get_list(
-        "/repos/basicmachines-co/basic-memory/pulls/123/files",
-        "github-token",
-    )
-
-    assert len(files) == 101
-    assert calls == [
-        "/repos/basicmachines-co/basic-memory/pulls/123/files?per_page=100&page=1",
-        "/repos/basicmachines-co/basic-memory/pulls/123/files?per_page=100&page=2",
-    ]
-
-
-def test_collect_handles_sparse_pull_request_payload(tmp_path: Path) -> None:
-    payload = {
-        "action": "closed",
-        "repository": {},
-        "pull_request": {
-            "number": 123,
-            "merged": True,
-            "labels": "not-a-list",
-        },
-    }
-    event_path = _write_json(tmp_path / "event.json", payload)
-
-    context = collect_project_update_context(
-        event_name="pull_request",
-        event_path=event_path,
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-
-    assert context.eligible is True
-    assert context.repo is None
-    assert context.repo_url is None
-    assert context.labels == []
-    assert context.linked_issues == []
-
-
-def test_collect_handles_missing_repository_payload(tmp_path: Path) -> None:
-    payload = {
-        "action": "closed",
-        "pull_request": {
-            "number": 123,
-            "merged": True,
-        },
-    }
-    event_path = _write_json(tmp_path / "event.json", payload)
-
-    context = collect_project_update_context(
-        event_name="pull_request",
-        event_path=event_path,
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-
-    assert context.eligible is True
-    assert context.repo is None
-    assert context.repo_url is None
-
-
-def test_collect_rejects_missing_payload_shapes(tmp_path: Path) -> None:
-    pr_context = collect_project_update_context(
-        event_name="pull_request",
-        event_path=_write_json(tmp_path / "pr.json", {"action": "closed"}),
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-    workflow_context = collect_project_update_context(
-        event_name="workflow_run",
-        event_path=_write_json(tmp_path / "workflow.json", {"action": "completed"}),
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-
-    assert pr_context.eligible is False
-    assert pr_context.skip_reason == "pull request payload missing"
-    assert workflow_context.eligible is False
-    assert workflow_context.skip_reason == "workflow run payload missing"
-
-
-def test_collect_ignores_non_closed_pull_request_action(tmp_path: Path) -> None:
-    payload = _pr_payload()
-    payload["action"] = "opened"
-    event_path = _write_json(tmp_path / "event.json", payload)
-
-    context = collect_project_update_context(
-        event_name="pull_request",
-        event_path=event_path,
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-
-    assert context.eligible is False
-    assert context.skip_reason == "pull request action was not closed"
-
-
-def test_collect_ignores_closed_unmerged_pull_request(tmp_path: Path) -> None:
-    event_path = _write_json(tmp_path / "event.json", _pr_payload(merged=False))
-
-    context = collect_project_update_context(
-        event_name="pull_request",
-        event_path=event_path,
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-
-    assert context.eligible is False
-    assert context.skip_reason == "pull request was closed without merging"
-
-
-def test_collect_successful_configured_production_deploy(tmp_path: Path) -> None:
-    payload = {
-        "action": "completed",
-        "repository": {
-            "full_name": "basicmachines-co/basic-memory-cloud",
-            "html_url": "https://github.com/basicmachines-co/basic-memory-cloud",
-        },
-        "workflow_run": {
-            "id": 98765,
-            "name": "Deploy Production",
-            "conclusion": "success",
-            "html_url": "https://github.com/basicmachines-co/basic-memory-cloud/actions/runs/98765",
-            "head_sha": "def456",
-            "updated_at": "2026-06-04T19:10:00Z",
-        },
-    }
-    event_path = _write_json(tmp_path / "event.json", payload)
-
-    context = collect_project_update_context(
-        event_name="workflow_run",
-        event_path=event_path,
-        config=ProjectUpdateConfig(
-            project="cloud-memory",
-            deploy_workflows=["Deploy Production"],
-            production_environments=["production"],
-        ),
-    )
-
-    assert context.eligible is True
-    assert context.source_event == "production_deploy_succeeded"
-    assert context.workflow_run_id == "98765"
-    assert context.environment == "production"
-    assert context.idempotency_key == (
-        "github:basicmachines-co/basic-memory-cloud:production_deploy_succeeded:production:98765"
-    )
-
-
-def test_collect_ignores_failed_or_unconfigured_deploy(tmp_path: Path) -> None:
-    payload = {
-        "action": "completed",
-        "repository": {"full_name": "basicmachines-co/basic-memory"},
-        "workflow_run": {"id": 1, "name": "Tests", "conclusion": "failure"},
-    }
-    event_path = _write_json(tmp_path / "event.json", payload)
-
-    context = collect_project_update_context(
-        event_name="workflow_run",
-        event_path=event_path,
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-
-    assert context.eligible is False
-    assert context.skip_reason == "workflow conclusion was failure"
-
-
-def test_collect_ignores_successful_unconfigured_deploy(tmp_path: Path) -> None:
-    payload = {
-        "action": "completed",
-        "repository": {"full_name": "basicmachines-co/basic-memory"},
-        "workflow_run": {"id": 1, "name": "Tests", "conclusion": "success"},
-    }
-    event_path = _write_json(tmp_path / "event.json", payload)
-
-    context = collect_project_update_context(
-        event_name="workflow_run",
-        event_path=event_path,
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-
-    assert context.eligible is False
-    assert context.skip_reason == "workflow 'Tests' is not configured for project updates"
-
-
-def test_collect_ignores_unsupported_event(tmp_path: Path) -> None:
-    event_path = _write_json(tmp_path / "event.json", {})
-
-    context = collect_project_update_context(
-        event_name="push",
-        event_path=event_path,
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-
-    assert context.eligible is False
-    assert context.skip_reason == "unsupported GitHub event: push"
-
-
-def test_collect_rejects_missing_or_invalid_event_payload(tmp_path: Path) -> None:
-    with pytest.raises(ValueError, match="not found"):
-        collect_project_update_context(
-            event_name="pull_request",
-            event_path=tmp_path / "missing.json",
-            config=ProjectUpdateConfig(project="team-memory"),
-        )
-
-    invalid_json = tmp_path / "invalid.json"
-    invalid_json.write_text("{", encoding="utf-8")
-    with pytest.raises(ValueError, match="not valid JSON"):
-        collect_project_update_context(
-            event_name="pull_request",
-            event_path=invalid_json,
-            config=ProjectUpdateConfig(project="team-memory"),
-        )
-
-    list_json = tmp_path / "list.json"
-    list_json.write_text("[]", encoding="utf-8")
-    with pytest.raises(ValueError, match="JSON object"):
-        collect_project_update_context(
-            event_name="pull_request",
-            event_path=list_json,
-            config=ProjectUpdateConfig(project="team-memory"),
-        )
-
-
-def test_build_project_update_note_uses_deterministic_identity_fields(tmp_path: Path) -> None:
-    event_path = _write_json(tmp_path / "event.json", _pr_payload())
-    context = collect_project_update_context(
-        event_name="pull_request",
-        event_path=event_path,
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-    synthesis = AgentSynthesis.model_validate(
-        _synthesis_payload(
-            why_it_matters="Future agents can recover the delivery narrative.",
-            repo="evil/repo",
-            source_event="production_deploy_succeeded",
-            verification=["Unit tests cover event normalization."],
-        )
-    )
-
-    note = build_project_update_note(context=context, synthesis=synthesis)
-
-    assert note.title == "PR #123: Remember project updates"
-    assert note.directory == "project-updates/github/basicmachines-co/basic-memory"
-    assert note.metadata["repo"] == "basicmachines-co/basic-memory"
-    assert note.metadata["source_event"] == "pull_request_merged"
-    assert note.metadata["idempotency_key"] == context.idempotency_key
-    assert "evil/repo" not in note.content
-
-
-def test_build_project_update_note_renders_story_sections(tmp_path: Path) -> None:
-    event_path = _write_json(tmp_path / "event.json", _pr_payload())
-    context = collect_project_update_context(
-        event_name="pull_request",
-        event_path=event_path,
-        config=ProjectUpdateConfig(project="team-memory"),
-    )
-    synthesis = AgentSynthesis.model_validate(
-        {
-            "summary": "Auto BM now publishes durable project updates.",
-            "story": (
-                "Auto BM needed to preserve the delivery narrative, not just the mechanics. "
-                "The change adds a CI handoff where Codex synthesizes context and bm publishes it."
-            ),
-            "problem_addressed": "Project context was lost after meaningful GitHub delivery events.",
-            "solution": "Collect GitHub facts, let Codex synthesize intent, then publish idempotently.",
-            "system_impact": "Merges now leave durable memory for future humans and agents.",
-            "why_it_matters": "Future work can recover why the delivery happened.",
-            "components_changed": [
-                "basic_memory.ci.project_updates",
-                "basic_memory.cli.commands.ci",
-            ],
-            "complexity_introduced": ["Adds a CI-only agent synthesis boundary."],
-            "refactors_or_removals": ["Keeps Basic Memory auth out of the agent step."],
-            "verification": ["Unit tests cover collect and publish behavior."],
-        }
-    )
-
-    note = build_project_update_note(context=context, synthesis=synthesis)
-
-    assert "## Story" in note.content
-    assert "## Problem Addressed" in note.content
-    assert "## How The Change Solves It" in note.content
-    assert "## Impact On The System" in note.content
-    assert "## Project Memory" in note.content
-    assert "## Why It Matters" not in note.content
-    assert "## Components Changed" in note.content
-    assert "basic_memory.ci.project_updates" in note.content
-    assert "## Complexity Introduced" in note.content
-    assert "## Refactors Or Removals" in note.content
-
-
-def test_build_project_update_note_renders_linked_issue_details_as_links() -> None:
-    context = ProjectUpdateContext(
-        eligible=True,
-        source_event="pull_request_merged",
-        repo="basicmachines-co/basic-memory",
-        repo_url="https://github.com/basicmachines-co/basic-memory",
-        source_url="https://github.com/basicmachines-co/basic-memory/pull/123",
-        idempotency_key="github:basicmachines-co/basic-memory:pull_request_merged:123",
-        pr_number=123,
-        title="Remember project updates",
-        linked_issues=["#77", "#88"],
-        linked_issue_details=[
-            project_updates.LinkedIssueDetail(
-                number=77,
-                title="Codex structured output rejects optional schema fields",
-                state="closed",
-                url="https://github.com/basicmachines-co/basic-memory/issues/77",
-            )
-        ],
-    )
-    synthesis = AgentSynthesis.model_validate(_synthesis_payload())
-
-    note = build_project_update_note(context=context, synthesis=synthesis)
-
-    assert (
-        "- Linked issue: [#77 Codex structured output rejects optional schema fields "
-        "(closed)](https://github.com/basicmachines-co/basic-memory/issues/77)" in note.content
-    )
-    assert (
-        "- Linked issue: [#88](https://github.com/basicmachines-co/basic-memory/issues/88)"
-        in note.content
-    )
-    assert "- Linked issues: #77, #88" not in note.content
-
-
-def test_build_project_update_note_for_production_deploy(tmp_path: Path) -> None:
-    payload = {
-        "action": "completed",
-        "repository": {
-            "full_name": "basicmachines-co/basic-memory-cloud",
-            "html_url": "https://github.com/basicmachines-co/basic-memory-cloud",
-        },
-        "workflow_run": {
-            "id": 98765,
-            "name": "Deploy Production",
-            "conclusion": "success",
-            "html_url": "https://github.com/basicmachines-co/basic-memory-cloud/actions/runs/98765",
-            "head_sha": "def456",
-            "updated_at": "2026-06-04T19:10:00Z",
-        },
-    }
-    context = collect_project_update_context(
-        event_name="workflow_run",
-        event_path=_write_json(tmp_path / "event.json", payload),
-        config=ProjectUpdateConfig(
-            project="cloud-memory",
-            deploy_workflows=["Deploy Production"],
-            production_environments=["production"],
-        ),
-    )
-    synthesis = AgentSynthesis.model_validate(
-        _synthesis_payload(
-            summary="Production deploy completed.",
-            story=(
-                "A configured production workflow completed successfully. "
-                "The deploy SHA is now recorded as durable project memory."
-            ),
-            problem_addressed="Production delivery needed a durable deployment record.",
-            solution="Publish a project update for the successful workflow run.",
-            system_impact="The production deploy is connected to its workflow run and SHA.",
-            why_it_matters="The latest project update reached users.",
-        )
-    )
-
-    note = build_project_update_note(context=context, synthesis=synthesis)
-
-    assert note.title == "Production deploy: 2026-06-04"
-    assert note.metadata["workflow_run_id"] == "98765"
-    assert note.metadata["environment"] == "production"
-    assert "https://github.com/basicmachines-co/basic-memory-cloud/actions/runs/98765" in (
-        note.content
-    )
-
-
-def test_build_project_update_note_rejects_invalid_context() -> None:
-    synthesis = AgentSynthesis.model_validate(
-        _synthesis_payload(
-            summary="Auto BM records project updates.",
-            why_it_matters="Future agents can recover context.",
-        )
-    )
-    with pytest.raises(ValueError, match="ineligible"):
-        build_project_update_note(
-            context=ProjectUpdateContext(eligible=False, skip_reason="not useful"),
-            synthesis=synthesis,
-        )
-
-    with pytest.raises(ValueError, match="deterministic identity"):
-        build_project_update_note(
-            context=ProjectUpdateContext(
-                eligible=True,
-                source_event="pull_request_merged",
-                repo="basicmachines-co/basic-memory",
-            ),
-            synthesis=synthesis,
-        )
-
-
-def test_agent_synthesis_requires_summary_and_why_it_matters() -> None:
-    missing_why = _synthesis_payload()
-    missing_why.pop("why_it_matters")
-    with pytest.raises(ValidationError):
-        AgentSynthesis.model_validate(missing_why)
-
-    with pytest.raises(ValidationError):
-        AgentSynthesis.model_validate(_synthesis_payload(summary=" "))
-
-
-def test_agent_synthesis_requires_delivery_narrative_fields() -> None:
-    with pytest.raises(ValidationError):
-        AgentSynthesis.model_validate(
-            {
-                "summary": "Auto BM records project updates.",
-                "why_it_matters": "Future agents can recover context.",
-            }
-        )
-
-
-def test_project_update_config_requires_non_empty_lists() -> None:
-    with pytest.raises(ValueError, match="at least one"):
-        ProjectUpdateConfig(deploy_workflows=[" "])
-
-
-def test_render_workflow_invokes_codex_read_only_without_basic_memory_secret() -> None:
-    workflow = render_workflow(
-        ProjectUpdateConfig(
-            project="team-memory",
-            deploy_workflows=["Deploy Production"],
-            production_environments=["production"],
-        )
-    )
-
-    assert "openai/codex-action@v1" in workflow
-    assert "sandbox: read-only" in workflow
-    assert "output-schema-file: ${{ runner.temp }}/agent-synthesis.schema.json" in workflow
-    assert "BASIC_MEMORY_CLOUD_API_KEY: ${{ secrets.BASIC_MEMORY_API_KEY }}" in workflow
-    assert "BASIC_MEMORY_CLOUD_HOST: ${{ vars.BASIC_MEMORY_CLOUD_HOST || '' }}" not in workflow
-    assert "BASIC_MEMORY_CI_CLOUD_HOST: ${{ vars.BASIC_MEMORY_CLOUD_HOST }}" in workflow
-    assert 'if [ -n "$BASIC_MEMORY_CI_CLOUD_HOST" ]' in workflow
-    assert "--context .github/basic-memory/project-update-context.json" in workflow
-    assert "GITHUB_TOKEN: ${{ github.token }}" in workflow
-    assert "--cloud \\" in workflow
-    codex_step = workflow.split("- name: Synthesize project update with Codex", 1)[1].split(
-        "- name: Publish project update", 1
-    )[0]
-    assert "BASIC_MEMORY_API_KEY" not in codex_step
-
-
-def test_render_workflow_outputs_valid_github_actions_yaml() -> None:
-    workflow = render_workflow(ProjectUpdateConfig(project="team-memory"))
-
-    parsed = yaml.safe_load(workflow)
-
-    assert isinstance(parsed, dict)
-    assert parsed["on"]["pull_request"]["types"] == ["closed"]
-    assert parsed["on"]["workflow_run"]["types"] == ["completed"]
-
-
-def test_render_capture_prompt_uses_workspace_context_path() -> None:
-    prompt = render_capture_prompt()
-
-    assert ".github/basic-memory/project-update-context.json" in prompt
-    assert ".github/basic-memory/SOUL.md" in prompt
-    assert "${{ runner.temp }}" not in prompt
-    assert "Do not write a fill-in-the-blanks note" in prompt
-    assert "Read the PR diff before writing" in prompt
-    assert "problem -> solution -> impact" in prompt
-    assert "It is okay to say when the code is messy" in prompt
-    assert "Ground all judgments" in prompt
-
-
-def test_render_soul_template_guides_personality_without_overriding_facts() -> None:
-    soul = render_soul_template()
-
-    assert soul.startswith("# Auto BM Soul")
-    assert "It is okay to say when code is messy" in soul
-    assert "Notice good simplifications" in soul
-    assert "Do not invent intent, impact, tests, or drama" in soul
-    assert "Keep personality in service of memory" in soul
-
-
-def test_render_agent_synthesis_schema_is_ci_guardrail_not_domain_schema() -> None:
-    schema = json.loads(render_agent_synthesis_schema())
-
-    assert schema["title"] == "AgentSynthesis"
-    assert "summary" in schema["required"]
-    assert "story" in schema["required"]
-    assert "problem_addressed" in schema["required"]
-    assert "solution" in schema["required"]
-    assert "system_impact" in schema["required"]
-    assert "components_changed" in schema["required"]
-    assert "why_it_matters" in schema["required"]
-    assert set(schema["required"]) == set(schema["properties"])
-    assert "project_update" not in json.dumps(schema)
-
-
-def test_schema_seed_specs_are_basic_memory_schema_notes() -> None:
-    specs = schema_seed_specs()
-
-    assert {spec.entity for spec in specs} == {
-        "ProjectUpdate",
-        "GitHubPullRequestUpdate",
-        "GitHubProductionDeployUpdate",
-    }
-    assert all(spec.metadata["type"] == "schema" for spec in specs)
-    assert all(spec.metadata["settings"]["validation"] == "warn" for spec in specs)
-    project_update = next(spec for spec in specs if spec.entity == "ProjectUpdate")
-    assert "story" in project_update.metadata["schema"]
-    assert "problem_addressed" in project_update.metadata["schema"]
-
-
-def test_parse_github_remote_accepts_https_and_ssh() -> None:
-    assert parse_github_remote("https://github.com/basicmachines-co/basic-memory.git") == (
-        "basicmachines-co",
-        "basic-memory",
-    )
-    assert parse_github_remote("git@github.com:basicmachines-co/basic-memory.git") == (
-        "basicmachines-co",
-        "basic-memory",
-    )
-
-
-def test_parse_github_remote_rejects_non_github_remote() -> None:
-    with pytest.raises(ValueError, match="GitHub remote"):
-        parse_github_remote("https://example.com/basicmachines-co/basic-memory.git")
-
-
-def test_detect_github_repo_requires_origin_remote(tmp_path: Path) -> None:
-    with pytest.raises(ValueError, match="No remote.origin.url"):
-        detect_github_repo(tmp_path)
-
-
-def test_load_project_update_config_handles_missing_and_invalid_yaml(tmp_path: Path) -> None:
-    assert load_project_update_config(tmp_path / "missing.yml") == ProjectUpdateConfig()
-
-    invalid = tmp_path / "invalid.yml"
-    invalid.write_text("- not\n- an\n- object\n", encoding="utf-8")
-    with pytest.raises(ValueError, match="YAML object"):
-        load_project_update_config(invalid)
-
-
-def test_private_note_helpers_reject_invalid_repo_shape() -> None:
-    context = ProjectUpdateContext(eligible=True, repo="not-owner-repo")
-    with pytest.raises(ValueError, match="owner/repo"):
-        project_updates._note_directory(context, ProjectUpdateConfig(project="team-memory"))
-
-    missing_repo = ProjectUpdateContext(eligible=True)
-    with pytest.raises(ValueError, match="missing repo"):
-        project_updates._note_directory(missing_repo, ProjectUpdateConfig(project="team-memory"))
-
-
-def test_private_note_title_uses_generic_fallback_for_unknown_event() -> None:
-    context = ProjectUpdateContext(eligible=True, source_event="unknown")
-
-    assert project_updates._note_title(context) == "Project update"

tests/ci/test_testmon_cache.py

@@ -1,114 +0,0 @@
-from pathlib import Path
-
-import pytest
-
-from scripts import testmon_cache
-
-
-def _write_testmon_file(directory: Path, filename: str, content: str) -> Path:
-    directory.mkdir(parents=True, exist_ok=True)
-    path = directory / filename
-    path.write_text(content, encoding="utf-8")
-    return path
-
-
-def test_seed_testmon_data_reports_missing_shared_cache(tmp_path: Path) -> None:
-    repo_root = tmp_path / "repo"
-    cache_dir = tmp_path / "cache"
-    repo_root.mkdir()
-
-    result = testmon_cache.seed_testmon_data(repo_root=repo_root, cache_dir=cache_dir)
-
-    assert result.status == "missing"
-    assert result.copied == ()
-    assert not (repo_root / ".testmondata").exists()
-
-
-def test_seed_testmon_data_keeps_existing_local_data(tmp_path: Path) -> None:
-    repo_root = tmp_path / "repo"
-    cache_dir = tmp_path / "cache"
-    local_datafile = _write_testmon_file(repo_root, ".testmondata", "local")
-    _write_testmon_file(cache_dir, ".testmondata", "shared")
-
-    result = testmon_cache.seed_testmon_data(repo_root=repo_root, cache_dir=cache_dir)
-
-    assert result.status == "exists"
-    assert result.copied == ()
-    assert local_datafile.read_text(encoding="utf-8") == "local"
-
-
-def test_seed_testmon_data_replaces_stale_sidecars(tmp_path: Path) -> None:
-    repo_root = tmp_path / "repo"
-    cache_dir = tmp_path / "cache"
-    _write_testmon_file(repo_root, ".testmondata-shm", "stale sidecar")
-    _write_testmon_file(cache_dir, ".testmondata", "shared main")
-    _write_testmon_file(cache_dir, ".testmondata-wal", "shared wal")
-
-    result = testmon_cache.seed_testmon_data(repo_root=repo_root, cache_dir=cache_dir)
-
-    assert result.status == "seeded"
-    assert {path.name for path in result.copied} == {".testmondata", ".testmondata-wal"}
-    assert (repo_root / ".testmondata").read_text(encoding="utf-8") == "shared main"
-    assert (repo_root / ".testmondata-wal").read_text(encoding="utf-8") == "shared wal"
-    assert not (repo_root / ".testmondata-shm").exists()
-
-
-def test_refresh_testmon_data_requires_local_data(tmp_path: Path) -> None:
-    repo_root = tmp_path / "repo"
-    cache_dir = tmp_path / "cache"
-    repo_root.mkdir()
-
-    with pytest.raises(FileNotFoundError):
-        testmon_cache.refresh_testmon_data(repo_root=repo_root, cache_dir=cache_dir)
-
-
-def test_refresh_testmon_data_replaces_shared_cache(tmp_path: Path) -> None:
-    repo_root = tmp_path / "repo"
-    cache_dir = tmp_path / "cache"
-    _write_testmon_file(repo_root, ".testmondata", "local main")
-    _write_testmon_file(repo_root, ".testmondata-shm", "local shm")
-    _write_testmon_file(cache_dir, ".testmondata", "old main")
-    _write_testmon_file(cache_dir, ".testmondata-wal", "old wal")
-
-    result = testmon_cache.refresh_testmon_data(repo_root=repo_root, cache_dir=cache_dir)
-
-    assert result.status == "refreshed"
-    assert {path.name for path in result.copied} == {".testmondata", ".testmondata-shm"}
-    assert (cache_dir / ".testmondata").read_text(encoding="utf-8") == "local main"
-    assert (cache_dir / ".testmondata-shm").read_text(encoding="utf-8") == "local shm"
-    assert not (cache_dir / ".testmondata-wal").exists()
-
-
-def test_resolve_cache_dir_prefers_explicit_path_over_env(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    repo_root = tmp_path / "repo"
-    env_cache_dir = tmp_path / "env-cache"
-    explicit_cache_dir = tmp_path / "explicit-cache"
-    repo_root.mkdir()
-    monkeypatch.setenv(testmon_cache.TESTMON_CACHE_ENV, str(env_cache_dir))
-
-    assert testmon_cache.resolve_cache_dir(repo_root) == env_cache_dir.resolve()
-    assert (
-        testmon_cache.resolve_cache_dir(repo_root, explicit_cache_dir)
-        == explicit_cache_dir.resolve()
-    )
-
-
-def test_status_command_prints_local_and_shared_paths(
-    tmp_path: Path, capsys: pytest.CaptureFixture[str]
-) -> None:
-    repo_root = tmp_path / "repo"
-    cache_dir = tmp_path / "cache"
-    repo_root.mkdir()
-    _write_testmon_file(repo_root, ".testmondata", "local main")
-
-    exit_code = testmon_cache.main(
-        ["--repo-root", str(repo_root), "--cache-dir", str(cache_dir), "status"]
-    )
-
-    assert exit_code == 0
-    output = capsys.readouterr().out
-    assert f"Repo root:      {repo_root.resolve()}" in output
-    assert "Worktree ready: True" in output
-    assert "Cache ready:    False" in output

tests/ci/test_validate_skills.py

@@ -1,92 +0,0 @@
-from pathlib import Path
-
-import pytest
-
-from scripts.validate_skills import parse_frontmatter
-
-
-def test_parse_frontmatter_rejects_unquoted_mapping_colon(tmp_path: Path) -> None:
-    skill = tmp_path / "SKILL.md"
-    skill.write_text(
-        "\n".join(
-            [
-                "---",
-                "name: bm-qa",
-                "description: Use when validating fixes. Drives the full loop: map issue to commit.",
-                "---",
-                "# Skill",
-                "",
-            ]
-        ),
-        encoding="utf-8",
-    )
-
-    with pytest.raises(SystemExit, match="invalid YAML"):
-        parse_frontmatter(skill)
-
-
-def test_parse_frontmatter_allows_url_colons_in_plain_values(tmp_path: Path) -> None:
-    skill = tmp_path / "SKILL.md"
-    skill.write_text(
-        "\n".join(
-            [
-                "---",
-                "name: memory-notes",
-                "description: See https://docs.basicmemory.com for usage.",
-                "---",
-                "# Skill",
-                "",
-            ]
-        ),
-        encoding="utf-8",
-    )
-
-    frontmatter = parse_frontmatter(skill)
-
-    assert frontmatter["description"] == "See https://docs.basicmemory.com for usage."
-
-
-def test_parse_frontmatter_strips_matching_single_quotes(tmp_path: Path) -> None:
-    skill = tmp_path / "SKILL.md"
-    skill.write_text(
-        "\n".join(
-            [
-                "---",
-                "name: memory-notes",
-                "description: 'Use when values contain mapping-like text: safely.'",
-                "---",
-                "# Skill",
-                "",
-            ]
-        ),
-        encoding="utf-8",
-    )
-
-    frontmatter = parse_frontmatter(skill)
-
-    assert frontmatter["description"] == "Use when values contain mapping-like text: safely."
-
-
-def test_parse_frontmatter_keeps_nested_fields_nested(tmp_path: Path) -> None:
-    schema = tmp_path / "schema.md"
-    schema.write_text(
-        "\n".join(
-            [
-                "---",
-                "type: schema",
-                "entity: Task",
-                "schema:",
-                "  type: object",
-                "---",
-                "# Task",
-                "",
-            ]
-        ),
-        encoding="utf-8",
-    )
-
-    frontmatter = parse_frontmatter(schema)
-
-    assert frontmatter["type"] == "schema"
-    assert frontmatter["entity"] == "Task"
-    assert frontmatter["schema"] == ""

tests/cli/cloud/test_project_sync_command.py

@@ -15,6 +15,21 @@ from basic_memory.schemas.cloud import WorkspaceInfo
 runner = CliRunner()
 
 
+@pytest.mark.parametrize(
+    "command",
+    ["sync", "bisync", "check", "bisync-reset"],
+)
+def test_cloud_mirror_command_help_marks_personal_workspaces_only(command):
+    """Mirror-family help must say Personal-only and point Team users at push/pull (#851)."""
+    result = runner.invoke(app, ["cloud", command, "--help"])
+
+    assert result.exit_code == 0, result.output
+    output = " ".join(result.output.split())
+    assert "Personal workspaces only" in output
+    assert "bm cloud pull" in output
+    assert "bm cloud push" in output
+
+
 @pytest.mark.parametrize(
     "argv",
     [

tests/cli/test_cli_tool_rich_output.py

@@ -0,0 +1,991 @@
+"""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/mcp/clients/test_clients.py

@@ -133,6 +133,36 @@ class TestKnowledgeClient:
         result = await client.resolve_entity("my-note")
         assert result == "entity-uuid-123"
 
+    @pytest.mark.asyncio
+    async def test_sync_file(self, monkeypatch):
+        """Test sync_file posts the file path to the sync-file endpoint."""
+        from basic_memory.mcp.clients import knowledge as knowledge_mod
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "permalink": "notes/disk-note",
+            "title": "Disk Note",
+            "file_path": "notes/disk-note.md",
+            "note_type": "note",
+            "content_type": "text/markdown",
+            "observations": [],
+            "relations": [],
+            "created_at": "2024-01-01T00:00:00",
+            "updated_at": "2024-01-01T00:00:00",
+        }
+
+        async def mock_call_post(client, url, **kwargs):
+            assert "/v2/projects/proj-123/knowledge/sync-file" in url
+            assert kwargs.get("json") == {"file_path": "notes/disk-note.md"}
+            return mock_response
+
+        monkeypatch.setattr(knowledge_mod, "call_post", mock_call_post)
+
+        mock_http = MagicMock()
+        client = KnowledgeClient(mock_http, "proj-123")
+        result = await client.sync_file("notes/disk-note.md")
+        assert result.file_path == "notes/disk-note.md"
+
     @pytest.mark.asyncio
     async def test_get_orphans_validates_response(self, monkeypatch):
         """Orphan responses are validated into GraphNode objects."""

tests/mcp/test_tool_edit_note.py

@@ -1,10 +1,14 @@
 """Tests for the edit_note MCP tool."""
 
+from pathlib import Path
 from unittest.mock import patch
 
+import httpx
 import pytest
+from mcp.server.fastmcp.exceptions import ToolError
 
-from basic_memory.mcp.tools.edit_note import edit_note
+from basic_memory.mcp.clients import KnowledgeClient
+from basic_memory.mcp.tools.edit_note import _resolve_after_disk_recovery, edit_note
 from basic_memory.mcp.tools.read_note import read_note
 from basic_memory.mcp.tools.write_note import write_note
 
@@ -1263,3 +1267,264 @@ async def test_edit_note_skips_detection_when_project_id_provided(
     assert isinstance(result, str)
     assert "Edited note (append)" in result
     assert f"project: {test_project.name}" in result
+
+
+@pytest.mark.asyncio
+async def test_edit_note_find_replace_recovers_file_on_disk_not_indexed(client, test_project):
+    """find_replace should index and edit a file written directly to disk (#581)."""
+    note_path = Path(test_project.path) / "notes" / "disk-note.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Disk Note\n\nstatus: draft\n", encoding="utf-8")
+
+    result = await edit_note(
+        project=test_project.name,
+        identifier="notes/disk-note",
+        operation="find_replace",
+        content="status: final",
+        find_text="status: draft",
+    )
+
+    assert isinstance(result, str)
+    assert "Edited note (find_replace)" in result
+    assert "status: final" in note_path.read_text(encoding="utf-8")
+
+
+@pytest.mark.asyncio
+async def test_edit_note_recovers_identifier_with_md_extension(client, test_project):
+    """An identifier already ending in .md should recover via the exact file path (#581)."""
+    note_path = Path(test_project.path) / "notes" / "exact-path.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Exact Path\n\nversion: v1\n", encoding="utf-8")
+
+    result = await edit_note(
+        project=test_project.name,
+        identifier="notes/exact-path.md",
+        operation="find_replace",
+        content="version: v2",
+        find_text="version: v1",
+    )
+
+    assert isinstance(result, str)
+    assert "Edited note (find_replace)" in result
+    assert "version: v2" in note_path.read_text(encoding="utf-8")
+
+
+@pytest.mark.asyncio
+async def test_edit_note_recovers_identifier_with_markdown_extension(client, test_project):
+    """A .markdown identifier must recover via the exact path, not '<path>.markdown.md' (#581)."""
+    note_path = Path(test_project.path) / "notes" / "alt-suffix.markdown"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Alt Suffix\n\nstate: pending\n", encoding="utf-8")
+
+    result = await edit_note(
+        project=test_project.name,
+        identifier="notes/alt-suffix.markdown",
+        operation="find_replace",
+        content="state: done",
+        find_text="state: pending",
+    )
+
+    assert isinstance(result, str)
+    assert "Edited note (find_replace)" in result
+    assert "state: done" in note_path.read_text(encoding="utf-8")
+
+
+@pytest.mark.asyncio
+async def test_edit_note_refuses_ignored_on_disk_file(client, test_project):
+    """An on-disk file matched by .gitignore must be refused, not shadowed by auto-create."""
+    project_path = Path(test_project.path)
+    (project_path / ".gitignore").write_text("private/\n", encoding="utf-8")
+    note_path = project_path / "private" / "secret.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    original_content = "# Secret\n\nGitignored content.\n"
+    note_path.write_text(original_content, encoding="utf-8")
+
+    result = await edit_note(
+        project=test_project.name,
+        identifier="private/secret",
+        operation="append",
+        content="\nShould never be written.",
+    )
+
+    assert isinstance(result, str)
+    assert "ignore rules" in result
+    assert "will not be edited" in result
+    assert "Edited note" not in result
+    assert "Created note" not in result
+
+    # The ignored file is untouched and auto-create did not shadow it with a new entity
+    assert note_path.read_text(encoding="utf-8") == original_content
+    assert [entry.name for entry in (project_path / "private").iterdir()] == ["secret.md"]
+
+
+@pytest.mark.asyncio
+async def test_edit_note_append_recovers_file_on_disk_instead_of_autocreate(client, test_project):
+    """append to an unindexed on-disk file should edit it, not auto-create a replacement (#581)."""
+    note_path = Path(test_project.path) / "notes" / "disk-append.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Disk Append\n\nOriginal disk content.\n", encoding="utf-8")
+
+    result = await edit_note(
+        project=test_project.name,
+        identifier="notes/disk-append",
+        operation="append",
+        content="\nAppended line.",
+    )
+
+    assert isinstance(result, str)
+    assert "Edited note (append)" in result
+    assert "Created note" not in result
+
+    final_content = note_path.read_text(encoding="utf-8")
+    assert "Original disk content." in final_content
+    assert "Appended line." in final_content
+
+
+@pytest.mark.asyncio
+async def test_edit_note_append_recovers_markdown_suffix_file_from_stem(client, test_project):
+    """A stem identifier for an on-disk .markdown file edits it, not auto-creates .md (#581).
+
+    Recovery probes the identifier as-is, then '.md', then '.markdown'; without the
+    '.markdown' probe, append would auto-create 'notes/alt-stem.md' next to the real
+    file instead of editing it.
+    """
+    note_path = Path(test_project.path) / "notes" / "alt-stem.markdown"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Alt Stem\n\nOriginal markdown-suffix content.\n", encoding="utf-8")
+
+    result = await edit_note(
+        project=test_project.name,
+        identifier="notes/alt-stem",
+        operation="append",
+        content="\nAppended line.",
+    )
+
+    assert isinstance(result, str)
+    assert "Edited note (append)" in result
+    assert "Created note" not in result
+
+    final_content = note_path.read_text(encoding="utf-8")
+    assert "Original markdown-suffix content." in final_content
+    assert "Appended line." in final_content
+    # The real file was edited in place; no shadow .md entity was created beside it
+    assert [entry.name for entry in note_path.parent.iterdir()] == ["alt-stem.markdown"]
+
+
+@pytest.mark.asyncio
+async def test_edit_note_append_recovers_wrong_cased_identifier(client, test_project):
+    """A wrong-cased identifier edits the canonical on-disk file after recovery (#581).
+
+    The sync-file endpoint canonicalizes casing by matching real directory entries,
+    so syncing 'notes/Disk-Note.md' indexes 'notes/disk-note.md' identically on
+    case-sensitive (CI) and case-insensitive (macOS) filesystems — no filesystem
+    probe is needed here. The regression: the retry used to strictly re-resolve the
+    raw wrong-cased identifier, which can miss the just-indexed canonical entity;
+    the fix returns the entity identity straight from the sync-file response.
+    """
+    note_path = Path(test_project.path) / "notes" / "disk-note.md"
+    note_path.parent.mkdir(parents=True, exist_ok=True)
+    note_path.write_text("# Disk Note\n\nOriginal cased content.\n", encoding="utf-8")
+
+    result = await edit_note(
+        project=test_project.name,
+        identifier="notes/Disk-Note",
+        operation="append",
+        content="\nAppended line.",
+    )
+
+    assert isinstance(result, str)
+    assert "Edited note (append)" in result
+    assert "Created note" not in result
+
+    final_content = note_path.read_text(encoding="utf-8")
+    assert "Original cased content." in final_content
+    assert "Appended line." in final_content
+    # The canonical file was edited; no wrong-cased duplicate was created beside it
+    assert [entry.name for entry in note_path.parent.iterdir()] == ["disk-note.md"]
+
+
+@pytest.mark.asyncio
+async def test_resolve_after_disk_recovery_falls_back_to_strict_resolve():
+    """Older servers that omit external_id from sync-file trigger a strict re-resolve.
+
+    The recovery path prefers the entity identity from the sync-file response; when a
+    server predates that field, the only safe option is a strict re-resolve of the
+    raw identifier (which fails loudly on a miss instead of guessing).
+    """
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("/sync-file"):
+            return httpx.Response(
+                200,
+                json={
+                    "permalink": "notes/old-server-note",
+                    "title": "Old Server Note",
+                    "file_path": "notes/old-server-note.md",
+                    "note_type": "note",
+                    "content_type": "text/markdown",
+                    "observations": [],
+                    "relations": [],
+                    "created_at": "2024-01-01T00:00:00",
+                    "updated_at": "2024-01-01T00:00:00",
+                },
+            )
+        assert request.url.path.endswith("/resolve")
+        return httpx.Response(200, json={"external_id": "resolved-entity-uuid"})
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as http_client:
+        knowledge_client = KnowledgeClient(http_client, "project-external-id")
+        result = await _resolve_after_disk_recovery(knowledge_client, "notes/old-server-note")
+
+    assert result == "resolved-entity-uuid"
+
+
+@pytest.mark.asyncio
+async def test_resolve_after_disk_recovery_propagates_unexpected_errors():
+    """Server-side failures during disk recovery must not be masked as a not-found miss.
+
+    Only 400/404 sync-file rejections mean "nothing to recover"; a 500 (or auth
+    failure) would otherwise be swallowed and edit_note would continue into
+    auto-create with a misleading not-found error.
+    """
+
+    def server_error(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(500, json={"detail": "boom"})
+
+    transport = httpx.MockTransport(server_error)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as http_client:
+        knowledge_client = KnowledgeClient(http_client, "project-external-id")
+        with pytest.raises(ToolError, match="boom"):
+            await _resolve_after_disk_recovery(knowledge_client, "notes/unlucky-note")
+
+
+@pytest.mark.asyncio
+async def test_edit_note_append_traversal_identifier_is_blocked(client, test_project):
+    """A traversal identifier must be rejected by both disk recovery and auto-create."""
+    result = await edit_note(
+        project=test_project.name,
+        identifier="../escape-note",
+        operation="append",
+        content="should never be written",
+    )
+
+    assert isinstance(result, str)
+    assert "# Error" in result
+    assert "paths must stay within project boundaries" in result
+    assert not (Path(test_project.path).parent / "escape-note.md").exists()
+
+
+@pytest.mark.asyncio
+async def test_edit_note_append_traversal_identifier_json_error(client, test_project):
+    """JSON mode reports a structured security error for traversal identifiers."""
+    result = await edit_note(
+        project=test_project.name,
+        identifier="../escape-json-note",
+        operation="append",
+        content="should never be written",
+        output_format="json",
+    )
+
+    assert isinstance(result, dict)
+    assert result["error"] == "SECURITY_VALIDATION_ERROR"
+    assert result["fileCreated"] is False

tests/mcp/test_tool_search.py

@@ -1694,6 +1694,175 @@ async def test_search_notes_tags_comma_string_filters_via_mcp(mcp, client, test_
         assert not await found("gamma")
 
 
+def test_search_notes_tags_annotation_rejects_non_string_types():
+    """Unsupported tag types must fail validation, not be stringified (#932 follow-up).
+
+    Bare parse_tags coerces anything to strings (42 -> ["42"], {"a": 1} -> junk tags),
+    silently turning caller mistakes into no-result searches. The strict_search_tags
+    wrapper only normalizes str/list/None and lets Pydantic reject everything else.
+    """
+    from pydantic import ValidationError
+
+    annotation = inspect.signature(search_notes).parameters["tags"].annotation
+    adapter = TypeAdapter(annotation)
+
+    with pytest.raises(ValidationError):
+        adapter.validate_python(42)
+    with pytest.raises(ValidationError):
+        adapter.validate_python({"a": 1})
+
+    # Lists with non-string elements must also fail, not be stringified ([42] -> ["42"]).
+    with pytest.raises(ValidationError):
+        adapter.validate_python([42])
+    with pytest.raises(ValidationError):
+        adapter.validate_python([{"a": 1}])
+    with pytest.raises(ValidationError):
+        adapter.validate_python(["ok", 42])
+
+    # JSON-array strings with non-string elements must fail the same way — parse_tags
+    # would otherwise recursively stringify them before Pydantic validates List[str].
+    with pytest.raises(ValidationError):
+        adapter.validate_python("[42]")
+    with pytest.raises(ValidationError):
+        adapter.validate_python('[{"a": 1}]')
+    with pytest.raises(ValidationError):
+        adapter.validate_python('["ok", 42]')
+
+    # All-string lists and all-string JSON-array strings remain valid.
+    assert adapter.validate_python(["a", "b"]) == ["a", "b"]
+    assert adapter.validate_python('["a","b"]') == ["a", "b"]
+
+    # None stays a valid "no filter" input.
+    assert adapter.validate_python(None) in (None, [])
+
+
+@pytest.mark.asyncio
+async def test_search_notes_tags_invalid_type_rejected_via_mcp(mcp, client, test_project):
+    """tags=42 through the real MCP layer must raise a validation error (#932 follow-up)."""
+    from fastmcp import Client
+    from fastmcp.exceptions import ToolError
+
+    async with Client(mcp) as mcp_client:
+        with pytest.raises(ToolError):
+            await mcp_client.call_tool(
+                "search_notes",
+                {
+                    "project": test_project.name,
+                    "query": "anything",
+                    "tags": 42,
+                },
+            )
+        with pytest.raises(ToolError):
+            await mcp_client.call_tool(
+                "search_notes",
+                {
+                    "project": test_project.name,
+                    "query": "anything",
+                    "tags": {"a": 1},
+                },
+            )
+        # Lists with non-string elements must be rejected too, not stringified.
+        with pytest.raises(ToolError):
+            await mcp_client.call_tool(
+                "search_notes",
+                {
+                    "project": test_project.name,
+                    "query": "anything",
+                    "tags": [42],
+                },
+            )
+        with pytest.raises(ToolError):
+            await mcp_client.call_tool(
+                "search_notes",
+                {
+                    "project": test_project.name,
+                    "query": "anything",
+                    "tags": [{"a": 1}],
+                },
+            )
+        with pytest.raises(ToolError):
+            await mcp_client.call_tool(
+                "search_notes",
+                {
+                    "project": test_project.name,
+                    "query": "anything",
+                    "tags": ["ok", 42],
+                },
+            )
+        # JSON-array strings with non-string elements (clients that serialize arrays as
+        # strings) must be rejected too, not recursively stringified by parse_tags.
+        with pytest.raises(ToolError):
+            await mcp_client.call_tool(
+                "search_notes",
+                {
+                    "project": test_project.name,
+                    "query": "anything",
+                    "tags": "[42]",
+                },
+            )
+        with pytest.raises(ToolError):
+            await mcp_client.call_tool(
+                "search_notes",
+                {
+                    "project": test_project.name,
+                    "query": "anything",
+                    "tags": '[{"a": 1}]',
+                },
+            )
+        with pytest.raises(ToolError):
+            await mcp_client.call_tool(
+                "search_notes",
+                {
+                    "project": test_project.name,
+                    "query": "anything",
+                    "tags": '["ok", 42]',
+                },
+            )
+        # Sanity: a valid all-string JSON-array string is still accepted.
+        await mcp_client.call_tool(
+            "search_notes",
+            {
+                "project": test_project.name,
+                "query": "anything",
+                "tags": '["a","b"]',
+            },
+        )
+
+
+@pytest.mark.asyncio
+async def test_search_notes_direct_call_splits_comma_tags(client, test_project):
+    """Direct callers bypass the BeforeValidator, so the body must normalize tags.
+
+    Regression for the CLI path: `bm tool search-notes --tag alpha,beta` calls this
+    function directly with Typer's collected list ["alpha,beta"], which must split
+    into ["alpha", "beta"] instead of matching nothing (#910, #932 follow-up).
+    """
+    await write_note(
+        project=test_project.name,
+        title="Direct Tag Split Note",
+        directory="test",
+        content="# Direct Tag Split Note\nDirectTagToken body",
+        tags=["alpha", "beta"],
+    )
+
+    async def found(tags_value: list[str] | None) -> bool:
+        result = await search_notes(
+            project=test_project.name,
+            query="DirectTagToken",
+            search_type="text",
+            output_format="json",
+            tags=tags_value,
+        )
+        assert isinstance(result, dict), f"search failed: {result}"
+        return any(r["title"] == "Direct Tag Split Note" for r in result["results"])
+
+    assert await found(["alpha"]), "plain tag list must match (sanity)"
+    # The CLI regression: Typer collects --tag alpha,beta as the single element "alpha,beta".
+    assert await found(["alpha,beta"])
+    # Negative control: the filter is still applied.
+    assert not await found(["gamma"])
+
+
 # --- Tests for text output format (#641) -----------------------------------
 
 

tests/scripts/test_bm_bossbot_status.py

@@ -1,434 +0,0 @@
-import json
-from pathlib import Path
-from typing import Any, Mapping
-
-import pytest
-from typer.testing import CliRunner
-
-from scripts import bm_bossbot_status
-
-
-def _event_payload(body: str = "Event snapshot body") -> dict[str, object]:
-    return {
-        "repository": {"full_name": "basicmachines-co/basic-memory"},
-        "pull_request": {
-            "number": 925,
-            "body": body,
-            "head": {"sha": "abc123"},
-        },
-    }
-
-
-def test_status_script_is_uv_typer_entrypoint() -> None:
-    source = bm_bossbot_status.__file__
-    assert source is not None
-    text = open(source, encoding="utf-8").read()
-
-    assert text.startswith("#!/usr/bin/env -S uv run --script\n")
-    assert "# /// script" in text
-    assert "typer" in text
-    assert hasattr(bm_bossbot_status, "app")
-
-
-def test_status_payload_uses_required_context() -> None:
-    payload = bm_bossbot_status.build_status_payload(
-        state="pending",
-        description="BM Bossbot is reviewing this head SHA",
-        target_url="https://github.com/basicmachines-co/basic-memory/actions/runs/1",
-    )
-
-    assert payload == {
-        "state": "pending",
-        "context": "BM Bossbot Approval",
-        "description": "BM Bossbot is reviewing this head SHA",
-        "target_url": "https://github.com/basicmachines-co/basic-memory/actions/runs/1",
-    }
-
-
-def test_upsert_summary_block_replaces_existing_block() -> None:
-    body = "\n".join(
-        [
-            "Intro",
-            "<!-- BM_BOSSBOT_SUMMARY:start -->",
-            "Old summary",
-            "<!-- BM_BOSSBOT_SUMMARY:end -->",
-            "Footer",
-        ]
-    )
-
-    updated = bm_bossbot_status.upsert_summary_block(body, "New summary")
-
-    assert "Old summary" not in updated
-    assert "New summary" in updated
-    assert updated.startswith("Intro")
-    assert updated.endswith("Footer")
-
-
-def test_finalize_review_fetches_current_pr_body_before_upserting(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    event_path = tmp_path / "event.json"
-    event_path.write_text(json.dumps(_event_payload()), encoding="utf-8")
-    monkeypatch.setenv("GITHUB_TOKEN", "token")
-
-    updated_bodies: list[str] = []
-    statuses: list[Mapping[str, str]] = []
-
-    def fake_get_pull_request_body(*, token: str, repo: str, number: int) -> str:
-        assert token == "token"
-        assert repo == "basicmachines-co/basic-memory"
-        assert number == 925
-        return "Current body edited while the workflow was running"
-
-    def fake_update_pull_request_body(*, token: str, repo: str, number: int, body: str) -> None:
-        updated_bodies.append(body)
-
-    def fake_set_commit_status(
-        *,
-        token: str,
-        repo: str,
-        sha: str,
-        payload: Mapping[str, str],
-    ) -> None:
-        statuses.append(payload)
-
-    monkeypatch.setattr(bm_bossbot_status, "get_pull_request_body", fake_get_pull_request_body)
-    monkeypatch.setattr(
-        bm_bossbot_status, "update_pull_request_body", fake_update_pull_request_body
-    )
-    monkeypatch.setattr(bm_bossbot_status, "set_commit_status", fake_set_commit_status)
-    monkeypatch.setattr(bm_bossbot_status, "count_unresolved_review_threads", lambda **_: 0)
-
-    result = bm_bossbot_status.finalize_review(
-        event_path=event_path,
-        trusted=True,
-        repo=None,
-        run_url="https://github.com/basicmachines-co/basic-memory/actions/runs/1",
-        token_env="GITHUB_TOKEN",
-    )
-
-    assert result.approved is True
-    assert "Current body edited while the workflow was running" in updated_bodies[0]
-    assert "Event snapshot body" not in updated_bodies[0]
-    assert "Gate: deterministic" in updated_bodies[0]
-    assert statuses[0]["state"] == "success"
-
-
-def test_finalize_review_blocks_approval_on_unresolved_review_threads(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    event_path = tmp_path / "event.json"
-    event_path.write_text(json.dumps(_event_payload()), encoding="utf-8")
-    monkeypatch.setenv("GITHUB_TOKEN", "token")
-
-    statuses: list[Mapping[str, str]] = []
-
-    monkeypatch.setattr(bm_bossbot_status, "get_pull_request_body", lambda **_: "Body")
-    monkeypatch.setattr(bm_bossbot_status, "update_pull_request_body", lambda **_: None)
-    monkeypatch.setattr(
-        bm_bossbot_status,
-        "set_commit_status",
-        lambda *, token, repo, sha, payload: statuses.append(payload),
-    )
-    monkeypatch.setattr(bm_bossbot_status, "count_unresolved_review_threads", lambda **_: 2)
-
-    result = bm_bossbot_status.finalize_review(
-        event_path=event_path,
-        trusted=True,
-        repo=None,
-        run_url="https://github.com/basicmachines-co/basic-memory/actions/runs/1",
-        token_env="GITHUB_TOKEN",
-    )
-
-    assert result.approved is False
-    assert result.state == "failure"
-    assert result.description == "BM Bossbot found 2 unresolved review thread(s)"
-    assert statuses[0]["state"] == "failure"
-
-
-def test_finalize_review_fails_untrusted_author_without_counting_threads(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    event_path = tmp_path / "event.json"
-    event_path.write_text(json.dumps(_event_payload()), encoding="utf-8")
-    monkeypatch.setenv("GITHUB_TOKEN", "token")
-
-    monkeypatch.setattr(bm_bossbot_status, "get_pull_request_body", lambda **_: "Body")
-    monkeypatch.setattr(bm_bossbot_status, "update_pull_request_body", lambda **_: None)
-    monkeypatch.setattr(bm_bossbot_status, "set_commit_status", lambda **_: None)
-
-    def fail_count(**_: object) -> int:
-        raise AssertionError("thread count must not run for untrusted authors")
-
-    monkeypatch.setattr(bm_bossbot_status, "count_unresolved_review_threads", fail_count)
-
-    result = bm_bossbot_status.finalize_review(
-        event_path=event_path,
-        trusted=False,
-        repo=None,
-        run_url="https://github.com/basicmachines-co/basic-memory/actions/runs/1",
-        token_env="GITHUB_TOKEN",
-    )
-
-    assert result.approved is False
-    assert result.description == "BM Bossbot only gates owner/member/collaborator PRs"
-
-
-def test_count_unresolved_review_threads_pages_through_graphql_results(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    pages = [
-        {
-            "data": {
-                "repository": {
-                    "pullRequest": {
-                        "reviewThreads": {
-                            "pageInfo": {"hasNextPage": True, "endCursor": "CUR"},
-                            "nodes": [{"isResolved": False}, {"isResolved": True}],
-                        }
-                    }
-                }
-            }
-        },
-        {
-            "data": {
-                "repository": {
-                    "pullRequest": {
-                        "reviewThreads": {
-                            "pageInfo": {"hasNextPage": False, "endCursor": None},
-                            "nodes": [{"isResolved": False}],
-                        }
-                    }
-                }
-            }
-        },
-    ]
-    cursors: list[object] = []
-
-    # Signature mirrors bm_bossbot_status._github_request (payload: Mapping[str, Any] | None).
-    def fake_github_request(
-        *, method: str, path: str, token: str, payload: Mapping[str, Any] | None = None
-    ) -> object:
-        assert method == "POST"
-        assert path == "/graphql"
-        assert payload is not None
-        cursors.append(payload["variables"]["cursor"])
-        return pages.pop(0)
-
-    monkeypatch.setattr(bm_bossbot_status, "_github_request", fake_github_request)
-
-    count = bm_bossbot_status.count_unresolved_review_threads(
-        token="token", repo="basicmachines-co/basic-memory", number=925
-    )
-
-    assert count == 2
-    assert cursors == [None, "CUR"]
-
-
-def test_recheck_marks_failure_when_threads_are_unresolved(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    monkeypatch.setenv("GITHUB_TOKEN", "token")
-    statuses: list[Mapping[str, str]] = []
-
-    monkeypatch.setattr(bm_bossbot_status, "get_pull_request_head_sha", lambda **_: "abc123")
-    monkeypatch.setattr(bm_bossbot_status, "count_unresolved_review_threads", lambda **_: 3)
-    monkeypatch.setattr(
-        bm_bossbot_status,
-        "set_commit_status",
-        lambda *, token, repo, sha, payload: statuses.append({"sha": sha, **payload}),
-    )
-
-    bm_bossbot_status.recheck_threads(
-        repo="basicmachines-co/basic-memory",
-        number=925,
-        run_url="https://github.com/basicmachines-co/basic-memory/actions/runs/2",
-        token_env="GITHUB_TOKEN",
-    )
-
-    assert statuses[0]["sha"] == "abc123"
-    assert statuses[0]["state"] == "failure"
-    assert "3 unresolved review thread(s)" in statuses[0]["description"]
-
-
-def test_recheck_restores_prior_approval_when_threads_resolve(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    monkeypatch.setenv("GITHUB_TOKEN", "token")
-    statuses: list[Mapping[str, str]] = []
-
-    monkeypatch.setattr(bm_bossbot_status, "get_pull_request_head_sha", lambda **_: "abc123")
-    monkeypatch.setattr(bm_bossbot_status, "count_unresolved_review_threads", lambda **_: 0)
-    monkeypatch.setattr(bm_bossbot_status, "head_sha_was_approved", lambda **_: True)
-    monkeypatch.setattr(
-        bm_bossbot_status,
-        "set_commit_status",
-        lambda *, token, repo, sha, payload: statuses.append(payload),
-    )
-
-    bm_bossbot_status.recheck_threads(
-        repo="basicmachines-co/basic-memory",
-        number=925,
-        run_url="https://github.com/basicmachines-co/basic-memory/actions/runs/2",
-        token_env="GITHUB_TOKEN",
-    )
-
-    assert statuses[0]["state"] == "success"
-    assert statuses[0]["description"] == "BM Bossbot approved this head SHA"
-
-
-def test_recheck_leaves_status_alone_without_prior_approval(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    monkeypatch.setenv("GITHUB_TOKEN", "token")
-
-    monkeypatch.setattr(bm_bossbot_status, "get_pull_request_head_sha", lambda **_: "abc123")
-    monkeypatch.setattr(bm_bossbot_status, "count_unresolved_review_threads", lambda **_: 0)
-    monkeypatch.setattr(bm_bossbot_status, "head_sha_was_approved", lambda **_: False)
-
-    def fail_set_status(**_: object) -> None:
-        raise AssertionError("status must not change without a prior approval")
-
-    monkeypatch.setattr(bm_bossbot_status, "set_commit_status", fail_set_status)
-
-    bm_bossbot_status.recheck_threads(
-        repo="basicmachines-co/basic-memory",
-        number=925,
-        run_url="https://github.com/basicmachines-co/basic-memory/actions/runs/2",
-        token_env="GITHUB_TOKEN",
-    )
-
-
-def test_head_sha_was_approved_matches_only_the_approval_record(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    history = [
-        {
-            "context": "BM Bossbot Approval",
-            "state": "failure",
-            "description": "BM Bossbot found 2 unresolved review thread(s)",
-        },
-        {
-            "context": "BM Bossbot Approval",
-            "state": "success",
-            "description": "BM Bossbot approved this head SHA",
-        },
-        {"context": "license/cla", "state": "success", "description": "ok"},
-    ]
-
-    def _paged(pages: list[list[dict]]):
-        def fake(*, method: str, path: str, token: str, payload=None):
-            # Anchor on "&page=N" — a bare "page=N" substring also matches
-            # "per_page=100", which served page 1 forever and hung the loop.
-            for number, page in enumerate(pages, start=1):
-                if f"&page={number}" in path:
-                    return page
-            return []
-
-        return fake
-
-    monkeypatch.setattr(bm_bossbot_status, "_github_request", _paged([history]))
-    assert (
-        bm_bossbot_status.head_sha_was_approved(
-            token="token", repo="basicmachines-co/basic-memory", sha="abc123"
-        )
-        is True
-    )
-
-    monkeypatch.setattr(bm_bossbot_status, "_github_request", _paged([history[:1]]))
-    assert (
-        bm_bossbot_status.head_sha_was_approved(
-            token="token", repo="basicmachines-co/basic-memory", sha="abc123"
-        )
-        is False
-    )
-
-
-def test_head_sha_was_approved_pages_past_first_page_of_statuses(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """The recheck path can post >100 statuses; the approval may sit on page 2+."""
-    failure = {
-        "context": "BM Bossbot Approval",
-        "state": "failure",
-        "description": "BM Bossbot found 1 unresolved review thread(s)",
-    }
-    approval = {
-        "context": "BM Bossbot Approval",
-        "state": "success",
-        "description": "BM Bossbot approved this head SHA",
-    }
-    pages_served: list[str] = []
-
-    def fake(*, method: str, path: str, token: str, payload=None):
-        pages_served.append(path)
-        if "&page=1" in path:
-            return [failure] * 100
-        if "&page=2" in path:
-            return [approval]
-        return []
-
-    monkeypatch.setattr(bm_bossbot_status, "_github_request", fake)
-
-    assert (
-        bm_bossbot_status.head_sha_was_approved(
-            token="token", repo="basicmachines-co/basic-memory", sha="abc123"
-        )
-        is True
-    )
-    assert len(pages_served) == 2
-
-
-def test_finalize_cli_exits_nonzero_for_untrusted_author(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    event_path = tmp_path / "event.json"
-    event_path.write_text(json.dumps(_event_payload(body="Current body")), encoding="utf-8")
-    monkeypatch.setenv("GITHUB_TOKEN", "token")
-
-    updated_bodies: list[str] = []
-    statuses: list[Mapping[str, str]] = []
-
-    def fake_get_pull_request_body(*, token: str, repo: str, number: int) -> str:
-        return "Current body"
-
-    def fake_update_pull_request_body(*, token: str, repo: str, number: int, body: str) -> None:
-        updated_bodies.append(body)
-
-    def fake_set_commit_status(
-        *,
-        token: str,
-        repo: str,
-        sha: str,
-        payload: Mapping[str, str],
-    ) -> None:
-        statuses.append(payload)
-
-    monkeypatch.setattr(bm_bossbot_status, "get_pull_request_body", fake_get_pull_request_body)
-    monkeypatch.setattr(
-        bm_bossbot_status, "update_pull_request_body", fake_update_pull_request_body
-    )
-    monkeypatch.setattr(bm_bossbot_status, "set_commit_status", fake_set_commit_status)
-
-    result = CliRunner().invoke(
-        bm_bossbot_status.app,
-        [
-            "finalize",
-            "--event",
-            str(event_path),
-            "--trusted",
-            "false",
-            "--repo",
-            "basicmachines-co/basic-memory",
-            "--run-url",
-            "https://github.com/basicmachines-co/basic-memory/actions/runs/1",
-        ],
-    )
-
-    assert result.exit_code == 1
-    assert "BM Bossbot only gates owner/member/collaborator PRs" in updated_bodies[0]
-    assert statuses[0]["state"] == "failure"

tests/scripts/test_generate_pr_infographic.py

@@ -1,445 +0,0 @@
-import json
-from pathlib import Path
-
-import pytest
-from click import unstyle
-from typer.testing import CliRunner
-
-from scripts import generate_infographic, generate_pr_infographic
-
-
-def test_infographic_scripts_are_uv_typer_entrypoints() -> None:
-    for module in (generate_infographic, generate_pr_infographic):
-        source = module.__file__
-        assert source is not None
-        text = Path(source).read_text(encoding="utf-8")
-
-        assert text.startswith("#!/usr/bin/env -S uv run --script\n")
-        assert "# /// script" in text
-        assert "typer" in text
-        assert hasattr(module, "app")
-
-
-def test_generate_pr_infographic_cli_help_exposes_useful_options() -> None:
-    result = CliRunner().invoke(generate_pr_infographic.app, ["--help"])
-    help_text = unstyle(result.output)
-
-    assert result.exit_code == 0
-    assert "--pr-number" in help_text
-    assert "--pr-context-file" in help_text
-    assert "--output" in help_text
-    assert "--theme" in help_text
-    assert "--provenance-output" in help_text
-    assert "--print-prompt" in help_text
-    assert "--dry-run" in help_text
-
-
-def test_extract_pr_content_strips_managed_bot_blocks() -> None:
-    body = "\n".join(
-        [
-            "## Summary",
-            "Adds per-workspace rclone remotes for Team push/pull.",
-            "<!-- BM_BOSSBOT_SUMMARY:start -->",
-            "Reviewed SHA: abc123",
-            "Verdict: approve",
-            "<!-- BM_BOSSBOT_SUMMARY:end -->",
-            "<!-- pr-infographic:start -->",
-            "![BM Bossbot image for PR #42](https://example.test/img.webp)",
-            "<!-- pr-infographic:end -->",
-            "<!-- BM_INFOGRAPHIC_PROVENANCE:start -->",
-            "provenance details",
-            "<!-- BM_INFOGRAPHIC_PROVENANCE:end -->",
-            "## Test plan",
-            "pytest passes.",
-        ]
-    )
-
-    content = generate_pr_infographic.extract_pr_content(body)
-
-    assert "Adds per-workspace rclone remotes" in content
-    assert "pytest passes." in content
-    assert "Verdict: approve" not in content
-    assert "Reviewed SHA" not in content
-    assert "provenance details" not in content
-    assert "BM Bossbot image for PR" not in content
-
-
-def test_extract_pr_content_handles_body_without_managed_blocks() -> None:
-    assert generate_pr_infographic.extract_pr_content("Plain description") == "Plain description"
-
-
-def test_build_change_shape_digests_delivery_context() -> None:
-    context = {
-        "labels": [{"name": "enhancement"}, {"name": "cloud"}],
-        "closingIssuesReferences": [{"number": 581, "title": "edit_note recovery"}],
-        "commits": [
-            {"messageHeadline": "fix(mcp): recover edit_note"},
-            {"messageHeadline": "fix(api): canonicalize sync-file paths"},
-        ],
-        "files": [
-            {"path": "src/basic_memory/mcp/tools/edit_note.py", "additions": 120, "deletions": 30},
-            {"path": "tests/mcp/test_tool_edit_note.py", "additions": 80, "deletions": 5},
-        ],
-    }
-
-    context["commits"].append({"messageHeadline": "Merge branch 'main' into fix/581"})
-    shape = generate_pr_infographic.build_change_shape(context)
-
-    assert "Labels: enhancement, cloud" in shape
-    assert "#581: edit_note recovery" in shape
-    assert "Commit subjects (2 total):" in shape
-    assert "Merge branch" not in shape
-    assert "fix(mcp): recover edit_note" in shape
-    assert "Files changed (2 total, +200/-35):" in shape
-    assert "src/basic_memory/mcp/tools/edit_note.py (+120/-30)" in shape
-
-
-def test_build_change_shape_caps_long_lists_and_handles_empty_context() -> None:
-    context = {
-        "commits": [{"messageHeadline": f"commit {i}"} for i in range(15)],
-        "files": [{"path": f"file-{i}.py", "additions": i, "deletions": 0} for i in range(15)],
-    }
-
-    shape = generate_pr_infographic.build_change_shape(context)
-
-    assert "Commit subjects (15 total):" in shape
-    assert "- ... and 5 more" in shape
-    assert "- ... and 5 more files" in shape
-    # Files are ranked by churn, so the biggest file leads the list.
-    assert shape.index("file-14.py") < shape.index("file-5.py")
-
-    assert (
-        generate_pr_infographic.build_change_shape({}) == "(no additional change context available)"
-    )
-
-
-def test_extract_infographic_theme_from_pr_body() -> None:
-    body = "\n".join(
-        [
-            "Before",
-            "<!-- BM_INFOGRAPHIC_THEME:start -->",
-            "Italian movie poster with a release-route map",
-            "<!-- BM_INFOGRAPHIC_THEME:end -->",
-            "After",
-        ]
-    )
-
-    theme = generate_pr_infographic.extract_infographic_theme(body)
-
-    assert theme == "Italian movie poster with a release-route map"
-
-
-def test_extract_infographic_theme_is_optional() -> None:
-    assert generate_pr_infographic.extract_infographic_theme("No theme") is None
-
-
-def test_select_image_theme_reports_source() -> None:
-    body = "\n".join(
-        [
-            "<!-- BM_INFOGRAPHIC_THEME:start -->",
-            "paintings: Rembrandt-inspired merge gate",
-            "<!-- BM_INFOGRAPHIC_THEME:end -->",
-        ]
-    )
-
-    from_body = generate_pr_infographic.select_image_theme(
-        pr_number=42,
-        pr_title="feat(ci): add a merge gate",
-        pr_body=body,
-        theme_override=None,
-    )
-    from_cli = generate_pr_infographic.select_image_theme(
-        pr_number=42,
-        pr_title="feat(ci): add a merge gate",
-        pr_body=body,
-        theme_override="80's action movies",
-    )
-    from_auto = generate_pr_infographic.select_image_theme(
-        pr_number=42,
-        pr_title="feat(ci): add a merge gate",
-        pr_body="No theme",
-        theme_override=None,
-    )
-
-    assert from_body.theme == "paintings: Rembrandt-inspired merge gate"
-    assert from_body.source == generate_pr_infographic.ThemeSource.PR_BODY
-    assert from_cli.theme == "80's action movies"
-    assert from_cli.source == generate_pr_infographic.ThemeSource.CLI
-    assert from_auto.theme in generate_pr_infographic.BM_IMAGE_THEME_POOL
-    assert from_auto.source == generate_pr_infographic.ThemeSource.AUTO
-
-
-def test_build_infographic_prompt_depicts_pr_content_not_review_outcome() -> None:
-    prompt = generate_pr_infographic.build_infographic_prompt(
-        pr_number=42,
-        pr_title="feat(sync): stream large files during cloud sync",
-        pr_content="Streams PDFs in chunks instead of loading them fully into memory.",
-        change_shape="Labels: sync\nFiles changed (3 total, +90/-20):\n- src/basic_memory/sync/x.py (+80/-15)",
-        theme="WWII propaganda posters with home-front logistics routes",
-        theme_source=generate_pr_infographic.ThemeSource.CLI,
-    )
-
-    assert "PR #42" in prompt
-    assert "stream large files during cloud sync" in prompt
-    assert "Streams PDFs in chunks" in prompt
-    assert "WWII propaganda posters" in prompt
-    assert "User-supplied visual direction" in prompt
-    assert "style inspiration only" in prompt
-    assert "polished landscape WebP editorial image" in prompt
-    assert "image-first composition" in prompt
-    assert "symbolic tableau" in prompt
-    assert "Do not render an infographic" in prompt
-    assert "dashboard" in prompt
-    assert "flowchart" in prompt
-    assert "copyrighted characters" in prompt
-    # The subject is the change itself; review-process imagery is banned.
-    assert "CONTENT of the pull request" in prompt
-    assert "do not depict review verdicts" in prompt
-    assert "approval" in prompt.lower()
-    assert "stamps" in prompt
-    assert "checkmarks" in prompt
-    # Change shape grounds the imagery but must not become captions.
-    assert "Change shape" in prompt
-    assert "src/basic_memory/sync/x.py" in prompt
-    assert "never render file" in prompt
-    # The old prompt fed the review summary and named the approval status,
-    # which produced literal "BOSSBOT APPROVED" stamp images.
-    assert "BM Bossbot summary:" not in prompt
-    assert "BM Bossbot Approval" not in prompt
-
-
-def test_build_infographic_provenance_block_includes_image_choices_without_prompt() -> None:
-    block = generate_pr_infographic.build_infographic_provenance_block(
-        pr_number=42,
-        output_path=Path("docs/assets/infographics/pr-42.webp"),
-        model="gpt-image-2",
-        size="1536x1024",
-        quality="high",
-        theme="classic black-and-white photography",
-        theme_source=generate_pr_infographic.ThemeSource.CLI,
-    )
-
-    assert generate_pr_infographic.PROVENANCE_START in block
-    assert generate_pr_infographic.PROVENANCE_END in block
-    assert "BM Bossbot image choices" in block
-    assert "Generated asset: `docs/assets/infographics/pr-42.webp`" in block
-    assert "Image model: `gpt-image-2`" in block
-    assert "Size: `1536x1024`" in block
-    assert "Quality: `high`" in block
-    assert "Image mode: `editorial-image`" in block
-    assert "Theme source: `cli`" in block
-    assert "classic black-and-white photography" in block
-    assert "Image prompt sent to" not in block
-    assert "Images API revised prompt" not in block
-
-
-def test_upsert_managed_block_appends_and_replaces() -> None:
-    first = "\n".join(
-        [
-            generate_pr_infographic.PROVENANCE_START,
-            "first",
-            generate_pr_infographic.PROVENANCE_END,
-        ]
-    )
-    second = "\n".join(
-        [
-            generate_pr_infographic.PROVENANCE_START,
-            "second",
-            generate_pr_infographic.PROVENANCE_END,
-        ]
-    )
-
-    appended = generate_pr_infographic.upsert_managed_block(
-        "Existing body",
-        block=first,
-        start=generate_pr_infographic.PROVENANCE_START,
-        end=generate_pr_infographic.PROVENANCE_END,
-    )
-    replaced = generate_pr_infographic.upsert_managed_block(
-        appended,
-        block=second,
-        start=generate_pr_infographic.PROVENANCE_START,
-        end=generate_pr_infographic.PROVENANCE_END,
-    )
-
-    assert appended == f"Existing body\n\n{first}\n"
-    assert "first" not in replaced
-    assert "second" in replaced
-    assert replaced.count(generate_pr_infographic.PROVENANCE_START) == 1
-
-
-def test_build_infographic_prompt_uses_auto_theme_as_visual_direction() -> None:
-    theme = generate_pr_infographic.select_image_theme(
-        pr_number=42,
-        pr_title="feat(ci): add a merge gate",
-        pr_body="No theme",
-        theme_override=None,
-    )
-    prompt = generate_pr_infographic.build_infographic_prompt(
-        pr_number=42,
-        pr_title="feat(ci): add a merge gate",
-        pr_content="Adds a deterministic merge gate for pull requests.",
-        change_shape="(no additional change context available)",
-        theme=theme.theme,
-        theme_source=theme.source,
-    )
-
-    assert "Selected BM visual direction" in prompt
-    assert theme.theme in prompt
-    assert "Use image-first composition" in prompt
-    assert "movie poster" in prompt
-    assert "painting" in prompt
-    assert "classic photograph" in prompt
-    assert "scene" in prompt
-    assert "poster" in prompt
-    assert "cover image" in prompt
-    assert "symbolic tableau" in prompt
-    assert "Use at most a short title" in prompt
-    assert "Do not render an infographic" in prompt
-    assert "dashboard" in prompt
-    assert "flowchart" in prompt
-    assert "bullet-list panel" in prompt
-
-
-@pytest.mark.parametrize("flag", ["--print-prompt", "--dry-run"])
-def test_generate_pr_infographic_can_print_prompt_without_image_call(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-    flag: str,
-) -> None:
-    context_file = tmp_path / "pr-context.json"
-    context_file.write_text(
-        json.dumps(
-            {
-                "title": "feat(ci): add a merge gate",
-                "body": "\n".join(
-                    [
-                        "Adds a deterministic merge gate for pull requests.",
-                        "<!-- BM_BOSSBOT_SUMMARY:start -->",
-                        "Verdict: approve",
-                        "Summary: review artifact that must not reach the image.",
-                        "<!-- BM_BOSSBOT_SUMMARY:end -->",
-                        "<!-- BM_INFOGRAPHIC_THEME:start -->",
-                        "space exploration and astronomy",
-                        "<!-- BM_INFOGRAPHIC_THEME:end -->",
-                    ]
-                ),
-                "labels": [{"name": "ci"}],
-                "commits": [{"messageHeadline": "feat(ci): add a merge gate"}],
-                "files": [{"path": ".github/workflows/gate.yml", "additions": 40, "deletions": 2}],
-                "closingIssuesReferences": [{"number": 7, "title": "merge gate"}],
-            }
-        ),
-        encoding="utf-8",
-    )
-
-    def fail_generate_image_result(**_: object) -> generate_infographic.GeneratedImage:
-        raise AssertionError("print-prompt mode must not call image generation")
-
-    monkeypatch.setattr(
-        generate_pr_infographic, "generate_image_result", fail_generate_image_result
-    )
-    output = tmp_path / "docs/assets/infographics/pr-42.webp"
-
-    result = CliRunner().invoke(
-        generate_pr_infographic.app,
-        [
-            "--pr-number",
-            "42",
-            "--pr-context-file",
-            str(context_file),
-            "--output",
-            str(output),
-            flag,
-        ],
-    )
-
-    assert result.exit_code == 0, result.output
-    assert (
-        "Create a polished landscape WebP editorial image for Basic Memory PR #42" in result.output
-    )
-    assert "feat(ci): add a merge gate" in result.output
-    assert "Adds a deterministic merge gate" in result.output
-    assert "space exploration and astronomy" in result.output
-    assert "Labels: ci" in result.output
-    assert ".github/workflows/gate.yml" in result.output
-    assert "#7: merge gate" in result.output
-    assert "image-first composition" in result.output
-    assert "Do not render an infographic" in result.output
-    assert "Verdict: approve" not in result.output
-    assert "must not reach the image" not in result.output
-    assert not output.exists()
-
-
-def test_generate_pr_infographic_writes_provenance_after_image_generation(
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    context_file = tmp_path / "pr-context.json"
-    context_file.write_text(
-        json.dumps(
-            {
-                "title": "feat(ci): add a merge gate",
-                "body": "\n".join(
-                    [
-                        "Adds a merge gate.",
-                        "<!-- BM_INFOGRAPHIC_THEME:start -->",
-                        "paintings: Rembrandt-inspired merge gate",
-                        "<!-- BM_INFOGRAPHIC_THEME:end -->",
-                    ]
-                ),
-            }
-        ),
-        encoding="utf-8",
-    )
-
-    def fake_generate_image_result(**kwargs: object) -> generate_infographic.GeneratedImage:
-        output_path = kwargs["output_path"]
-        assert isinstance(output_path, Path)
-        output_path.parent.mkdir(parents=True, exist_ok=True)
-        output_path.write_bytes(b"fake-webp")
-        return generate_infographic.GeneratedImage(
-            path=output_path,
-            revised_prompt="A Rembrandt-inspired painting of a robot guarding a merge gate.",
-        )
-
-    monkeypatch.setattr(
-        generate_pr_infographic, "generate_image_result", fake_generate_image_result
-    )
-    output = tmp_path / "docs/assets/infographics/pr-42.webp"
-    provenance = tmp_path / "provenance.md"
-
-    result = CliRunner().invoke(
-        generate_pr_infographic.app,
-        [
-            "--pr-number",
-            "42",
-            "--pr-context-file",
-            str(context_file),
-            "--output",
-            str(output),
-            "--provenance-output",
-            str(provenance),
-        ],
-    )
-
-    assert result.exit_code == 0, result.output
-    assert output.exists()
-    text = provenance.read_text(encoding="utf-8")
-    assert "Generated asset:" in text
-    assert "Image mode: `editorial-image`" in text
-    assert "Theme source: `pr-body`" in text
-    assert "paintings: Rembrandt-inspired merge gate" in text
-    assert "Image prompt sent to" not in text
-    assert "Images API revised prompt" not in text
-    assert "robot guarding a merge gate" not in text
-    assert "Adds a merge gate" not in text
-
-
-def test_validate_output_path_must_stay_under_docs_assets_infographics(tmp_path: Path) -> None:
-    good = tmp_path / "docs/assets/infographics/pr-42.webp"
-    bad = tmp_path / "docs/assets/pr-42.webp"
-
-    assert generate_infographic.validate_output_path(good, repo_root=tmp_path) == good
-    with pytest.raises(ValueError, match="docs/assets/infographics"):
-        generate_infographic.validate_output_path(bad, repo_root=tmp_path)

tests/services/test_context_service.py

@@ -212,6 +212,47 @@ async def test_build_context_with_observations(context_service, test_graph):
     assert "Root note" in note_observation.content
 
 
+@pytest.mark.asyncio
+async def test_build_context_observation_permalinks_match_search_index(
+    context_service, search_service, entity_service
+):
+    """Regression test for #929: observation permalinks must match the search index.
+
+    build_context used to rebuild the synthetic observation permalink inline,
+    without the 200-char truncation (#446) or the content digest (#931) that
+    Observation.permalink applies, so for long observations it returned
+    permalinks the search index doesn't contain.
+    """
+    from basic_memory.schemas.base import Entity as EntitySchema
+    from basic_memory.schemas.search import SearchQuery
+
+    long_observation = "x" * 210 + " LONG_OBS_MARKER"
+    entity, _ = await entity_service.create_or_update_entity(
+        EntitySchema(
+            title="Long Obs Entity",
+            note_type="test",
+            directory="test",
+            content=f"# Long Obs Entity\n- [note] {long_observation}\n",
+        )
+    )
+    await search_service.index_entity(entity)
+
+    url = memory_url.validate_strings(f"memory://{entity.permalink}")
+    context_result = await context_service.build_context(url, include_observations=True)
+    assert len(context_result.results) == 1
+    context_item = context_result.results[0]
+    assert len(context_item.observations) == 1
+    obs_row = context_item.observations[0]
+
+    # The model property is the single definition of the permalink format
+    assert obs_row.permalink == entity.observations[0].permalink
+
+    # The search index row for this observation carries the same permalink
+    index_rows = await search_service.search(SearchQuery(text="LONG_OBS_MARKER"))
+    obs_permalinks = [r.permalink for r in index_rows if r.type == SearchItemType.OBSERVATION.value]
+    assert obs_permalinks == [obs_row.permalink]
+
+
 @pytest.mark.asyncio
 async def test_build_context_not_found(context_service):
     """Test handling non-existent permalinks."""

tests/test_coerce.py

@@ -1,9 +1,9 @@
-"""Tests for coerce_list and coerce_dict utility functions.
+"""Tests for coerce_list, coerce_dict, and strict_search_tags utility functions.
 
 These must fail until the helpers are implemented in utils.py.
 """
 
-from basic_memory.utils import coerce_list, coerce_dict
+from basic_memory.utils import coerce_list, coerce_dict, strict_search_tags
 
 
 class TestCoerceList:
@@ -33,6 +33,64 @@ class TestCoerceList:
         assert coerce_list(42) == 42
 
 
+class TestStrictSearchTags:
+    """Tests for strict_search_tags (the search_notes tags boundary coercer)."""
+
+    def test_none_parses_to_empty_list(self):
+        assert strict_search_tags(None) == []
+
+    def test_comma_string_splits(self):
+        assert strict_search_tags("a,b") == ["a", "b"]
+
+    def test_list_with_comma_element_splits(self):
+        assert strict_search_tags(["alpha,beta"]) == ["alpha", "beta"]
+
+    def test_plain_list_passthrough(self):
+        assert strict_search_tags(["a", "b"]) == ["a", "b"]
+
+    def test_json_array_string(self):
+        assert strict_search_tags('["a", "b"]') == ["a", "b"]
+
+    def test_int_passthrough_for_pydantic_rejection(self):
+        """Unsupported types pass through unchanged so Pydantic rejects them."""
+        assert strict_search_tags(42) == 42
+
+    def test_dict_passthrough_for_pydantic_rejection(self):
+        value = {"a": 1}
+        assert strict_search_tags(value) is value
+
+    def test_int_list_passthrough_for_pydantic_rejection(self):
+        """Lists with non-string elements pass through unchanged so Pydantic rejects them."""
+        value = [42]
+        assert strict_search_tags(value) is value
+
+    def test_dict_list_passthrough_for_pydantic_rejection(self):
+        value = [{"a": 1}]
+        assert strict_search_tags(value) is value
+
+    def test_mixed_list_passthrough_for_pydantic_rejection(self):
+        """One bad element poisons the whole list — no partial stringification."""
+        value = ["ok", 42]
+        assert strict_search_tags(value) is value
+
+    def test_json_array_string_with_int_passthrough_for_pydantic_rejection(self):
+        """A JSON-array string with non-string elements must not be stringified."""
+        value = "[42]"
+        assert strict_search_tags(value) is value
+
+    def test_json_array_string_with_dict_passthrough_for_pydantic_rejection(self):
+        value = '[{"a": 1}]'
+        assert strict_search_tags(value) is value
+
+    def test_json_array_string_mixed_passthrough_for_pydantic_rejection(self):
+        """One bad element poisons the whole JSON-array string — no partial parse."""
+        value = '["ok", 42]'
+        assert strict_search_tags(value) is value
+
+    def test_json_array_string_all_strings_still_parses(self):
+        assert strict_search_tags('["a","b"]') == ["a", "b"]
+
+
 class TestCoerceDict:
     """Tests for coerce_dict."""
 

uv.lock

@@ -322,6 +322,7 @@ dev = [
     { name = "pytest-asyncio" },
     { name = "pytest-cov" },
     { name = "pytest-mock" },
+    { name = "pytest-split" },
     { name = "pytest-testmon" },
     { name = "pytest-timeout" },
     { name = "pytest-xdist" },
@@ -390,6 +391,7 @@ dev = [
     { name = "pytest-asyncio", specifier = ">=0.24.0" },
     { name = "pytest-cov", specifier = ">=4.1.0" },
     { name = "pytest-mock", specifier = ">=3.12.0" },
+    { name = "pytest-split", specifier = ">=0.11.0" },
     { name = "pytest-testmon", specifier = ">=2.2.0" },
     { name = "pytest-timeout", specifier = ">=2.4.0" },
     { name = "pytest-xdist", specifier = ">=3.0.0" },
@@ -3060,6 +3062,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/5a/cc/06253936f4a7fa2e0f48dfe6d851d9c56df896a9ab09ac019d70b760619c/pytest_mock-3.15.1-py3-none-any.whl", hash = "sha256:0a25e2eb88fe5168d535041d09a4529a188176ae608a6d249ee65abc0949630d", size = 10095, upload-time = "2025-09-16T16:37:25.734Z" },
 ]
 
+[[package]]
+name = "pytest-split"
+version = "0.11.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pytest" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/2f/16/8af4c5f2ceb3640bb1f78dfdf5c184556b10dfe9369feaaad7ff1c13f329/pytest_split-0.11.0.tar.gz", hash = "sha256:8ebdb29cc72cc962e8eb1ec07db1eeb98ab25e215ed8e3216f6b9fc7ce0ec2b5", size = 13421, upload-time = "2026-02-03T09:14:31.469Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ae/a1/d4423657caaa8be9b31e491592b49cebdcfd434d3e74512ce71f6ec39905/pytest_split-0.11.0-py3-none-any.whl", hash = "sha256:899d7c0f5730da91e2daf283860eb73b503259cb416851a65599368849c7f382", size = 11911, upload-time = "2026-02-03T09:14:33.708Z" },
+]
+
 [[package]]
 name = "pytest-testmon"
 version = "2.2.0"