Compare commits

..

7 Commits

Author SHA1 Message Date
phernandez 2c30922e65 docs(core): add Basic Machines agent style guidance
Signed-off-by: phernandez <paul@basicmachines.co>
2026-06-08 15:37:43 -05:00
phernandez e8cda55ddb test(cli): make copyto path assertions Windows-portable
The Windows SQLite unit job failed because two new copyto tests asserted on
raw POSIX dest strings. On Windows `str(Path("/tmp/research"))` renders with
backslashes, so the copyto dest is `\tmp\research/notes/...` (rclone accepts the
mixed separators — the product is fine). Compare the local dest via `Path(...)`
like the other tests, which normalizes separators cross-platform.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-06-08 13:00:48 -05:00
phernandez c850450ddf fix(cli): address push/pull review feedback (fatal-error guard, UX, tests)
From the Codex and claude-review automated reviews on PR #917:

- project_diff: fail fast on fatal `rclone check` errors. A non-zero exit with
  no combined listing (auth/network/missing-remote) previously produced an empty
  plan, so the transfer ran as a no-op and reported success. Now raises
  RcloneError with rclone's stderr. (Codex P2 / claude-review #2)
- sync-setup "Next steps": stop pointing every user at the now-Personal-only
  `bm cloud sync`; lead with the Team-safe `pull`/`push` and note bisync is
  Personal-only. (claude-review #1)
- Cosmetic: capitalize the push/pull abort headlines for consistency; document
  the two ConflictStrategy definitions (Typer enum vs engine Literal). (nits)
- Tests: cover keep-local+pull and keep-cloud+push (preserve-destination cases),
  project_transfer aborting on a mid-loop conflict-copy failure, and project_diff
  raising on a fatal check error vs not raising on a normal differences exit.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-06-08 12:30:20 -05:00
phernandez ad1f8621f1 docs(cli): document cloud push/pull and Personal-only sync/bisync
Update docs/cloud-cli.md for the docs site:
- Add a Team Workspaces push/pull section (additive, git-style transfers,
  the --on-conflict {fail|keep-local|keep-cloud|keep-both} resolution, and the
  no-baseline limitations tracked by #862).
- Mark `sync` and `bisync` as Personal-workspace-only mirrors and point Team
  users at push/pull, with a Personal-vs-Team overview table.
- Fix stale command namespace: `bm project sync|bisync|check|bisync-reset` ->
  `bm cloud ...` (these live under `bm cloud`, not `bm project`).
- Add push/pull to the command reference and the summary workflows.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-06-08 12:25:22 -05:00
phernandez a7937d4066 fix(cli): compare by checksum on overwrite push/pull to match conflict detection
Second code-review finding (PLAUSIBLE): project_diff detects conflicts with
`rclone check` (content/hash), but project_copy used `rclone copy` with its
default size+modtime comparison. On an overwrite strategy (keep-cloud on pull /
keep-local on push), copy could skip a file the diff had flagged as a conflict
when sizes matched and the destination was not older — silently ignoring the
user's explicit choice.

Add `--checksum` to the overwrite copy so the transfer decision uses the same
content basis as detection. New-only mode is unaffected: `--ignore-existing`
skips by existence, so the comparison basis is irrelevant there. project_sync
(the Personal-only mirror) is untouched and keeps its default comparison.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-06-08 11:48:20 -05:00
phernandez 9dbe313193 fix(cli): add --local-no-preallocate to conflict-copy copyto
Code-review finding: project_copy_file built its `rclone copyto` command
inline and omitted `--local-no-preallocate`. On a `pull --on-conflict keep-both`
that copyto writes the conflict copy to the local filesystem, which is exactly
the case the flag guards (NUL byte padding on virtual filesystems such as
Google Drive File Stream). Every other transfer path adds it via
_build_transfer_cmd; bring copyto in line.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-06-08 11:44:20 -05:00
phernandez e23363606a feat(cli): add Team-safe cloud push/pull and gate sync to Personal workspaces
Adds `bm cloud push` and `bm cloud pull` as git-style, fail-safe transfer
primitives that are usable on Team workspaces (issue #858), and restricts the
destructive `bm cloud sync` mirror to Personal workspaces.

Why
- `bm cloud sync` is a destructive local->cloud mirror; on a shared Team bucket
  it can delete a teammate's files. The only pull path was two-way `bisync`,
  which is already Personal-only (#849).
- Teams need a safe way to fetch teammates' notes and add their own without one
  stale local tree becoming authoritative for shared cloud state.

What
- push = `rclone copy` local->cloud, pull = `rclone copy` cloud->local. Both are
  additive (never delete on the destination), so neither can damage shared state.
- Conflicts (a file that differs on both sides) abort by default and list the
  paths, like git refusing to clobber local changes / rejecting a stale push.
  `--on-conflict {fail|keep-local|keep-cloud|keep-both}` lets the user decide;
  no vague --force, no silent winner.
- `sync` now requires a Personal workspace; its guard and the bisync guard point
  Team users at push/pull.

Limitations (surfaced in --help and command output, tracked by #862):
- No sync baseline yet, so deletions are not propagated and every divergence is
  treated as a conflict rather than auto-resolved. Real three-way merge needs
  the per-client manifest + Tigris snapshot baseline designed in #862.

Implementation
- rclone_commands.py: shared `_build_transfer_cmd`/`_transfer_endpoints`;
  refactor `project_sync` onto them (no behavior change); add `project_diff`
  (conflict detection via `rclone check --combined`), `project_copy`,
  `project_copy_file`, and `project_transfer` (strategy dispatch).
- project_sync.py: new `push`/`pull` commands (ungated, Team-safe) with
  `--on-conflict`/`--dry-run`; gate `sync` to Personal via a per-command guard
  message.
- Tests at the rclone-argv and CLI-command levels; existing sync/bisync tests
  updated for the new gating and messages.

Signed-off-by: phernandez <paul@basicmachines.co>
2026-06-08 11:35:13 -05:00
85 changed files with 1062 additions and 4716 deletions
@@ -1,5 +1,5 @@
---
name: code-review
name: basic-machines-review
description: Use when reviewing Basic Machines code for house style, architecture risk, pre-merge hardening, or whether a change fits basic-memory/basic-memory-cloud conventions.
license: MIT
---
@@ -18,9 +18,9 @@ Review the current diff or named files against:
- `docs/ENGINEERING_STYLE.md`
- The touched code paths and tests
Apply only the guidance for the active repo. In `basic-memory`, prioritize local-first
file/database/MCP boundaries. In `basic-memory-cloud`, prioritize tenant/workspace isolation,
cloud worker behavior, and web-v2 state/runtime boundaries.
For `basic-memory`, prioritize local-first file/database/MCP boundaries. For
`basic-memory-cloud`, prioritize tenant/workspace isolation, cloud worker behavior, and
web-v2 state/runtime boundaries.
## Review Rubric
@@ -53,12 +53,6 @@ Report only concrete, falsifiable risks:
Lead with findings ordered by severity. Each finding should include:
| Severity | Use for |
| -------- | ------- |
| `high` | A likely correctness, security, data-loss, or tenant/workspace isolation failure |
| `medium` | A concrete maintainability or boundary risk that can cause future defects |
| `low` | A minor consistency issue, ambiguous guidance, or review-only cleanup |
```text
severity | file:line | risk category | claim
Why: concrete behavior or code path that proves the risk.
-48
View File
@@ -1,48 +0,0 @@
---
name: fix-pr-issues
description: Use when addressing Basic Memory pull request feedback, failed checks, or BM Bossbot blockers from Codex.
---
# Fix Basic Memory PR Issues
Resolve PR feedback and failed checks, then wait for BM Bossbot to approve the
new head SHA. This skill never merges a PR.
## Gather
1. Identify the PR:
- `gh pr view --json number,url,headRefOid,mergeStateStatus,statusCheckRollup`
2. Collect feedback:
- PR comments and review summaries
- inline review comments and unresolved review threads
- failed GitHub Actions jobs and relevant logs
- the managed `BM_BOSSBOT_SUMMARY` block in the PR body
3. Build a short issue ledger:
- source
- concrete problem
- expected fix
- verification needed
## Fix
1. Address one ledger item at a time.
2. Read each file in full before editing it.
3. Keep diffs narrow and preserve unrelated user changes.
4. Run the smallest meaningful verification first, then widen as needed.
5. Commit with `git commit -s` when code or docs changed.
## Push And Recheck
1. Push the branch.
2. Watch checks for the new `headRefOid`.
3. Wait for the required `BM Bossbot Approval` status to pass on that exact SHA.
4. If BM Bossbot reviews an older SHA, treat the approval as stale and keep
waiting for the current one.
## Reply
For each addressed comment or blocker, reply with the fix commit, verification
run, and current BM Bossbot status. Do not resolve or dismiss substantive
feedback without evidence.
@@ -1,7 +0,0 @@
interface:
display_name: "Fix PR Issues"
short_description: "Address PR feedback and BM Bossbot blockers"
icon_small: "./assets/icon.svg"
icon_large: "./assets/icon.svg"
brand_color: "#2563EB"
default_prompt: "Use $fix-pr-issues to address PR feedback and wait for BM Bossbot Approval on the latest head SHA."
@@ -1,5 +0,0 @@
<svg xmlns="http://www.w3.org/2000/svg" width="128" height="128" viewBox="0 0 24 24" fill="none" stroke="#111827" stroke-width="1.8" stroke-linecap="round" stroke-linejoin="round">
<path d="M3 12h4l2-6 4 12 2-6h6"/>
<path d="M4 20h16"/>
</svg>

Before

Width:  |  Height:  |  Size: 249 B

-247
View File
@@ -1,247 +0,0 @@
---
name: infographics
description: Use when generating Basic Memory PR, changelog, release, or weekly images from Codex.
---
# Basic Memory Images
Generate repository visuals with evidence-grounded content and canonical output
paths. The file and marker names still say "infographic" for compatibility, but
PR generation is image-first: scene, poster, painting, photograph, cover,
tableau, staged artifact, or another editorial visual moment that describes the
intent of the PR. PR images are non-gating BM Bossbot artifacts; changelog and
release-summary images are manual evidence-pack workflows.
## Output Contract
- Base output directory: `docs/assets/infographics/`
- PR image: `docs/assets/infographics/pr-<number>.webp`
- Changelog image: `docs/assets/infographics/changelog.webp`
- Weekly image:
- This is always a 2-Week Retro window: previous ISO week through current ISO
week (`start-week = current-week - 1`, `end-week = current-week`).
- Same year window: `docs/assets/infographics/<year>-w<start-week>-w<end-week>.webp`
- Cross-year window:
`docs/assets/infographics/<start-year>-w<start-week>-<end-year>-w<end-week>.webp`
## PR Mode
PR mode uses the BM Bossbot summary block as source material. Do not hand-write
claims that are not present in the PR body.
1. Fetch the PR body:
```bash
gh pr view <number> --json body --jq '.body // ""' > /tmp/bm-pr-body.md
```
2. Generate the canonical asset:
```bash
uv run --script scripts/generate_pr_infographic.py \
--pr-number <number> \
--pr-body-file /tmp/bm-pr-body.md \
--theme "<optional visual theme>" \
--provenance-output /tmp/bm-infographic-provenance.md \
--output docs/assets/infographics/pr-<number>.webp
```
If the PR body contains a managed image theme block, the script reads it
automatically:
```markdown
<!-- BM_INFOGRAPHIC_THEME:start -->
<theme>
<!-- BM_INFOGRAPHIC_THEME:end -->
```
Before spending an image call, test the prompt path locally:
```bash
uv run --script scripts/generate_pr_infographic.py \
--pr-number <number> \
--pr-body-file /tmp/bm-pr-body.md \
--theme "<optional visual theme>" \
--output docs/assets/infographics/pr-<number>.webp \
--print-prompt
```
`--dry-run` is an alias for `--print-prompt`; both print the final prompt and
exit without calling OpenAI.
When no theme is supplied, the script selects a deterministic BM visual
direction from the style pool below based on the PR number and Bossbot summary.
This keeps repeated PR images from collapsing into the same generic visual.
When the image is generated, also write provenance with
`--provenance-output <path>`. BM Bossbot publishes that managed block into the
PR body with these markers:
```markdown
<!-- BM_INFOGRAPHIC_PROVENANCE:start -->
...
<!-- BM_INFOGRAPHIC_PROVENANCE:end -->
```
The provenance block records the generated asset path, image model, size,
quality, image mode, theme source, and selected visual direction. It
intentionally does not dump the full generated prompt into the PR body. Treat
this block as debugging and creative provenance only; it is not a merge gate.
The PR image is visual support only. The authoritative merge gate is the
GitHub commit status named `BM Bossbot Approval`.
## Changelog Mode
Build an evidence pack before writing a prompt:
- diff truth source: merged PR diffs, merge commits, or local reconstructed diffs
- changed-file orientation: `git diff --stat` plus key file reads
- impact ledger: before/after outcomes tied to actual changes
- discard list: misleading titles, reverted work, rename-only churn, speculative TODOs
- chosen image form: poster, scene, tableau, cover, painting, photograph,
staged artifact, or another editorial visual moment
- chosen BM style category: exactly one category from the selection pool below
Read these references before drafting the prompt:
- `references/prompt-blueprint.md`
- `references/style-balance.md`
Read the current `CHANGELOG.md` entries and include the latest meaningful
changes.
## Style And Category Selection
Select exactly one BM style category per image based on semantic fit. The
visual language should be recognizable and tasteful, while staying
business-readable.
Create an image-first visual form that communicates the change: poster, scene,
tableau, cover image, painting, photograph, staged artifact, or another
editorial visual moment. Maps, diagrams, dossiers, charts, and labels can appear
as props inside the scene, but do not make a text-heavy infographic.
BM category pool:
- computer science college textbooks: SICP-style diagrams, algorithms lectures,
compiler pipelines, automata, database systems, type theory, operating systems
- classic literature subjects: sea voyages, gothic manors, Dickensian city maps,
Austen social graphs, library marginalia, travel journals
- fantasy/D&D-inspired: quest maps, dungeon keys, guild ledgers, spellbooks,
bestiaries, tavern notice boards; no copyrighted settings
- Music: Metal, Hard Rock, Punk, techno, soul, reggae bands; no pop music, no
direct band logos, album covers, or musician likenesses
- sci-fi: Star Wars inspired knockoff, Spaceballs-adjacent space opera, fleet
routes, mission consoles, contraband manifests; avoid copyrighted characters,
logos, or named fictional universes
- Conan the barbarian-inspired sword-and-sorcery: ruined temples, desert routes,
battle standards, ancient maps; no named character likenesses
- Comic books: issue covers, splash pages, action-panel maps, caption boxes,
halftone energy, clean sound-effect typography
- French new wave movies: poster style, stark typography, city route maps,
jump-cut sequencing, high-contrast editorial photography cues
- WWII propaganda posters: home-front public-information poster language,
logistics arrows, ration charts, mobilization maps, bold simplified figures;
no real-world party symbols, hate imagery, dehumanizing slogans, or false
historical claims
- Italian movie posters: hand-painted drama, bold credits, expressive color,
route-map collage, 1960s or 1970s cinema energy; no direct film titles or
actor likenesses
- Shakespeare: stage maps, acts and scenes, dramatis personae, royal courts,
backstage cue sheets
- Greek mythology: temple diagrams, constellation routes, hero's journey maps,
oracle tablets, labyrinths, ship routes
- noir detective boards: case files, red-string maps, typed evidence labels,
precinct wall charts
- NASA mission-control dashboards: launch timelines, telemetry maps, orbital
routes, status boards
- space exploration and astronomy: celestial atlases, observatory charts,
star-field maps, orbital mechanics diagrams, planetary survey routes,
telescope annotations, mission trajectories, deep-space timelines
- paintings: abstract painting, classical landscape, Remington-inspired western
action painting, Rembrandt-inspired chiaroscuro, historical mural, stormy
seascape, allegorical editorial painting
- classic black-and-white photography: documentary field report, newsroom
archive print, editorial photo essay, street photography, high-contrast
darkroom print, contact sheet, civic infrastructure photograph
- 80's action movies: practical explosions, smoky backlit warehouses, neon city
streets, helicopter searchlights, mission dossiers, heroic silhouettes,
high-stakes countdowns, painted ensemble posters; no direct actor likenesses,
real film titles, franchise marks, or catchphrases
- alchemy manuscripts: transformation diagrams, annotated symbols, recipe-like
process maps, illuminated margins
- brutalist civic planning: transit maps, concrete signage, zoning blocks,
infrastructure diagrams
Selection rules:
- Pick one category only; do not create mixed mashups.
- Pick the most appropriate image form. Prefer an actual scene, poster,
painting, photograph, tableau, or cover over a text-heavy infographic.
- Match metaphor to content, but do not overthink it. The category is a creative
catalyst, not a semantic constraint.
- Use a polished editorial rendering direction: smooth anti-aliased
text, high contrast, clean edges, readable labels.
- Make the category drive the composition through a readable staged moment,
editorial composition, symbolic environment, route, artifact, or visual
metaphor.
- Keep the structure literal enough to aid understanding, but not so heavy that
it obscures engineering meaning.
- Give the image generator creative latitude on layout, structure, color palette,
and visual metaphors. Be precise about what content to show, loose about how
to show it.
- Do not use copyrighted characters, logos, or named fictional universes. Use
genre cues, knockoffs, and original compositions instead.
## Content-First Aesthetic Contract
The meaning must be readable and clearly hierarchical. Everything else is
creative territory: image form, layout, visual metaphors, decorative elements,
color choices, and category-specific visual language.
Hierarchy:
1. Meaning: what shipped, what changed, and why it matters must be clear.
2. If the image uses text, labels, sections, or evidence bullets, they must be
legible.
3. The selected category's visual DNA should drive the composition as a poster,
scene, painting, photograph, tableau, cover, or symbolic object arrangement.
4. Do not play it safe. A visually striking image that someone wants to look at
beats a correct but boring one.
Hard rules:
- Content sections and labels must be readable when present. Text cannot be
obscured by decorations.
- Do not use lore-heavy copy that competes with engineering or business meaning.
- Every prompt must include a clear image-first composition cue: a staged scene,
poster composition, painting, photograph, symbolic tableau, hero object,
mission room, dossier, artifact, route, or visual metaphor.
- Do not over-prescribe exact coordinates or panel geometry; give a composition
backbone and let the model compose around it.
## Generation
1. Write the final prompt to a temporary markdown file.
2. Generate with the shared image helper:
```bash
uv run --script scripts/generate_infographic.py \
--prompt-file /tmp/bm-infographic-prompt.md \
--output docs/assets/infographics/<name>.webp
```
3. Verify the image exists and is readable before reporting success.
## Quality Bar
- Tell a concrete before/after value story, not vague improvement claims.
- Stay understandable for both engineers and non-technical stakeholders.
- Use plain-language section titles and labels when text is present.
- Include clear visual hierarchy: title, staged focal point, symbolic scene,
evidence props, or hero object.
- Avoid invented facts; only use provided source material.
- Favor shipped outcomes over intermediate or reverted work.
- Preserve readability with high contrast, non-tiny labels, and uncluttered
layout.
@@ -1,7 +0,0 @@
interface:
display_name: "Infographics"
short_description: "Generate Basic Memory repo infographics"
icon_small: "./assets/icon.svg"
icon_large: "./assets/icon.svg"
brand_color: "#2563EB"
default_prompt: "Use $infographics to generate a Basic Memory PR or changelog infographic with canonical output paths."
@@ -1,5 +0,0 @@
<svg xmlns="http://www.w3.org/2000/svg" width="128" height="128" viewBox="0 0 24 24" fill="none" stroke="#111827" stroke-width="1.8" stroke-linecap="round" stroke-linejoin="round">
<path d="M3 12h4l2-6 4 12 2-6h6"/>
<path d="M4 20h16"/>
</svg>

Before

Width:  |  Height:  |  Size: 249 B

@@ -1,72 +0,0 @@
# Prompt Blueprint
Convert an evidence pack into a final visual prompt. Be precise about the
content and loose about visual execution.
## Required Inputs
- Diff truth source summary
- Changed-file orientation summary
- Impact ledger with before/after outcomes
- Discard list for excluded noise
- Chosen image form
- Chosen BM style category
## Prompt Shape
```text
Create a polished Basic Memory editorial image inspired by
<BM_STYLE_CATEGORY>. Use a poster, scene, tableau, painting, photograph, cover
image, staged artifact, or another image-first form that best communicates the
intent. Use HD editorial rendering with smooth anti-aliased text when text is
present. Go bold and let the selected category drive the visual language through
original, non-infringing cues.
TITLE:
- "<clear title>"
- "<scope subtitle>"
COMPOSITION:
- Recreate a clear staged moment or symbolic image that describes the PR
intent.
- Maps, diagrams, dossiers, route lines, labels, and artifacts can appear as
props inside the scene, but the output should read as an image rather than a
dense infographic.
- Take creative liberty with layout and styling.
- The hard rule: the meaning must be readable and clearly hierarchical.
- Keep labels plain-language and technical when labels are used.
CONTENT:
1. "<section>"
- <evidence-grounded outcome>
- <evidence-grounded outcome>
2. "<section>"
- <evidence-grounded outcome>
- <evidence-grounded outcome>
METRICS:
- <metric>
- <metric>
STYLE DIRECTION:
- Upscaled editorial, high contrast, anti-aliased text, smooth edges.
- Let the category's visual DNA drive the composition.
- Use genre/category cues only; do not use copyrighted characters, logos, named
fictional universes, direct band logos, album art, or celebrity likenesses.
DO NOT:
- Make text unreadable or let decoration obscure content.
- Render a text-heavy infographic, dashboard, flowchart, timeline strip,
checklist, bullet-list panel, or dense explanatory diagram.
- Use crunchy low-resolution pixel art.
- Invent facts not present in the evidence pack.
```
## Writing Rules
- Keep each bullet specific and evidence-grounded.
- Prefer outcome language over implementation trivia.
- Default to three or four sections; never exceed five.
- Give proportionally more space to dominant changes.
- Keep the final prompt short, energetic, and readable.
@@ -1,42 +0,0 @@
# Style Balance Rubric
## Core Principle
Be bold, not confusing. The selected BM style category should structure the
visual through a readable image-first composition, not decorate a generic grid.
Use an editorial scene, poster, painting, photograph, cover, staged artifact, or
tableau that turns the PR intent into a visual moment.
## Required Traits
- Anti-aliased typography
- Smooth edges
- High contrast between text and background
- Plain-language section labels
- Clear composition backbone: staged scene, editorial poster, painting,
photograph, symbolic tableau, hero artifact, dossier, mission room, or route
embodied as part of the scene
- A single coherent BM style category, expressed through original visual cues
## Reject Or Rewrite If
- Content text is unreadable.
- The prompt lacks a composition backbone.
- The prompt over-prescribes exact panel positions or a rigid grid.
- The style leans into crunchy low-resolution pixelation.
- Copy uses lore-heavy references instead of engineering meaning.
- The prompt uses copyrighted characters, logos, named fictional universes,
direct band logos, album art, or celebrity likenesses.
## Creative Integration Patterns
- Use category-native map details to organize content: textbook diagrams,
literary journeys, quest maps, tour posters, mission-control routes, stage
blocking, mythic constellations, star charts, mission trajectories, case
boards, or civic plans as props inside the image.
- Recreate a scene, editorial poster, painting, photograph, cover, artifact, or
tableau instead of sectioned bullets.
- Map engineering metrics to visual counters, route progress, or status boards
only when they naturally belong in the scene.
- Let headers and accents borrow from the selected style.
- Keep atmospheric details behind or around content, never over it.
-114
View File
@@ -1,114 +0,0 @@
---
name: pr-create
description: Use when creating or updating a Basic Memory pull request from Codex with BM Bossbot merge-gate monitoring.
---
# Create A Basic Memory PR
Create or update a pull request for the current branch, then wait for BM
Bossbot to approve the latest head SHA. This skill never merges a PR.
## Inputs
- Optional `<theme>`: free-form visual direction for the non-gating PR
image. Example: `$pr-create "Italian movie poster"`.
- Treat `<theme>` as style guidance only. It must not affect PR readiness,
BM Bossbot review, status checks, or merge behavior.
## How To Use
Ask Codex to use the skill from a feature branch:
```text
$pr-create
$pr-create "Italian movie poster"
$pr-create "80's action movies"
```
Use the plain form when you only want the PR workflow. Pass a theme when you
want the non-gating image to lean toward a particular visual direction. The
theme can be specific ("Rembrandt-inspired approval scene") or broad ("let the
model choose from BM categories").
## What Happens
1. Codex checks the branch, local verification, GitHub auth, commit sign-offs,
and semantic PR title shape.
2. Codex pushes the branch, creates or reuses the PR, and adds the optional
`BM_INFOGRAPHIC_THEME` block when a theme was supplied.
3. BM Bossbot runs from trusted base code, reviews sanitized PR metadata and
diff context, and sets the required `BM Bossbot Approval` status for the
exact head SHA.
4. If approval succeeds, BM Bossbot may publish a non-gating image block and a
provenance block:
```markdown
<!-- BM_INFOGRAPHIC_PROVENANCE:start -->
...
<!-- BM_INFOGRAPHIC_PROVENANCE:end -->
```
The provenance records the image mode, theme source, selected visual
direction, and image settings. It is for review/debugging context only.
5. Codex reports the PR URL, head SHA, checks watched, verification run, and BM
Bossbot verdict.
The skill never merges, never enables auto-merge, and never treats the image or
provenance block as a gate. The only required merge signal is the
`BM Bossbot Approval` status on the current PR head SHA.
## Preflight
1. Confirm the repo and branch:
- `git status --short --branch`
- stop if detached or on `main`
- keep unrelated user changes intact
2. Confirm GitHub access:
- `gh auth status`
- `gh repo view --json nameWithOwner,defaultBranchRef,url`
3. Check PR readiness:
- commits are signed off with `git commit -s`
- title uses the repo semantic format
- local verification appropriate to the change has run
## Create Or Reuse
1. Push the branch:
- `git push -u origin HEAD`
2. Check for an existing PR:
- `gh pr view --json number,url,headRefOid,mergeStateStatus,statusCheckRollup`
3. If no PR exists, create one:
- `gh pr create --fill`
- adjust the title if it does not satisfy the semantic PR title workflow
4. If `<theme>` is provided, add or update this managed block in the PR body:
```markdown
<!-- BM_INFOGRAPHIC_THEME:start -->
<theme>
<!-- BM_INFOGRAPHIC_THEME:end -->
```
Keep the rest of the PR body intact. The theme is non-gating image guidance
only.
5. Do not merge. Do not enable auto-merge.
## Watch The Gate
1. Trigger or wait for `.github/workflows/bm-bossbot.yml`.
2. Watch the required commit status named `BM Bossbot Approval`.
3. Treat approval as valid only when it is green for the current `headRefOid`.
4. If the branch changes after approval, wait for BM Bossbot to review the new
head SHA.
5. If BM Bossbot fails or requests changes, use `$fix-pr-issues`.
## Report
Return the PR URL, current head SHA, checks watched, verification run, and the
BM Bossbot verdict. Include the image `<theme>` if one was supplied. Be
explicit when any check is still pending.
@@ -1,7 +0,0 @@
interface:
display_name: "PR Create"
short_description: "Create PRs and wait for BM Bossbot"
icon_small: "./assets/icon.svg"
icon_large: "./assets/icon.svg"
brand_color: "#2563EB"
default_prompt: "Use $pr-create to create or update this Basic Memory PR and wait for BM Bossbot Approval."
-5
View File
@@ -1,5 +0,0 @@
<svg xmlns="http://www.w3.org/2000/svg" width="128" height="128" viewBox="0 0 24 24" fill="none" stroke="#111827" stroke-width="1.8" stroke-linecap="round" stroke-linejoin="round">
<path d="M3 12h4l2-6 4 12 2-6h6"/>
<path d="M4 20h16"/>
</svg>

Before

Width:  |  Height:  |  Size: 249 B

+2 -2
View File
@@ -6,14 +6,14 @@
},
"metadata": {
"description": "Official Basic Memory plugins from the canonical basic-memory repository",
"version": "0.22.0"
"version": "0.21.6"
},
"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.22.0",
"version": "0.21.6",
"author": {
"name": "Basic Machines"
},
+16 -15
View File
@@ -1,18 +1,26 @@
name: Claude Code Review
"on":
workflow_dispatch:
inputs:
pr_number:
description: Pull request number to review manually
required: true
on:
pull_request:
types: [opened, synchronize]
# Optional: Only run on specific file changes
# paths:
# - "src/**/*.ts"
# - "src/**/*.tsx"
# - "src/**/*.js"
# - "src/**/*.jsx"
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
jobs:
claude-review:
if: inputs.pr_number != ''
# Only run for organization members and collaborators
if: |
github.event.pull_request.author_association == 'OWNER' ||
github.event.pull_request.author_association == 'MEMBER' ||
github.event.pull_request.author_association == 'COLLABORATOR'
runs-on: ubuntu-latest
permissions:
contents: read
@@ -35,14 +43,7 @@ jobs:
track_progress: true # Enable visual progress tracking
allowed_bots: '*'
prompt: |
Review Basic Memory PR #${{ inputs.pr_number }} as an advisory manual review.
Use `gh pr view ${{ inputs.pr_number }}` and related `gh pr`/`gh api`
commands to inspect the pull request. Do not merge the PR and do not
treat this advisory review as the required merge gate. BM Bossbot owns
the required `BM Bossbot Approval` status.
Review the PR against our team checklist:
Review this Basic Memory PR against our team checklist:
## Code Quality & Standards
- [ ] Follows Basic Memory's coding conventions in CLAUDE.md
+8 -6
View File
@@ -13,10 +13,9 @@ env:
jobs:
docker:
runs-on: depot-ubuntu-24.04
runs-on: ubuntu-latest
permissions:
contents: read
id-token: write
packages: write
steps:
@@ -25,8 +24,10 @@ jobs:
with:
fetch-depth: 0
- name: Set up Depot
uses: depot/setup-action@v1
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
with:
platforms: linux/amd64,linux/arm64
- name: Log in to GitHub Container Registry
uses: docker/login-action@v4
@@ -48,12 +49,13 @@ jobs:
type=raw,value=latest,enable={{is_default_branch}}
- name: Build and push Docker image
uses: depot/build-push-action@v1
uses: docker/build-push-action@v7
with:
project: ${{ vars.DEPOT_BASIC_MEMORY_PROJECT_ID || vars.DEPOT_PROJECT_ID }}
context: .
file: ./Dockerfile
platforms: linux/amd64,linux/arm64
push: true
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
cache-from: type=gha
cache-to: type=gha,mode=max
+10 -123
View File
@@ -13,43 +13,9 @@ on:
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
# 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: ubuntu-latest
@@ -88,8 +54,6 @@ 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:
@@ -100,8 +64,6 @@ jobs:
python-version: "3.12"
- 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.
- os: ubuntu-latest
python-version: "3.14"
- os: windows-latest
@@ -125,19 +87,6 @@ jobs:
- uses: extractions/setup-just@v4
- name: Cache pytest-testmon results
uses: actions/cache@v4
with:
path: |
.testmondata
.testmondata-shm
.testmondata-wal
key: ${{ runner.os }}-testmon-sqlite-unit-py${{ matrix.python-version }}-${{ github.ref_name }}-${{ github.run_id }}
restore-keys: |
${{ runner.os }}-testmon-sqlite-unit-py${{ matrix.python-version }}-${{ github.ref_name }}-
${{ runner.os }}-testmon-sqlite-unit-py${{ matrix.python-version }}-main-
${{ runner.os }}-testmon-sqlite-unit-py${{ matrix.python-version }}-
- name: Create virtual env
run: |
uv venv
@@ -151,8 +100,6 @@ 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:
@@ -186,19 +133,6 @@ jobs:
- uses: extractions/setup-just@v4
- name: Cache pytest-testmon results
uses: actions/cache@v4
with:
path: |
.testmondata
.testmondata-shm
.testmondata-wal
key: ${{ runner.os }}-testmon-sqlite-integration-py${{ matrix.python-version }}-${{ github.ref_name }}-${{ github.run_id }}
restore-keys: |
${{ runner.os }}-testmon-sqlite-integration-py${{ matrix.python-version }}-${{ github.ref_name }}-
${{ runner.os }}-testmon-sqlite-integration-py${{ matrix.python-version }}-main-
${{ runner.os }}-testmon-sqlite-integration-py${{ matrix.python-version }}-
- name: Create virtual env
run: |
uv venv
@@ -212,19 +146,15 @@ jobs:
just test-int-sqlite
test-postgres-unit:
needs: changes
if: needs.changes.outputs.code == 'true'
name: Test Postgres Unit (Python ${{ matrix.python-version }}, shard ${{ matrix.group }}/3)
name: Test Postgres Unit (Python ${{ matrix.python-version }})
timeout-minutes: 60
strategy:
fail-fast: false
matrix:
# 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"]
include:
- python-version: "3.12"
- python-version: "3.13"
- python-version: "3.14"
runs-on: ubuntu-latest
services:
postgres:
@@ -260,20 +190,6 @@ jobs:
- uses: extractions/setup-just@v4
- name: Cache pytest-testmon results
uses: actions/cache@v4
with:
path: |
.testmondata
.testmondata-shm
.testmondata-wal
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 }}-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 }}-
- name: Create virtual env
run: |
uv venv
@@ -284,19 +200,18 @@ jobs:
- name: Run tests
run: |
BASIC_MEMORY_PYTEST_SPLIT_FLAGS="--splits 3 --group ${{ matrix.group }}" just test-unit-postgres
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:
# Latest Python only: SQLite carries version coverage, Postgres carries
# backend coverage.
python-version: ["3.14"]
include:
- python-version: "3.12"
- python-version: "3.13"
- python-version: "3.14"
runs-on: ubuntu-latest
services:
postgres:
@@ -332,19 +247,6 @@ jobs:
- uses: extractions/setup-just@v4
- name: Cache pytest-testmon results
uses: actions/cache@v4
with:
path: |
.testmondata
.testmondata-shm
.testmondata-wal
key: ${{ runner.os }}-testmon-postgres-integration-py${{ matrix.python-version }}-${{ github.ref_name }}-${{ github.run_id }}
restore-keys: |
${{ runner.os }}-testmon-postgres-integration-py${{ matrix.python-version }}-${{ github.ref_name }}-
${{ runner.os }}-testmon-postgres-integration-py${{ matrix.python-version }}-main-
${{ runner.os }}-testmon-postgres-integration-py${{ matrix.python-version }}-
- name: Create virtual env
run: |
uv venv
@@ -358,8 +260,6 @@ 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: ubuntu-latest
@@ -381,19 +281,6 @@ jobs:
- uses: extractions/setup-just@v4
- name: Cache pytest-testmon results
uses: actions/cache@v4
with:
path: |
.testmondata
.testmondata-shm
.testmondata-wal
key: ${{ runner.os }}-testmon-semantic-py3.12-${{ github.ref_name }}-${{ github.run_id }}
restore-keys: |
${{ runner.os }}-testmon-semantic-py3.12-${{ github.ref_name }}-
${{ runner.os }}-testmon-semantic-py3.12-main-
${{ runner.os }}-testmon-semantic-py3.12-
- name: Create virtual env
run: |
uv venv
+1 -1
View File
@@ -49,7 +49,7 @@ ENV/
/docs/.obsidian/
/examples/.obsidian/
/examples/.basic-memory/
/docs/assets
# claude action
claude-output
+3 -3
View File
@@ -133,9 +133,9 @@ short version for agents:
- **Full file read before edits**: Before editing any file, read it in full first to ensure complete context; partial reads lead to corrupted edits
- **Minimize diffs**: Prefer the smallest change that satisfies the request. Avoid unrelated refactors or style rewrites unless necessary for correctness
- **House style is canonical**: Follow the Programming Style section above for type-safe,
fail-fast code; do not hide unclear models with speculative attributes, broad exception
handling, casts, or unapproved fallback logic
- **No speculative getattr**: Never use `getattr(obj, "attr", default)` when unsure about attribute names. Check the class definition or source code first
- **Fail fast**: Write code with fail-fast logic by default. Do not swallow exceptions with errors or warnings
- **No fallback logic**: Do not add fallback logic unless explicitly told to and agreed with the user
- **No guessing**: Do not say "The issue is..." before you actually know what the issue is. Investigate first.
### Literate Programming Style
-65
View File
@@ -1,70 +1,5 @@
# 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
+4 -4
View File
@@ -216,14 +216,14 @@ Source: [`plugins/claude-code`](plugins/claude-code).
### Shared skills
Framework-agnostic `SKILL.md` files live in [`skills/`](skills). If your
Skills CLI supports repository subdirectory sources:
Skills CLI supports subpath installs:
```bash
npx skills add basicmachines-co/basic-memory/skills
npx skills add basicmachines-co/basic-memory --path skills
```
If your installed Skills CLI cannot load that source, update the CLI or copy
the `memory-*` directories from `skills/` into your agent's skills directory.
If it does not, copy the `memory-*` directories from `skills/` into your
agent's skills directory as a temporary Phase 1 install path.
### Hermes
+3 -5
View File
@@ -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 only | Verify mirror integrity (no changes) |
| `bm cloud check` | — | Personal + Team | Verify files match (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,12 +459,10 @@ bm cloud bisync --name research
- You edit in multiple places
- You want automatic conflict resolution
### Verify Sync Integrity (Personal only)
### Verify Sync Integrity
**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
```
@@ -1001,7 +999,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 - Personal workspaces only
# Integrity check
bm cloud check --name <project>
bm cloud check --name <project> --one-way
-7
View File
@@ -1,7 +0,0 @@
{
"$schema": "https://glama.ai/mcp/schemas/server.json",
"maintainers": [
"phernandez",
"groksrc"
]
}
+1 -1
View File
@@ -38,7 +38,7 @@ from typing import Any, Callable
from agent.memory_provider import MemoryProvider
from tools.registry import tool_error
__version__ = "0.22.0"
__version__ = "0.21.6"
logger = logging.getLogger("hermes.memory.basic-memory")
+1 -1
View File
@@ -1,5 +1,5 @@
name: basic-memory
version: 0.22.0
version: 0.21.6
description: "Basic Memory — persistent knowledge graph backed by the basic-memory MCP server"
pip_dependencies:
- mcp
+1 -1
View File
@@ -150,7 +150,7 @@ This plugin ships with workflow-oriented skills that are automatically loaded wh
No manual installation needed. To update skills or install new ones as they become available:
```bash
npx skills add basicmachines-co/basic-memory/skills --agent openclaw
npx skills add basicmachines-co/basic-memory --path skills --agent openclaw
```
See the canonical source at [`basic-memory/skills`](../../skills).
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@basicmemory/openclaw-basic-memory",
"version": "0.22.0",
"version": "0.21.6",
"type": "module",
"main": "./dist/index.js",
"types": "./dist/index.d.ts",
+24 -51
View File
@@ -1,12 +1,5 @@
# Basic Memory - Modern Command Runner
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:
uv sync
@@ -42,60 +35,40 @@ test-sqlite: test-unit-sqlite test-int-sqlite
test-postgres: test-unit-postgres test-int-postgres
# Run unit tests against SQLite
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
test-unit-sqlite:
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov 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
#!/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
test-unit-postgres:
BASIC_MEMORY_ENV=test BASIC_MEMORY_TEST_POSTGRES=1 uv run pytest -p pytest_mock -v --no-cov tests
# Run integration tests against SQLite (excludes semantic tests and on-demand benchmarks —
# use just test-semantic / run benchmark files explicitly)
test-int-sqlite: testmon-seed
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov {{TESTMON_FLAGS}} --testmon-env=int-sqlite -m "not semantic and not benchmark" test-int
# Run integration tests against SQLite (excludes semantic benchmarks — use just test-semantic)
test-int-sqlite:
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov -m "not semantic" test-int
# Run integration tests against Postgres
# Note: Uses timeout due to FastMCP Client + asyncpg cleanup hang (tests pass, process hangs on exit)
# See: https://github.com/jlowin/fastmcp/issues/1311
test-int-postgres: testmon-seed
test-int-postgres:
#!/usr/bin/env bash
set -euo pipefail
# Use gtimeout (macOS/Homebrew) or timeout (Linux)
TIMEOUT_CMD=$(command -v gtimeout || command -v timeout || echo "")
if [[ -n "$TIMEOUT_CMD" ]]; then
$TIMEOUT_CMD --signal=KILL 600 bash -c 'BASIC_MEMORY_ENV=test BASIC_MEMORY_TEST_POSTGRES=1 uv run pytest -p pytest_mock -v --no-cov {{TESTMON_FLAGS}} --testmon-env=int-postgres -m "not semantic and not benchmark" test-int' || test $? -eq 137
$TIMEOUT_CMD --signal=KILL 600 bash -c 'BASIC_MEMORY_ENV=test BASIC_MEMORY_TEST_POSTGRES=1 uv run pytest -p pytest_mock -v --no-cov -m "not semantic" test-int' || test $? -eq 137
else
echo "⚠️ No timeout command found, running without timeout..."
BASIC_MEMORY_ENV=test BASIC_MEMORY_TEST_POSTGRES=1 uv run pytest -p pytest_mock -v --no-cov {{TESTMON_FLAGS}} --testmon-env=int-postgres -m "not semantic and not benchmark" test-int
BASIC_MEMORY_ENV=test BASIC_MEMORY_TEST_POSTGRES=1 uv run pytest -p pytest_mock -v --no-cov -m "not semantic" test-int
fi
# Run tests impacted by recent changes (requires pytest-testmon)
# Pass paths or node ids after `just testmon` to limit the candidate set further.
testmon *args: testmon-seed
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov {{TESTMON_SELECT_FLAGS}} --testmon-env=local {{args}}
# Seed pytest-testmon data into this worktree from the shared Git cache.
testmon-seed:
uv run python scripts/testmon_cache.py seed
# Refresh the shared pytest-testmon cache from a full backend test run.
testmon-refresh:
#!/usr/bin/env bash
set -euo pipefail
BASIC_MEMORY_TESTMON_FLAGS="{{TESTMON_REFRESH_FLAGS}}" just test
uv run python scripts/testmon_cache.py refresh
# Show local and shared pytest-testmon cache locations.
testmon-status:
uv run python scripts/testmon_cache.py status
testmon *args:
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov --testmon {{args}}
# Run MCP smoke test (fast end-to-end loop)
test-smoke: testmon-seed
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov {{TESTMON_FLAGS}} --testmon-env=smoke -m smoke test-int/mcp/test_smoke_integration.py
test-smoke:
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov -m smoke test-int/mcp/test_smoke_integration.py
# Fast local loop: lint, format, typecheck, impacted tests via pytest-testmon
fast-check:
@@ -124,18 +97,18 @@ postgres-migrate:
# Run Windows-specific tests only (only works on Windows platform)
# These tests verify Windows-specific database optimizations (locking mode, NullPool)
# Will be skipped automatically on non-Windows platforms
test-windows: testmon-seed
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov {{TESTMON_FLAGS}} --testmon-env=windows -m windows tests test-int
test-windows:
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov -m windows tests test-int
# Run benchmark tests only (performance testing)
# These are slow tests that measure sync performance with various file counts
# Excluded from default test runs to keep CI fast
test-benchmark: testmon-seed
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov {{TESTMON_FLAGS}} --testmon-env=benchmark -m benchmark tests test-int
test-benchmark:
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov -m benchmark tests test-int
# Run semantic search quality benchmarks (all combos)
test-semantic: testmon-seed
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov {{TESTMON_FLAGS}} --testmon-env=semantic -m semantic test-int/semantic/
test-semantic:
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov -m semantic test-int/semantic/
# Run semantic benchmarks with JSON artifact output, then show report
test-semantic-report:
@@ -147,8 +120,8 @@ test-litellm-live *args:
BASIC_MEMORY_ENV=test BASIC_MEMORY_RUN_LITELLM_INTEGRATION=1 PYTHONPATH=test-int:src uv run python -m semantic.litellm_live_harness {{args}}
# Run semantic benchmarks (Postgres combos only)
test-semantic-postgres: testmon-seed
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov {{TESTMON_FLAGS}} --testmon-env=semantic-postgres -m semantic -k postgres test-int/semantic/
test-semantic-postgres:
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov -m semantic -k postgres test-int/semantic/
# View semantic benchmark results (rich formatted table)
# Usage: just semantic-report [--filter-combo sqlite] [--filter-suite paraphrase] [--sort-by avg_latency_ms]
@@ -164,8 +137,8 @@ benchmark-compare baseline candidate *args:
# Run all tests including Windows, Postgres, and Benchmarks (for CI/comprehensive testing)
# Use this before releasing to ensure everything works across all backends and platforms
test-all: testmon-seed
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov {{TESTMON_FLAGS}} --testmon-env=all tests test-int
test-all:
BASIC_MEMORY_ENV=test uv run pytest -p pytest_mock -v --no-cov tests test-int
# Generate HTML coverage report
coverage:
@@ -6,14 +6,14 @@
},
"metadata": {
"description": "Official plugins from Basic Machines for knowledge management and AI-assisted development",
"version": "0.22.0"
"version": "0.21.6"
},
"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.22.0",
"version": "0.21.6",
"author": {
"name": "Basic Machines"
},
@@ -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.22.0",
"version": "0.21.6",
"author": {
"name": "Basic Machines"
},
+1 -1
View File
@@ -24,7 +24,7 @@ Memory's durable graph**, rather than a memory layer of its own. See
configures the project for the plugin: maps it to a Basic Memory project (picking
an existing one or creating a new one), seeds the `session`/`decision`/`task`
schemas into the project, installs the shared `memory-*` skills via
`npx skills add basicmachines-co/basic-memory/skills` (the plugin doesn't
`npx skills add basicmachines-co/basic-memory --path skills` (the plugin doesn't
vendor its own copies — `skills/` is the single source of truth, shared with
OpenClaw), optionally learns the project's placement conventions, and enables the
capture reflexes. Writes the `basicMemory` block to
+1 -1
View File
@@ -156,7 +156,7 @@ git ls-files skills/ | grep -q memory- && echo "source repo - skip install"
Otherwise, run from the project root:
```
npx skills add basicmachines-co/basic-memory/skills
npx skills add basicmachines-co/basic-memory --path skills
```
This installs the canonical `memory-*` skills into the user's skills directory — the
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "codex",
"version": "0.22.0",
"version": "0.21.6",
"description": "A Codex-native bridge to Basic Memory for durable engineering context, decisions, and resumable checkpoints.",
"author": {
"name": "Basic Machines",
-6
View File
@@ -76,10 +76,6 @@ addopts = "--cov=basic_memory --cov-report term-missing"
testpaths = ["tests", "test-int"]
asyncio_mode = "strict"
asyncio_default_fixture_loop_scope = "function"
# Any test hanging >120s fails with a stack dump instead of stalling the CI job
# until the runner times out (the FastMCP/asyncpg cleanup-hang family).
timeout = 120
timeout_method = "thread"
filterwarnings = [
"ignore:The @wait_container_is_ready decorator is deprecated.*:DeprecationWarning:testcontainers\\.core\\.waiting_utils",
"ignore:The default datetime adapter is deprecated as of Python 3\\.12.*:DeprecationWarning:aiosqlite\\.core",
@@ -119,8 +115,6 @@ dev = [
"ty>=0.0.18",
"cst-lsp>=0.1.3",
"libcst>=1.8.6",
"pytest-timeout>=2.4.0",
"pytest-split>=0.11.0",
]
[tool.hatch.version]
-225
View File
@@ -1,225 +0,0 @@
#!/usr/bin/env python3
"""Seed and refresh shared pytest-testmon data for Git worktrees."""
from __future__ import annotations
import argparse
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import NamedTuple
TESTMON_FILENAMES = (".testmondata", ".testmondata-shm", ".testmondata-wal")
TESTMON_CACHE_ENV = "BM_TESTMON_CACHE_DIR"
class TestmonCacheResult(NamedTuple):
status: str
source_dir: Path
destination_dir: Path
copied: tuple[Path, ...]
def _run_git(args: list[str], cwd: Path) -> str:
return subprocess.check_output(["git", *args], cwd=cwd, text=True).strip()
def resolve_repo_root(repo_root: Path | None = None) -> Path:
if repo_root is not None:
return repo_root.expanduser().resolve()
return Path(_run_git(["rev-parse", "--show-toplevel"], Path.cwd())).resolve()
def resolve_cache_dir(repo_root: Path, cache_dir: Path | None = None) -> Path:
if cache_dir is not None:
return cache_dir.expanduser().resolve()
if env_cache_dir := os.environ.get(TESTMON_CACHE_ENV):
return Path(env_cache_dir).expanduser().resolve()
git_common_dir = Path(_run_git(["rev-parse", "--git-common-dir"], repo_root))
if not git_common_dir.is_absolute():
git_common_dir = repo_root / git_common_dir
return git_common_dir.resolve() / "testmon-cache" / "main"
def _testmon_datafile(directory: Path) -> Path:
return directory / ".testmondata"
def _testmon_files(directory: Path) -> list[Path]:
return [
directory / filename for filename in TESTMON_FILENAMES if (directory / filename).is_file()
]
def _remove_path(path: Path) -> None:
if path.is_dir():
shutil.rmtree(path)
elif path.exists():
path.unlink()
def _copy_testmon_files(source_dir: Path, destination_dir: Path) -> tuple[Path, ...]:
destination_dir.mkdir(parents=True, exist_ok=True)
copied: list[Path] = []
for source in _testmon_files(source_dir):
destination = destination_dir / source.name
shutil.copy2(source, destination)
copied.append(destination)
return tuple(copied)
def seed_testmon_data(repo_root: Path, cache_dir: Path) -> TestmonCacheResult:
local_datafile = _testmon_datafile(repo_root)
shared_datafile = _testmon_datafile(cache_dir)
if local_datafile.exists():
return TestmonCacheResult(
status="exists",
source_dir=cache_dir,
destination_dir=repo_root,
copied=(),
)
if not shared_datafile.exists():
return TestmonCacheResult(
status="missing",
source_dir=cache_dir,
destination_dir=repo_root,
copied=(),
)
# A worktree with sidecars but no main database is stale; replace the set
# together so SQLite never sees a mixed local/cache snapshot.
for filename in TESTMON_FILENAMES:
_remove_path(repo_root / filename)
copied = _copy_testmon_files(cache_dir, repo_root)
return TestmonCacheResult(
status="seeded",
source_dir=cache_dir,
destination_dir=repo_root,
copied=copied,
)
def refresh_testmon_data(repo_root: Path, cache_dir: Path) -> TestmonCacheResult:
local_datafile = _testmon_datafile(repo_root)
if not local_datafile.exists():
raise FileNotFoundError(
f"No local pytest-testmon data at {local_datafile}; run tests first."
)
cache_parent = cache_dir.parent
cache_parent.mkdir(parents=True, exist_ok=True)
temp_dir = Path(tempfile.mkdtemp(prefix=f".{cache_dir.name}.", dir=cache_parent))
backup_dir = cache_parent / f".{cache_dir.name}.previous-{os.getpid()}"
copied: tuple[Path, ...] = ()
try:
copied = _copy_testmon_files(repo_root, temp_dir)
_remove_path(backup_dir)
if cache_dir.exists():
cache_dir.rename(backup_dir)
try:
temp_dir.rename(cache_dir)
except Exception:
if backup_dir.exists() and not cache_dir.exists():
backup_dir.rename(cache_dir)
raise
finally:
_remove_path(temp_dir)
_remove_path(backup_dir)
return TestmonCacheResult(
status="refreshed",
source_dir=repo_root,
destination_dir=cache_dir,
copied=tuple(cache_dir / path.name for path in copied),
)
def _print_seed_result(result: TestmonCacheResult) -> None:
if result.status == "seeded":
print(f"Seeded pytest-testmon data from {result.source_dir} into {result.destination_dir}")
elif result.status == "exists":
print(f"Local pytest-testmon data already exists at {result.destination_dir}")
elif result.status == "missing":
print(
f"No shared pytest-testmon baseline at {result.source_dir}; "
"run `just testmon-refresh` after a full backend test run to create one."
)
else:
raise ValueError(f"Unexpected seed result: {result.status}")
def _print_refresh_result(result: TestmonCacheResult) -> None:
print(f"Published pytest-testmon data from {result.source_dir} to {result.destination_dir}")
def _print_status(repo_root: Path, cache_dir: Path) -> None:
print(f"Repo root: {repo_root}")
print(f"Worktree data: {_testmon_datafile(repo_root)}")
print(f"Shared cache: {_testmon_datafile(cache_dir)}")
print(f"Worktree ready: {_testmon_datafile(repo_root).exists()}")
print(f"Cache ready: {_testmon_datafile(cache_dir).exists()}")
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--repo-root",
type=Path,
help="Repository root to operate on (default: git rev-parse --show-toplevel)",
)
parser.add_argument(
"--cache-dir",
type=Path,
help=(
"Shared testmon cache directory "
f"(default: ${TESTMON_CACHE_ENV} or <git-common-dir>/testmon-cache/main)"
),
)
subparsers = parser.add_subparsers(dest="command", required=True)
subparsers.add_parser("seed", help="Copy shared testmon data into this worktree if missing")
subparsers.add_parser("refresh", help="Publish this worktree's testmon data to the cache")
subparsers.add_parser("status", help="Show local and shared testmon data paths")
return parser
def main(argv: list[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
repo_root = resolve_repo_root(args.repo_root)
cache_dir = resolve_cache_dir(repo_root, args.cache_dir)
if args.command == "seed":
_print_seed_result(seed_testmon_data(repo_root=repo_root, cache_dir=cache_dir))
return 0
if args.command == "refresh":
_print_refresh_result(refresh_testmon_data(repo_root=repo_root, cache_dir=cache_dir))
return 0
if args.command == "status":
_print_status(repo_root=repo_root, cache_dir=cache_dir)
return 0
parser.error(f"Unknown command: {args.command}")
return 2
if __name__ == "__main__":
sys.exit(main())
+2 -2
View File
@@ -6,12 +6,12 @@
"url": "https://github.com/basicmachines-co/basic-memory.git",
"source": "github"
},
"version": "0.22.0",
"version": "0.21.6",
"packages": [
{
"registryType": "pypi",
"identifier": "basic-memory",
"version": "0.22.0",
"version": "0.21.6",
"runtimeHint": "uvx",
"runtimeArguments": [
{
+3 -3
View File
@@ -54,13 +54,13 @@ Users can install or update skills with the [Skills CLI](https://github.com/verc
```bash
# Install all skills
npx skills add basicmachines-co/basic-memory/skills
npx skills add basicmachines-co/basic-memory --path skills
# Install a specific skill
npx skills add basicmachines-co/basic-memory/skills --skill memory-tasks
npx skills add basicmachines-co/basic-memory --path skills --skill memory-tasks
# Install for a specific agent
npx skills add basicmachines-co/basic-memory/skills --agent claude
npx skills add basicmachines-co/basic-memory --path skills --agent claude
```
## Adding a New Skill
+5 -5
View File
@@ -47,16 +47,16 @@ Install or update skills using the [Skills CLI](https://github.com/vercel-labs/s
```bash
# Install all skills
npx skills add basicmachines-co/basic-memory/skills
npx skills add basicmachines-co/basic-memory --path skills
# Install a specific skill
npx skills add basicmachines-co/basic-memory/skills --skill memory-tasks
npx skills add basicmachines-co/basic-memory --path skills --skill memory-tasks
# Install all skills for a specific agent
npx skills add basicmachines-co/basic-memory/skills --agent claude
npx skills add basicmachines-co/basic-memory --path skills --agent claude
# List available skills without installing
npx skills add basicmachines-co/basic-memory/skills --list
npx skills add basicmachines-co/basic-memory --path skills --list
# Check for updates
npx skills check
@@ -67,7 +67,7 @@ npx skills update
Skills are installed to your agent's skills directory (e.g., `~/.claude/skills/` for Claude Code global, or `.claude/skills/` for project-scoped).
If your installed Skills CLI cannot load `basicmachines-co/basic-memory/skills`, update the CLI or copy the `memory-*` directories manually.
If your installed Skills CLI does not support `--path`, copy the `memory-*` directories manually for now. Phase 2 will add a first-class Codex/package install path.
### Claude Desktop (claude.ai)
+4 -1
View File
@@ -1,4 +1,7 @@
"""basic-memory - Local-first knowledge management combining Zettelkasten with knowledge graphs"""
# Package version - updated by release automation
__version__ = "0.22.0"
__version__ = "0.21.6"
# API version for FastAPI - independent of package version
__api_version__ = "v0"
@@ -10,18 +10,10 @@ 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,
@@ -32,7 +24,6 @@ from basic_memory.deps import (
EntityRepositoryV2ExternalDep,
RelationRepositoryV2ExternalDep,
ProjectExternalIdPathDep,
SyncServiceV2ExternalDep,
TaskSchedulerDep,
)
from basic_memory.schemas import DeleteEntitiesResponse
@@ -49,10 +40,8 @@ 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"])
@@ -247,187 +236,6 @@ 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
@@ -32,33 +32,14 @@ def _rclone_exclude_filters(pattern: str) -> list[str]:
return [f"- {path_pattern}", f"- {path_pattern}/**"]
def _workspace_id_header(workspace_id: str | None) -> dict[str, str]:
"""Header that routes a /tenant/mount/* request to a specific tenant.
The mount endpoints resolve the workspace from X-Workspace-ID (validating
membership + subscription) and fall back to the user's default tenant when
it is absent so omitting it preserves the original default-tenant behavior.
"""
return {"X-Workspace-ID": workspace_id} if workspace_id else {}
async def get_mount_info(*, workspace_id: str | None = None) -> TenantMountInfo:
"""Get tenant mount info (bucket name + tenant id) from the cloud API.
Args:
workspace_id: Tenant id of the target workspace. When omitted, the API
uses the authenticated user's default tenant.
"""
async def get_mount_info() -> TenantMountInfo:
"""Get current tenant information from cloud API."""
try:
config_manager = ConfigManager()
config = config_manager.config
host_url = config.cloud_host.rstrip("/")
response = await make_api_request(
method="GET",
url=f"{host_url}/tenant/mount/info",
headers=_workspace_id_header(workspace_id),
)
response = await make_api_request(method="GET", url=f"{host_url}/tenant/mount/info")
return TenantMountInfo.model_validate(response.json())
except Exception as e:
@@ -66,24 +47,13 @@ async def get_mount_info(*, workspace_id: str | None = None) -> TenantMountInfo:
async def generate_mount_credentials(tenant_id: str) -> MountCredentials:
"""Generate scoped S3 credentials for syncing a specific tenant's bucket.
Args:
tenant_id: Tenant id whose bucket-scoped credentials to mint. Routed via
X-Workspace-ID so team workspaces get their own bucket's credentials.
"""
"""Generate scoped credentials for syncing."""
try:
config_manager = ConfigManager()
config = config_manager.config
host_url = config.cloud_host.rstrip("/")
# The mount endpoints resolve X-Workspace-ID by matching the workspace's
# tenant_id, so passing a tenant_id here is the correct routing key.
response = await make_api_request(
method="POST",
url=f"{host_url}/tenant/mount/credentials",
headers=_workspace_id_header(tenant_id),
)
response = await make_api_request(method="POST", url=f"{host_url}/tenant/mount/credentials")
return MountCredentials.model_validate(response.json())
except Exception as e:
@@ -26,52 +26,15 @@ from basic_memory.cli.commands.cloud.bisync_commands import (
generate_mount_credentials,
get_mount_info,
)
from basic_memory.cli.commands.cloud.rclone_config import (
configure_rclone_remote,
rclone_remote_exists,
remote_name_for_workspace,
)
from basic_memory.cli.commands.cloud.rclone_config import configure_rclone_remote
from basic_memory.cli.commands.cloud.rclone_installer import (
RcloneInstallError,
install_rclone,
)
from basic_memory.mcp.project_context import get_available_workspaces
from basic_memory.schemas.cloud import (
WorkspaceInfo,
format_workspace_choices,
format_workspace_selection_choices,
workspace_matches_exact_identifier,
)
console = Console()
def _resolve_setup_workspace(identifier: str) -> WorkspaceInfo:
"""Resolve a workspace identifier (slug, name, or tenant_id) for setup.
Errors with copyable choices when the identifier matches zero or multiple
workspaces, so the user can disambiguate.
"""
workspaces = run_with_cleanup(get_available_workspaces())
if not workspaces:
console.print("[red]No accessible cloud workspaces found for this account[/red]")
raise typer.Exit(1)
matches = [ws for ws in workspaces if workspace_matches_exact_identifier(ws, identifier)]
if len(matches) == 1:
return matches[0]
if not matches:
console.print(f"[red]No workspace matches '{identifier}'[/red]")
console.print("\nAvailable workspaces:")
console.print(format_workspace_choices(workspaces))
else:
console.print(f"[red]'{identifier}' matches multiple workspaces[/red]")
console.print("\nDisambiguate with the workspace slug or tenant_id:")
console.print(format_workspace_selection_choices(matches))
raise typer.Exit(1)
@cloud_app.command()
def login():
"""Authenticate with WorkOS using OAuth Device Authorization flow."""
@@ -201,28 +164,13 @@ def status() -> None:
@cloud_app.command("setup")
def setup(
workspace: str | None = typer.Option(
None,
"--workspace",
help="Set up sync for a specific workspace (slug, name, or tenant_id). "
"Omit for your default workspace.",
),
force: bool = typer.Option(
False,
"--force",
help="Reconfigure an rclone remote that already exists (mints new credentials).",
),
) -> None:
def setup() -> None:
"""Set up cloud sync by installing rclone and configuring credentials.
Run once per workspace you sync. The default workspace uses the
'basic-memory-cloud' remote; other (e.g. Team) workspaces each get their own
tenant-scoped remote, since Tigris credentials are bucket-scoped.
After setup, use the cloud sync commands:
bm cloud pull --name <name> # fetch cloud changes (Team-safe)
bm cloud push --name <name> # upload local changes (Team-safe)
After setup, use project commands for syncing:
bm project add <name> --cloud --local-path ~/projects/<name>
bm project bisync --name <name> --resync # First time
bm project bisync --name <name> # Subsequent syncs
"""
console.print("[bold blue]Basic Memory Cloud Setup[/bold blue]")
console.print("Setting up cloud sync with rclone...\n")
@@ -232,60 +180,35 @@ def setup(
console.print("[blue]Step 1: Installing rclone...[/blue]")
install_rclone()
# --- Resolve target workspace ---
# Trigger: --workspace given. Why: Tigris keys are tenant-scoped, so a
# non-default workspace needs its own bucket + remote. Outcome: scope the
# mount-info/credentials calls and name the remote after the workspace.
if workspace is not None:
target = _resolve_setup_workspace(workspace)
workspace_id: str | None = target.tenant_id
remote_name = remote_name_for_workspace(target.slug, is_default=target.is_default)
console.print(f"[dim]Workspace: {target.name} ({target.slug})[/dim]")
else:
workspace_id = None # default tenant
remote_name = remote_name_for_workspace(None, is_default=True)
# Trigger: the target rclone remote already exists.
# Why: re-running setup mints new credentials and overwrites the remote,
# which would silently repoint it — e.g. clobbering the shared
# basic-memory-cloud remote that served another tenant. Checked BEFORE
# minting so an abort wastes no credentials.
# Outcome: stop unless the user explicitly opts in with --force.
if rclone_remote_exists(remote_name) and not force:
console.print(f"[red]rclone remote '{remote_name}' is already configured.[/red]")
console.print(
"Re-running setup mints new credentials and overwrites it. "
"Pass --force to reconfigure."
)
raise typer.Exit(1)
# Step 2: Get tenant info (scoped to the target workspace when given)
# Step 2: Get tenant info
console.print("\n[blue]Step 2: Getting tenant information...[/blue]")
tenant_info = run_with_cleanup(get_mount_info(workspace_id=workspace_id))
tenant_info = run_with_cleanup(get_mount_info())
console.print(f"[green]Found tenant: {tenant_info.tenant_id}[/green]")
# Step 3: Generate credentials for that tenant's bucket
# Step 3: Generate credentials
console.print("\n[blue]Step 3: Generating sync credentials...[/blue]")
creds = run_with_cleanup(generate_mount_credentials(tenant_info.tenant_id))
console.print("[green]Generated secure credentials[/green]")
# Step 4: Configure the tenant's rclone remote
# Step 4: Configure rclone remote
console.print("\n[blue]Step 4: Configuring rclone remote...[/blue]")
configure_rclone_remote(
access_key=creds.access_key,
secret_key=creds.secret_key,
remote_name=remote_name,
)
console.print("\n[bold green]Cloud setup completed successfully![/bold green]")
console.print("\n[bold]Next steps:[/bold]")
console.print("1. Configure sync for a project:")
console.print("1. Add a project with local sync path:")
console.print(" bm project add research --cloud --local-path ~/Documents/research")
console.print("\n Or configure sync for an existing project:")
console.print(" bm cloud sync-setup research ~/Documents/research")
console.print("\n2. Preview a pull (recommended):")
console.print(" bm cloud pull --name research --dry-run")
console.print("\n3. Fetch cloud changes / upload local changes:")
console.print(" bm cloud pull --name research")
console.print(" bm cloud push --name research")
console.print("\n2. Preview the initial sync (recommended):")
console.print(" bm project bisync --name research --resync --dry-run")
console.print("\n3. If all looks good, run the actual sync:")
console.print(" bm project bisync --name research --resync")
console.print("\n4. Subsequent syncs (no --resync needed):")
console.print(" bm project bisync --name research")
console.print(
"\n[dim]Tip: Always use --dry-run first to preview changes before syncing[/dim]"
)
@@ -293,8 +216,6 @@ def setup(
except (RcloneInstallError, BisyncError, CloudAPIError) as e:
console.print(f"\n[red]Setup failed: {e}[/red]")
raise typer.Exit(1)
except typer.Exit:
raise
except Exception as e:
console.print(f"\n[red]Unexpected error during setup: {e}[/red]")
raise typer.Exit(1)
@@ -26,23 +26,13 @@ from basic_memory.cli.commands.cloud.rclone_commands import (
project_sync,
project_transfer,
)
from basic_memory.cli.commands.cloud.rclone_config import (
DEFAULT_RCLONE_REMOTE,
rclone_remote_exists,
remote_name_for_workspace,
)
from basic_memory.cli.commands.command_utils import run_with_cleanup
from basic_memory.cli.commands.routing import force_routing
from basic_memory.config import BasicMemoryConfig, ConfigManager, ProjectEntry
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.clients import ProjectClient
from basic_memory.mcp.project_context import get_available_workspaces
from basic_memory.schemas.cloud import (
WorkspaceInfo,
format_workspace_choices,
format_workspace_selection_choices,
workspace_matches_exact_identifier,
)
from basic_memory.schemas.cloud import WorkspaceInfo
from basic_memory.schemas.project_info import ProjectItem
from basic_memory.utils import generate_permalink, normalize_project_path
@@ -99,41 +89,12 @@ def _require_cloud_credentials(config: BasicMemoryConfig) -> None:
raise typer.Exit(1)
async def _get_workspace_for_project(
name: str,
config: BasicMemoryConfig,
*,
workspace_override: str | None = None,
) -> WorkspaceInfo:
"""Resolve the cloud workspace targeted by a project-scoped sync command.
``workspace_override`` (a slug, name, or tenant_id, e.g. from ``--workspace``)
takes precedence over config, letting the user disambiguate a project name
that exists in more than one workspace.
"""
async def _get_workspace_for_project(name: str, config: BasicMemoryConfig) -> WorkspaceInfo:
"""Resolve the cloud workspace targeted by a project-scoped sync command."""
workspaces = await get_available_workspaces()
if not workspaces:
raise ValueError("No accessible cloud workspaces found for this account")
# An explicit override wins over config — this is how the user disambiguates.
if workspace_override is not None:
matches = [
item
for item in workspaces
if workspace_matches_exact_identifier(item, workspace_override)
]
if len(matches) == 1:
return matches[0]
if not matches:
raise ValueError(
f"No accessible workspace matches '{workspace_override}'.\n"
f"{format_workspace_choices(workspaces)}"
)
raise ValueError(
f"'{workspace_override}' matches multiple workspaces; use a slug or tenant_id:\n"
f"{format_workspace_selection_choices(matches)}"
)
entry = config.projects.get(name)
workspace_id = entry.workspace_id if entry and entry.workspace_id else config.default_workspace
if workspace_id:
@@ -186,15 +147,9 @@ def _require_personal_workspace(
return workspace
async def _get_cloud_project(name: str, *, workspace_id: str | None = None) -> ProjectItem | None:
"""Fetch a project by name from the cloud API.
``workspace_id`` routes the lookup to a specific tenant so the project
metadata comes from the same workspace the transfer targets (otherwise
get_client would resolve the workspace from config/default and could read a
different tenant see #920 review).
"""
async with get_client(project_name=name, workspace=workspace_id) as client:
async def _get_cloud_project(name: str) -> ProjectItem | None:
"""Fetch a project by name from the cloud API."""
async with get_client(project_name=name) as client:
projects_list = await ProjectClient(client).list_projects()
for proj in projects_list.projects:
if generate_permalink(proj.name) == generate_permalink(name):
@@ -203,17 +158,10 @@ async def _get_cloud_project(name: str, *, workspace_id: str | None = None) -> P
def _get_sync_project(
name: str,
config: BasicMemoryConfig,
project_data: ProjectItem,
*,
remote_name: str = DEFAULT_RCLONE_REMOTE,
name: str, config: BasicMemoryConfig, project_data: ProjectItem
) -> tuple[SyncProject, str | None]:
"""Build a SyncProject and resolve local_sync_path from config.
``remote_name`` selects which tenant-scoped rclone remote the project routes
through (default tenant vs a team workspace remote).
Returns (sync_project, local_sync_path). Exits if no local_sync_path configured.
"""
sync_entry = config.projects.get(name)
@@ -229,7 +177,6 @@ def _get_sync_project(
name=project_data.name,
path=normalize_project_path(project_data.path),
local_sync_path=local_sync_path,
remote_name=remote_name,
)
return sync_project, local_sync_path
@@ -258,10 +205,7 @@ def sync_project_command(
_require_personal_workspace(name, config, unsupported_message=TEAM_WORKSPACE_SYNC_UNSUPPORTED)
try:
# Get tenant info for bucket name.
# TODO(#919): scope to the project's workspace like push/pull. Safe for now
# because these mirror commands are gated to the (default-tenant) Personal
# workspace, so the default mount info is correct.
# Get tenant info for bucket name
tenant_info = run_with_cleanup(get_mount_info())
bucket_name = tenant_info.bucket_name
@@ -315,64 +259,29 @@ def _run_directional_transfer(
on_conflict: ConflictStrategy,
dry_run: bool,
verbose: bool,
workspace: str | None = None,
) -> None:
"""Shared orchestration for `bm cloud push` / `bm cloud pull`.
Detects conflicts first, then aborts (the default) or applies the chosen
resolution. Uses additive `rclone copy`, so it never deletes on the
destination safe for Team workspaces and therefore not gated.
Routes through the resolved workspace's own tenant-scoped rclone remote, so a
Team project reads/writes the right bucket (see #919).
"""
config = ConfigManager().config
_require_cloud_credentials(config)
try:
# --- Resolve the target workspace and its tenant-scoped remote ---
# Tigris credentials are bucket/tenant-scoped, so each workspace has its
# own rclone remote. Resolve which workspace this project belongs to
# (config or --workspace override) before touching any bucket.
try:
target_workspace = run_with_cleanup(
_get_workspace_for_project(name, config, workspace_override=workspace)
)
except Exception as exc:
console.print(f"[red]Error resolving workspace for project '{name}': {exc}[/red]")
raise typer.Exit(1)
remote_name = remote_name_for_workspace(
target_workspace.slug, is_default=target_workspace.is_default
)
# Trigger: the workspace's remote has not been configured yet.
# Why: provisioning mints tenant-scoped credentials and must be explicit
# (no surprise key generation); push/pull only transfer.
# Outcome: stop with the exact setup command for this workspace.
if not rclone_remote_exists(remote_name):
setup_target = (
"" if target_workspace.is_default else f" --workspace {target_workspace.slug}"
)
console.print(f"[red]Workspace '{target_workspace.slug}' is not set up for sync.[/red]")
console.print(f"\nRun: bm cloud setup{setup_target}")
raise typer.Exit(1)
# Get tenant info for bucket name, scoped to the resolved workspace
tenant_info = run_with_cleanup(get_mount_info(workspace_id=target_workspace.tenant_id))
# Get tenant info for bucket name
tenant_info = run_with_cleanup(get_mount_info())
bucket_name = tenant_info.bucket_name
# Get project info from the same workspace we resolved above, so the
# project path and the bucket/remote all refer to one tenant.
# Get project info
with force_routing(cloud=True):
project_data = run_with_cleanup(
_get_cloud_project(name, workspace_id=target_workspace.tenant_id)
)
project_data = run_with_cleanup(_get_cloud_project(name))
if not project_data:
console.print(f"[red]Error: Project '{name}' not found[/red]")
raise typer.Exit(1)
sync_project, _ = _get_sync_project(name, config, project_data, remote_name=remote_name)
sync_project, _ = _get_sync_project(name, config, project_data)
# --- Detect before transferring ---
plan = project_diff(sync_project, bucket_name, direction)
@@ -446,11 +355,6 @@ def pull_project_command(
"--on-conflict",
help="Resolve files that differ on both sides (default: fail and list them)",
),
workspace: str | None = typer.Option(
None,
"--workspace",
help="Workspace (slug, name, or tenant_id) when the project name is ambiguous",
),
dry_run: bool = typer.Option(False, "--dry-run", help="Preview changes without pulling"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
) -> None:
@@ -464,10 +368,9 @@ def pull_project_command(
bm cloud pull --name research
bm cloud pull --name research --dry-run
bm cloud pull --name research --on-conflict keep-cloud
bm cloud pull --name research --workspace acme
"""
_run_directional_transfer(
name, "pull", on_conflict=on_conflict, dry_run=dry_run, verbose=verbose, workspace=workspace
name, "pull", on_conflict=on_conflict, dry_run=dry_run, verbose=verbose
)
@@ -479,11 +382,6 @@ def push_project_command(
"--on-conflict",
help="Resolve files that differ on both sides (default: fail and list them)",
),
workspace: str | None = typer.Option(
None,
"--workspace",
help="Workspace (slug, name, or tenant_id) when the project name is ambiguous",
),
dry_run: bool = typer.Option(False, "--dry-run", help="Preview changes without pushing"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
) -> None:
@@ -498,10 +396,9 @@ def push_project_command(
bm cloud push --name research
bm cloud push --name research --dry-run
bm cloud push --name research --on-conflict keep-local
bm cloud push --name research --workspace acme
"""
_run_directional_transfer(
name, "push", on_conflict=on_conflict, dry_run=dry_run, verbose=verbose, workspace=workspace
name, "push", on_conflict=on_conflict, dry_run=dry_run, verbose=verbose
)
@@ -512,11 +409,7 @@ 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 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.
"""Two-way sync: local <-> cloud (bidirectional sync).
Examples:
bm cloud bisync --name research --resync # First time
@@ -528,10 +421,7 @@ def bisync_project_command(
_require_personal_workspace(name, config)
try:
# Get tenant info for bucket name.
# TODO(#919): scope to the project's workspace like push/pull. Safe for now
# because these mirror commands are gated to the (default-tenant) Personal
# workspace, so the default mount info is correct.
# Get tenant info for bucket name
tenant_info = run_with_cleanup(get_mount_info())
bucket_name = tenant_info.bucket_name
@@ -580,11 +470,7 @@ 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 (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.
"""Verify file integrity between local and cloud.
Example:
bm cloud check --name research
@@ -593,10 +479,7 @@ def check_project_command(
_require_cloud_credentials(config)
try:
# Get tenant info for bucket name.
# TODO(#919): scope to the project's workspace like push/pull. Safe for now
# because these mirror commands are gated to the (default-tenant) Personal
# workspace, so the default mount info is correct.
# Get tenant info for bucket name
tenant_info = run_with_cleanup(get_mount_info())
bucket_name = tenant_info.bucket_name
@@ -632,9 +515,6 @@ 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.
"""
@@ -2,8 +2,7 @@
This module provides simplified, project-scoped rclone operations:
- Each project syncs independently
- Routes through the project's tenant-scoped remote (SyncProject.remote_name);
the default tenant keeps "basic-memory-cloud", others use their own (see #919)
- Uses single "basic-memory-cloud" remote (not tenant-specific)
- Balanced defaults from SPEC-8 Phase 4 testing
- Per-project bisync state tracking
@@ -114,15 +113,11 @@ class SyncProject:
name: Project name
path: Cloud path (e.g., "app/data/research")
local_sync_path: Local directory for syncing (optional)
remote_name: rclone remote serving this project's tenant bucket. Defaults
to the legacy single remote; team/non-default workspaces use their own
(see remote_name_for_workspace).
"""
name: str
path: str
local_sync_path: Optional[str] = None
remote_name: str = "basic-memory-cloud"
def get_bmignore_filter_path() -> Path:
@@ -184,13 +179,10 @@ def get_project_remote(project: SyncProject, bucket_name: str) -> str:
The API returns paths like "/app/data/basic-memory-llc" because the S3 bucket
is mounted at /app/data on the fly machine. We need to strip the /app/data/
prefix to get the actual S3 path within the bucket.
The remote name comes from the project so non-default/team workspaces route
through their own tenant-scoped remote (see #919).
"""
# Normalize path to strip /app/data/ mount point prefix
cloud_path = normalize_project_path(project.path).lstrip("/")
return f"{project.remote_name}:{bucket_name}/{cloud_path}"
return f"basic-memory-cloud:{bucket_name}/{cloud_path}"
# --- Directional transfer primitives (push / pull) ---
@@ -1,14 +1,11 @@
"""rclone configuration management for Basic Memory Cloud.
This module owns rclone remote configuration and naming. The default tenant uses
the "basic-memory-cloud" remote (from SPEC-20); non-default/team workspaces each
get their own tenant-scoped remote via remote_name_for_workspace (see #919),
since Tigris credentials are bucket-scoped.
This module provides simplified rclone configuration for SPEC-20.
Uses a single "basic-memory-cloud" remote for all operations.
"""
import configparser
import os
import re
import shutil
from pathlib import Path
from typing import Optional
@@ -64,65 +61,25 @@ def save_rclone_config(config: configparser.ConfigParser) -> None:
console.print(f"[dim]Updated rclone config: {config_path}[/dim]")
# The default remote serves the account's default tenant (back-compat with SPEC-20,
# which used a single "basic-memory-cloud" remote). Non-default workspaces each get
# their own remote, since Tigris credentials are bucket/tenant-scoped (see #919).
DEFAULT_RCLONE_REMOTE = "basic-memory-cloud"
# rclone remote section names allow letters, digits, hyphens, and underscores.
# The slug comes from the cloud API (a trust boundary), so validate it before
# splicing it into a remote name to avoid a broken/unusable rclone.conf section.
_SAFE_SLUG = re.compile(r"^[A-Za-z0-9_-]+$")
def remote_name_for_workspace(slug: str | None, *, is_default: bool) -> str:
"""Return the rclone remote name for a workspace.
The default workspace keeps the legacy ``basic-memory-cloud`` remote so
existing setups keep working; other workspaces get ``basic-memory-cloud-<slug>``.
Raises:
RcloneConfigError: If a non-default workspace slug contains characters
that are not valid in an rclone remote name.
"""
if is_default or not slug:
return DEFAULT_RCLONE_REMOTE
if not _SAFE_SLUG.match(slug):
raise RcloneConfigError(
f"Workspace slug '{slug}' cannot be used as an rclone remote name "
"(allowed: letters, digits, hyphens, underscores)."
)
return f"{DEFAULT_RCLONE_REMOTE}-{slug}"
def rclone_remote_exists(remote_name: str) -> bool:
"""Return whether an rclone remote section is already configured."""
return load_rclone_config().has_section(remote_name)
def configure_rclone_remote(
access_key: str,
secret_key: str,
endpoint: str = "https://fly.storage.tigris.dev",
region: str = "auto",
remote_name: str = DEFAULT_RCLONE_REMOTE,
) -> str:
"""Configure an rclone remote for one tenant's bucket.
"""Configure single rclone remote named 'basic-memory-cloud'.
Each tenant (personal or team) has its own bucket-scoped credentials, so a
remote maps 1:1 to a tenant. The default tenant keeps ``basic-memory-cloud``;
other workspaces pass ``remote_name`` from :func:`remote_name_for_workspace`.
This is the simplified approach from SPEC-20 that uses one remote
for all Basic Memory cloud operations (not tenant-specific).
Args:
access_key: S3 access key ID
secret_key: S3 secret access key
endpoint: S3-compatible endpoint URL
region: S3 region (default: auto)
remote_name: rclone remote section name to write
Returns:
The remote name that was configured
The remote name: "basic-memory-cloud"
"""
# Backup existing config
backup_rclone_config()
@@ -130,21 +87,24 @@ def configure_rclone_remote(
# Load existing config
config = load_rclone_config()
# Add/update the remote section
if not config.has_section(remote_name):
config.add_section(remote_name)
# Single remote name (not tenant-specific)
REMOTE_NAME = "basic-memory-cloud"
config.set(remote_name, "type", "s3")
config.set(remote_name, "provider", "Other")
config.set(remote_name, "access_key_id", access_key)
config.set(remote_name, "secret_access_key", secret_key)
config.set(remote_name, "endpoint", endpoint)
config.set(remote_name, "region", region)
# Add/update the remote section
if not config.has_section(REMOTE_NAME):
config.add_section(REMOTE_NAME)
config.set(REMOTE_NAME, "type", "s3")
config.set(REMOTE_NAME, "provider", "Other")
config.set(REMOTE_NAME, "access_key_id", access_key)
config.set(REMOTE_NAME, "secret_access_key", secret_key)
config.set(REMOTE_NAME, "endpoint", endpoint)
config.set(REMOTE_NAME, "region", region)
# Prevent unnecessary encoding of filenames (only encode slashes and invalid UTF-8)
# This prevents files with spaces like "Hello World.md" from being quoted
config.set(remote_name, "encoding", "Slash,InvalidUtf8")
config.set(REMOTE_NAME, "encoding", "Slash,InvalidUtf8")
# Save updated config
save_rclone_config(config)
console.print(f"[green]Configured rclone remote: {remote_name}[/green]")
return remote_name
console.print(f"[green]Configured rclone remote: {REMOTE_NAME}[/green]")
return REMOTE_NAME
-9
View File
@@ -7,15 +7,6 @@ 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 = {
-29
View File
@@ -276,35 +276,6 @@ 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]:
-4
View File
@@ -36,11 +36,7 @@ from basic_memory.mcp.tools.chatgpt_tools import search, fetch
# Schema tools
from basic_memory.mcp.tools.schema import schema_validate, schema_infer, schema_diff
# Diagnostics tool
from basic_memory.mcp.tools.basic_memory_diagnostics import basic_memory_diagnostics
__all__ = [
"basic_memory_diagnostics",
"build_context",
"canvas",
"cloud_info",
@@ -1,132 +0,0 @@
"""Diagnostic tool for Basic Memory version and system information."""
import json
import platform
import sys
from urllib.parse import urlparse, urlunparse
import basic_memory
from basic_memory.config import CONFIG_FILE_NAME, ConfigManager
from basic_memory.mcp.server import mcp
# Fields in BasicMemoryConfig that contain secrets and must never be surfaced.
_SECRET_FIELDS = frozenset({"cloud_api_key"})
# Fields whose values are URLs that may embed user:password credentials.
# The userinfo component is stripped before surfacing.
_URL_FIELDS = frozenset({"database_url"})
def _redact_url(url: str) -> str:
"""Strip the userinfo (user:password) from a URL string.
Replaces any credentials with *** so the host/path remain visible for
diagnostics (e.g. ``postgresql://***@localhost/mydb``). If the value
cannot be parsed as a URL it is returned unchanged.
"""
try:
parsed = urlparse(url)
except Exception: # pragma: no cover — urlparse is very permissive
return url
if not parsed.hostname:
# Not a meaningful URL (e.g. a bare file path); leave it alone.
return url
if parsed.username or parsed.password:
# Rebuild with credentials replaced by a placeholder.
netloc = f"***@{parsed.hostname}"
if parsed.port:
netloc = f"{netloc}:{parsed.port}"
redacted = parsed._replace(netloc=netloc)
return urlunparse(redacted)
return url
def _redact_config(raw: dict) -> dict:
"""Return a copy of the raw config dict with secret fields removed.
- Keys in ``_SECRET_FIELDS`` are dropped entirely.
- Keys in ``_URL_FIELDS`` have their userinfo component stripped so that
host and database name remain visible for diagnostics.
Only top-level keys are processed. Nested keys within project entries are
not currently credential-bearing, but the two sets make the pattern easy
to extend.
"""
result: dict = {}
for k, v in raw.items():
if k in _SECRET_FIELDS:
# Drop entirely — value has no diagnostic value.
continue
if k in _URL_FIELDS and isinstance(v, str):
result[k] = _redact_url(v)
else:
result[k] = v
return result
@mcp.tool(
"basic_memory_diagnostics",
title="Basic Memory Diagnostics",
tags={"diagnostics"},
annotations={"readOnlyHint": True, "openWorldHint": False},
# The report is a markdown string; suppress FastMCP's wrap_result so the
# payload isn't duplicated into structuredContent.
output_schema=None,
)
def basic_memory_diagnostics() -> str:
"""Return version, system, and configuration diagnostics for Basic Memory.
Provides:
- Basic Memory package version
- Python version and platform details
- Config file path and its contents (secrets redacted)
Useful for troubleshooting installations and gathering information for
support requests. Read-only; never emits secrets or API keys.
"""
# --- Version information ---
bm_version = basic_memory.__version__
# --- System information ---
python_version = sys.version
platform_info = platform.platform()
machine = platform.machine()
# --- Configuration ---
manager = ConfigManager()
config_file = manager.config_dir / CONFIG_FILE_NAME
config_exists = config_file.exists()
if config_exists:
try:
raw_config = json.loads(config_file.read_text(encoding="utf-8"))
safe_config = _redact_config(raw_config)
config_dump = json.dumps(safe_config, indent=2, default=str)
except Exception as exc: # pragma: no cover
config_dump = f"<error reading config: {exc}>"
else:
config_dump = "<config file not found>"
lines = [
"# Basic Memory Diagnostics",
"",
"## Version",
f"- basic-memory: {bm_version}",
"",
"## System",
f"- Python: {python_version}",
f"- Platform: {platform_info}",
f"- Architecture: {machine}",
"",
"## Configuration",
f"- Config path: {config_file}",
f"- Config exists: {config_exists}",
"",
"```json",
config_dump,
"```",
]
return "\n".join(lines)
+8 -104
View File
@@ -1,19 +1,13 @@
"""Edit note tool for Basic Memory MCP server."""
from typing import TYPE_CHECKING, Annotated, Optional, Literal
from typing import 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,
@@ -22,7 +16,6 @@ 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 (
@@ -59,79 +52,6 @@ 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],
@@ -206,8 +126,7 @@ 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. **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**:
3. **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
@@ -424,9 +343,7 @@ 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. 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
search_notes() first to find the correct identifier. The tool provides detailed
error messages with suggestions if operations fail.
"""
# Resolve effective default: allow MCP clients to send null for optional int field
@@ -548,27 +465,14 @@ 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
# 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
if is_not_found and operation in ("append", "prepend"):
title, directory = _parse_identifier_to_title_and_directory(identifier)
# Validate directory path (same security check as write_note)
+34 -115
View File
@@ -1,14 +1,13 @@
"""Read note tool for Basic Memory MCP server."""
from textwrap import dedent
from typing import Annotated, Optional, Literal, cast
from typing import Optional, Literal, cast
import logfire
import yaml
from loguru import logger
from fastmcp import Context
from pydantic import AliasChoices, Field
from basic_memory.config import ConfigManager
from basic_memory.mcp.project_context import (
@@ -21,17 +20,6 @@ from basic_memory.mcp.tools.search import search_notes
from basic_memory.schemas.memory import memory_url_path
from basic_memory.utils import validate_project_path
# The title-match fallback exists to find THE note by exact title, so it scans
# fixed-size pages of title results instead of the caller's page/page_size
# (which apply only to the text-search suggestion listing).
_TITLE_LOOKUP_PAGE_SIZE = 10
# Hard safety cap on title-lookup pages. The loop normally stops as soon as an
# exact match is found or results run out (has_more=False); the cap only bounds
# pathological knowledge bases where hundreds of fuzzy titles contain the
# queried phrase. Exhausting the cap falls through to the suggestion behavior.
_TITLE_LOOKUP_MAX_PAGES = 10
def _is_exact_title_match(identifier: str, title: str) -> bool:
"""Return True when identifier exactly matches a title (case-insensitive)."""
@@ -83,21 +71,6 @@ async def read_note(
identifier: str,
project: Optional[str] = None,
project_id: Optional[str] = None,
# Accept common pagination aliases models reach for from training data
# (page_number/limit/per_page), matching the sibling navigation tools
# (search_notes, build_context, recent_activity). The schema advertises
# only the canonical names; aliases are silently mapped at validation time.
# `offset` is intentionally NOT aliased: offset is item-indexed (skip N
# items) while page is a 1-indexed page-number, so direct aliasing would
# return the wrong slice.
page: Annotated[
int,
Field(default=1, validation_alias=AliasChoices("page", "page_number")),
] = 1,
page_size: Annotated[
int,
Field(default=10, validation_alias=AliasChoices("page_size", "limit", "per_page")),
] = 10,
output_format: Literal["text", "json"] = "text",
include_frontmatter: bool = False,
context: Context | None = None,
@@ -126,14 +99,6 @@ async def read_note(
workspaces. Takes precedence over `project`. Get from list_memory_projects().
identifier: The title or permalink of the note to read
Can be a full memory:// URL, a permalink, a title, or search text
page: Page of fallback-search results to use when the identifier does not
resolve to a note directly (default: 1). A direct or exact-title match
always returns the full note content page/page_size never chunk the
note itself, and the title-match lookup pages through fixed-size pages
of title results until an exact match is found or results are
exhausted, regardless of page or page_size.
page_size: Number of fallback-search results per page (default: 10). When no
match is found, this caps how many related-note suggestions are listed.
output_format: "text" returns markdown content or guidance text.
"json" returns a structured object with title/permalink/file_path/content/frontmatter.
include_frontmatter: When output_format="json", whether content should include the
@@ -157,9 +122,6 @@ async def read_note(
# Read recent meeting notes
read_note("team-docs", "Weekly Standup")
# Page through fallback-search suggestions when nothing matches directly
read_note("unknown topic", page=2, page_size=5)
Raises:
HTTPError: If project doesn't exist or is inaccessible
SecurityError: If identifier attempts path traversal
@@ -168,15 +130,6 @@ async def read_note(
If the exact note isn't found, this tool provides helpful suggestions
including related notes, search commands, and note creation templates.
"""
# Trigger: page < 1 or page_size < 1 (e.g. page_size=0 or negative).
# Why: both flow into the fallback search's server-side slicing, where
# non-positive values produce empty result pages with unreachable
# pagination. Fail fast, matching search_notes/build_context.
if page < 1:
raise ValueError(f"page must be >= 1, got {page}")
if page_size < 1:
raise ValueError(f"page_size must be >= 1, got {page_size}")
# Detect project from a memory URL or permalink prefix before routing.
# project_id routes by external UUID, so it bypasses URL discovery entirely.
if project is None and project_id is None:
@@ -194,8 +147,6 @@ async def read_note(
tool_name="read_note",
requested_project=project,
requested_project_id=project_id,
page=page,
page_size=page_size,
output_format=output_format,
include_frontmatter=include_frontmatter,
):
@@ -294,7 +245,7 @@ async def read_note(
]
async def _search_candidates(
identifier_text: str, *, title_only: bool, lookup_page: int = 1
identifier_text: str, *, title_only: bool
) -> dict[str, object]:
# Trigger: direct entity resolution failed for the caller's identifier.
# Why: search_notes applies the same memory:// normalization and tool-level
@@ -305,24 +256,11 @@ async def read_note(
# Without this, project names that collide across workspaces could re-resolve
# to a different tenant via the default-workspace fallback (CLI/context=None).
search_type = "title" if title_only else "text"
# Trigger: title_only — the title search exists to find THE note by
# exact title, not to page through suggestions.
# Why: paginating it by the caller's page would skip an exact match
# sitting on page 1 (read_note("Exact Title", page=2)), and a
# small caller page_size could let a higher-ranked fuzzy title
# displace the exact match out of the lookup window
# (read_note("Foo Bar", page_size=1) when "Foo Bar Foo Bar"
# ranks first) — both returning suggestions instead of the note.
# Outcome: title lookup uses its own lookup_page with a fixed lookup
# size, walked by the caller below; caller page/page_size
# apply only to the text-search suggestion listing.
response = await search_notes(
project=active_project.name,
project_id=active_project.external_id,
query=identifier_text,
search_type=search_type,
page=lookup_page if title_only else page,
page_size=_TITLE_LOOKUP_PAGE_SIZE if title_only else page_size,
output_format="json",
context=context,
)
@@ -359,24 +297,12 @@ async def read_note(
logger.info(f"Direct lookup failed for '{entity_path}': {e}")
# Continue to fallback methods
# Fallback 1: Try title search via API, walking fixed-size pages of
# title results until an exact match is found or results run out.
# A single page is not enough: when more than _TITLE_LOOKUP_PAGE_SIZE
# higher-ranked fuzzy titles contain the queried phrase, the exact
# title lands on a later page and a one-page lookup would miss it.
# Fallback 1: Try title search via API
logger.info(f"Search title for: {identifier}")
result: dict[str, object] | None = None
for lookup_page in range(1, _TITLE_LOOKUP_MAX_PAGES + 1):
title_results = await _search_candidates(
identifier, title_only=True, lookup_page=lookup_page
)
title_candidates = _search_results(title_results)
if not title_candidates:
logger.info(
f"No results in title search for: {identifier} "
f"in project {active_project.name}"
)
break
title_results = await _search_candidates(identifier, title_only=True)
title_candidates = _search_results(title_results)
if title_candidates:
# Trigger: direct resolution failed and title search returned candidates.
# Why: avoid returning unrelated notes when search yields only fuzzy matches.
# Outcome: fetch content only when a true exact title match exists.
@@ -388,37 +314,33 @@ async def read_note(
),
None,
)
if result is not None:
break
# Trigger: this page held only fuzzy titles and the server reports
# no further pages (has_more is False or absent).
# Why: continuing past the last page would issue empty lookups.
# Outcome: give up on the title fallback and try text search below.
if title_results.get("has_more") is not True:
if not result:
logger.info(f"No exact title match found for: {identifier}")
break
if result is not None and _result_permalink(result):
try:
# Resolve the permalink to entity ID
entity_id = await knowledge_client.resolve_entity(
_result_permalink(result) or "", strict=True
)
# Fetch content using the entity ID
response = await resource_client.read(entity_id)
if response.status_code == 200:
logger.info(
f"Found note by exact title search: {_result_permalink(result)}"
elif _result_permalink(result):
try:
# Resolve the permalink to entity ID
entity_id = await knowledge_client.resolve_entity(
_result_permalink(result) or "", strict=True
)
if output_format == "json":
return await _read_json_payload(entity_id)
return response.text
except Exception as e: # pragma: no cover
logger.info(
f"Failed to fetch content for found title match {_result_permalink(result)}: {e}"
)
# Fetch content using the entity ID
response = await resource_client.read(entity_id)
if response.status_code == 200:
logger.info(
f"Found note by exact title search: {_result_permalink(result)}"
)
if output_format == "json":
return await _read_json_payload(entity_id)
return response.text
except Exception as e: # pragma: no cover
logger.info(
f"Failed to fetch content for found title match {_result_permalink(result)}: {e}"
)
else:
logger.info(
f"No results in title search for: {identifier} in project {active_project.name}"
)
# Fallback 2: Text search as a last resort
logger.info(f"Title search failed, trying text search for: {identifier}")
@@ -430,9 +352,6 @@ async def read_note(
if output_format == "json":
return _empty_json_payload()
return format_not_found_message(active_project.name, identifier)
# The fallback search is paginated server-side to page_size, so list
# the whole returned page instead of a hardcoded cap — otherwise the
# caller's page_size would be silently ignored past the cap.
if output_format == "json":
payload = _empty_json_payload()
payload["related_results"] = [
@@ -441,10 +360,10 @@ async def read_note(
"permalink": _result_permalink(result),
"file_path": _result_file_path(result),
}
for result in text_candidates
for result in text_candidates[:5]
]
return payload
return format_related_results(active_project.name, identifier, text_candidates)
return format_related_results(active_project.name, identifier, text_candidates[:5])
def format_not_found_message(project: str | None, identifier: str) -> str:
+3 -26
View File
@@ -11,13 +11,7 @@ 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,
strict_search_tags,
)
from basic_memory.utils import build_canonical_permalink, coerce_dict, coerce_list
from basic_memory.mcp.async_client import (
_explicit_routing,
_force_local_mode,
@@ -682,15 +676,9 @@ async def search_notes(
Dict[str, Any] | None,
BeforeValidator(coerce_dict),
] = None,
# 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(strict_search_tags),
BeforeValidator(coerce_list),
] = None,
status: Optional[str] = None,
min_similarity: Annotated[
@@ -807,9 +795,7 @@ async def search_notes(
observations whose category matches exactly.
after_date: Optional date filter for recent content (e.g., "1 week", "2d", "2024-01-01")
metadata_filters: Optional structured frontmatter filters (e.g., {"status": "in-progress"})
tags: Optional tag filter (frontmatter tags); shorthand for metadata_filters["tags"].
Accepts a list (["a", "b"]) or a comma-separated string ("a,b"), matching the
write_note tags convention and the tag: query shorthand.
tags: Optional tag filter (frontmatter tags); shorthand for metadata_filters["tags"]
status: Optional status filter (frontmatter status); shorthand for metadata_filters["status"]
min_similarity: Optional float to override the global semantic_min_similarity threshold
for this query. E.g., 0.0 to see all vector results, or 0.8 for high precision.
@@ -901,15 +887,6 @@ 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,
+2 -6
View File
@@ -87,12 +87,8 @@ async def write_note(
Use forward slashes (/) as separators. Use "/" or "" to write to project root.
Examples: "notes", "projects/2025", "research/ml", "/" (root)
project: Project name to write to. Optional - server will resolve using the
hierarchy above. Use "workspace/project" to route to a project in a
specific cloud workspace. A bare name that exists in multiple
workspaces resolves to the default workspace, so use the qualified
form (or project_id) to disambiguate. If unknown, use
list_memory_projects() to discover available projects and their
qualified names.
hierarchy above. If unknown, use list_memory_projects() to discover
available projects.
project_id: Project external_id (UUID). Prefer this over `project` when known
it routes to the exact project regardless of name collisions across cloud
workspaces. Takes precedence over `project`. Get from list_memory_projects().
+2 -13
View File
@@ -1,6 +1,5 @@
"""Knowledge graph models."""
import hashlib
import uuid
from datetime import datetime
from basic_memory.utils import ensure_timezone_aware
@@ -253,18 +252,8 @@ class Observation(Base):
Content is truncated to 200 chars to stay under PostgreSQL's
btree index limit of 2704 bytes.
"""
if len(self.content) > 200:
# Trigger: content exceeds the 200-char budget imposed by PostgreSQL's
# 2704-byte btree index row limit, so the permalink can only carry a prefix.
# Why: two distinct observations with the same category and an identical
# 200-char prefix would collide on the same synthetic permalink, and the
# search index (permalink-keyed upsert) silently drops the second one.
# Outcome: a short stable digest of the FULL content disambiguates
# truncated permalinks while staying well under the index limit.
digest = hashlib.sha256(self.content.encode("utf-8")).hexdigest()[:12]
content_for_permalink = f"{self.content[:200]}-{digest}"
else:
content_for_permalink = self.content
# Truncate content to avoid exceeding PostgreSQL's btree index limit
content_for_permalink = self.content[:200] if len(self.content) > 200 else self.content
return generate_permalink(
f"{self.entity.permalink}/observations/{self.category}/{content_for_permalink}"
)
-2
View File
@@ -9,7 +9,6 @@ from basic_memory.schemas.v2.entity import (
DeleteDirectoryRequestV2,
ProjectResolveRequest,
ProjectResolveResponse,
SyncFileRequest,
)
from basic_memory.schemas.v2.graph import (
GraphEdge,
@@ -32,7 +31,6 @@ __all__ = [
"DeleteDirectoryRequestV2",
"ProjectResolveRequest",
"ProjectResolveResponse",
"SyncFileRequest",
"GraphEdge",
"GraphNode",
"GraphResponse",
-15
View File
@@ -54,21 +54,6 @@ 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.
+3 -6
View File
@@ -245,12 +245,9 @@ class ContextService:
type="observation",
id=obs.id,
title=f"{obs.category}: {obs.content[:50]}...",
# 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,
permalink=generate_permalink(
f"{primary_item.permalink}/observations/{obs.category}/{obs.content}"
),
file_path=primary_item.file_path,
content=obs.content,
category=obs.category,
-32
View File
@@ -568,38 +568,6 @@ 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:
-7
View File
@@ -381,13 +381,6 @@ def app_config(
sync_changes=False, # Disable file sync in tests - prevents lifespan from starting blocking task
database_backend=database_backend,
database_url=database_url,
# Trigger: semantic_search_enabled defaults to True whenever fastembed/sqlite-vec
# are importable, which they are in dev and CI environments.
# Why: with it on, every test that syncs pays the ONNX embedding stack (~5-7s per
# sync) — embeddings are covered by test-int/semantic/, which configures
# semantic_search_enabled explicitly in its own conftest.
# Outcome: non-semantic integration tests skip embedding work entirely.
semantic_search_enabled=False,
)
return app_config
@@ -4,8 +4,6 @@ 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
@@ -790,39 +788,3 @@ 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
@@ -1,141 +0,0 @@
"""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]}"
)
+8 -48
View File
@@ -13,49 +13,9 @@ import pytest
from fastmcp import Client
# --- read_note: pagination params restored in #883 ---
# `page` / `page_size` were removed in #693 because they were no-ops on the
# resource read. #883 restored them for parity with the sibling navigation
# tools: they paginate the server-side search fallback (a direct match still
# returns the full note content).
@pytest.mark.asyncio
async def test_read_note_accepts_pagination_params_and_aliases(mcp_server, app, test_project):
"""Agents passing page/page_size (or their aliases) must not get a validation error."""
async with Client(mcp_server) as client:
await client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "Paged Read Note",
"directory": "test",
"content": "# Paged Read Note\n\npaged read body",
},
)
result = await client.call_tool(
"read_note",
{
"project": test_project.name,
"identifier": "Paged Read Note",
"page": 1,
"page_size": 10,
},
)
assert "paged read body" in result.content[0].text
# Aliases map silently: page_number -> page, limit -> page_size
result = await client.call_tool(
"read_note",
{
"project": test_project.name,
"identifier": "Paged Read Note",
"page_number": 1,
"limit": 10,
},
)
assert "paged read body" in result.content[0].text
# --- read_note: pagination params removed in #693 (were no-ops) ---
# The `page` / `page_size` parameters were removed because the API endpoint
# silently dropped them. Search-fallback pagination is unrelated to read_note.
# --- edit_note: find_text / content / section aliases ---
@@ -525,12 +485,12 @@ async def test_aliases_not_advertised_in_schema(mcp_server, app):
# tool_name -> (must_have_canonical, must_not_have_aliases)
checks = {
# read_note pagination restored in #883 (paginates the search fallback).
# Accepted aliases (limit/page_number/per_page) plus the rejected
# `offset` must stay out of the advertised schema.
# read_note has no pagination params (#693 — they were no-ops; removed).
# The must_not_have list still includes the rejected aliases so future
# contributors don't reintroduce them.
"read_note": (
["page", "page_size"],
["offset", "limit", "page_number", "per_page"],
[],
["page", "page_size", "offset", "limit", "page_number", "per_page"],
),
"edit_note": (
["find_text", "section", "content"],
-52
View File
@@ -497,55 +497,3 @@ 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}"
)
+2 -15
View File
@@ -17,13 +17,12 @@ this path.
import os
import sqlite3
from unittest.mock import patch
import pytest
from sqlalchemy import text
from basic_memory import db
from basic_memory.config import BasicMemoryConfig, ConfigManager, DatabaseBackend
from basic_memory.config import BasicMemoryConfig, DatabaseBackend
from basic_memory.repository.entity_repository import EntityRepository
from basic_memory.repository.project_repository import ProjectRepository
from basic_memory.repository.sqlite_search_repository import SQLiteSearchRepository
@@ -154,19 +153,7 @@ async def test_embedding_status_reads_real_vec0_table(engine_factory, test_proje
project_repository = ProjectRepository(session_maker)
project_service = ProjectService(project_repository)
# Test fixtures run with semantic search disabled; the status call reads the global
# ConfigManager, so patch it to report semantic enabled for this regression path.
def _config_manager_semantic_enabled() -> ConfigManager:
cm = ConfigManager()
cm.config.semantic_search_enabled = True
return cm
with patch.object(
type(project_service),
"config_manager",
new_callable=lambda: property(lambda self: _config_manager_semantic_enabled()),
):
status = await project_service.get_embedding_status(project_id)
status = await project_service.get_embedding_status(project_id)
assert status.semantic_search_enabled is True
# The vec0 JOIN must succeed, so the table is reported as present and healthy.
-515
View File
@@ -1,23 +1,17 @@
"""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
@@ -961,512 +955,3 @@ 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"]
+718
View File
@@ -0,0 +1,718 @@
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"
+92
View File
@@ -0,0 +1,92 @@
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"] == ""
+21 -214
View File
@@ -15,21 +15,6 @@ 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",
[
@@ -402,37 +387,23 @@ def test_bisync_reset_skips_workspace_check_without_credentials(monkeypatch, tmp
assert "No bisync state found for project 'research'" in result.output
def _stub_transfer_env(
monkeypatch, module, *, plan, transfer_result=True, recorder=None, workspace=None
):
def _stub_transfer_env(monkeypatch, module, *, plan, transfer_result=True, recorder=None):
"""Stub the push/pull dependency chain so only diff/transfer logic is exercised."""
monkeypatch.setattr(module, "_require_cloud_credentials", lambda _config: None)
ws = workspace or _workspace("personal-tenant", "personal", "personal", is_default=True)
monkeypatch.setattr(
module,
"_get_workspace_for_project",
lambda _name, _config, **_kwargs: _async_value(ws),
)
monkeypatch.setattr(module, "rclone_remote_exists", lambda _remote: True)
monkeypatch.setattr(
module,
"get_mount_info",
lambda **_kwargs: _async_value(
SimpleNamespace(bucket_name="tenant-bucket", tenant_id=ws.tenant_id)
),
lambda: _async_value(SimpleNamespace(bucket_name="tenant-bucket")),
)
monkeypatch.setattr(
module,
"_get_cloud_project",
lambda _name, **_kwargs: _async_value(SimpleNamespace(name="research", path="research")),
lambda _name: _async_value(SimpleNamespace(name="research", path="research")),
)
monkeypatch.setattr(
module,
"_get_sync_project",
lambda _name, _config, _project_data, **kwargs: (
SimpleNamespace(name="research", remote_name=kwargs.get("remote_name")),
"/tmp/research",
),
lambda _name, _config, _project_data: (SimpleNamespace(name="research"), "/tmp/research"),
)
monkeypatch.setattr(module, "project_diff", lambda *args, **kwargs: plan)
@@ -539,195 +510,31 @@ def test_cloud_push_keep_local_resolves_conflict(monkeypatch, config_manager):
def test_cloud_push_allows_organization_workspace(monkeypatch, config_manager):
"""push is additive and Team-safe — an organization workspace is allowed (no Personal gate)."""
"""push is additive and Team-safe — it must not invoke the Personal-only guard."""
module = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
org_ws = _workspace("team-tenant", "organization", "acme", is_default=False)
config = config_manager.load_config()
config.cloud_api_key = "bmc_test"
config.projects["research"] = ProjectEntry(
path="/tmp/research",
mode=ProjectMode.CLOUD,
workspace_id="team-tenant",
local_sync_path="/tmp/research",
)
config_manager.save_config(config)
plan = TransferPlan(new=["new.md"], conflicts=[], dest_only=[], errors=[])
recorder: dict = {}
_stub_transfer_env(monkeypatch, module, plan=plan, recorder=recorder, workspace=org_ws)
_stub_transfer_env(monkeypatch, module, plan=plan)
monkeypatch.setattr(
module,
"get_available_workspaces",
lambda: pytest.fail("push/pull must not gate on workspace type"),
)
result = runner.invoke(app, ["cloud", "push", "--name", "research"])
assert result.exit_code == 0, result.output
assert "research push completed successfully" in result.output
# Routed through the team workspace's own remote, against its tenant's bucket.
assert recorder["args"][0].remote_name == "basic-memory-cloud-acme"
def test_cloud_pull_workspace_override_routes_through_workspace_remote(monkeypatch, config_manager):
"""pull --workspace routes through the named workspace's own remote and bucket."""
module = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
org_ws = _workspace("team-tenant", "organization", "acme", is_default=False)
plan = TransferPlan(new=["new.md"], conflicts=[], dest_only=[], errors=[])
recorder: dict = {}
# _get_workspace_for_project must receive the override and return the org workspace.
def _resolve(_name, _config, *, workspace_override=None):
assert workspace_override == "acme"
return _async_value(org_ws)
_stub_transfer_env(monkeypatch, module, plan=plan, recorder=recorder, workspace=org_ws)
monkeypatch.setattr(module, "_get_workspace_for_project", _resolve)
result = runner.invoke(app, ["cloud", "pull", "--name", "research", "--workspace", "acme"])
assert result.exit_code == 0, result.output
assert recorder["args"][0].remote_name == "basic-memory-cloud-acme"
assert recorder["args"][2] == "pull"
def test_cloud_push_errors_when_workspace_remote_not_set_up(monkeypatch, config_manager):
"""If the workspace's remote isn't configured, push stops with the setup command."""
module = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
org_ws = _workspace("team-tenant", "organization", "acme", is_default=False)
plan = TransferPlan(new=["new.md"], conflicts=[], dest_only=[], errors=[])
recorder: dict = {}
_stub_transfer_env(monkeypatch, module, plan=plan, recorder=recorder, workspace=org_ws)
# Override: this workspace has not been set up yet.
monkeypatch.setattr(module, "rclone_remote_exists", lambda _remote: False)
result = runner.invoke(app, ["cloud", "push", "--name", "research"])
assert result.exit_code == 1, result.output
output = " ".join(result.output.split())
assert "not set up for sync" in output
assert "bm cloud setup --workspace acme" in output
assert "args" not in recorder # never transferred
def test_get_workspace_for_project_override_resolves(monkeypatch, config_manager):
"""An explicit --workspace override selects that workspace regardless of config."""
module = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
config = config_manager.load_config()
monkeypatch.setattr(
module,
"get_available_workspaces",
lambda: _async_value(
[
_workspace("personal-tenant", "personal", "personal", is_default=True),
_workspace("team-tenant", "organization", "acme"),
]
),
)
ws = module.run_with_cleanup(
module._get_workspace_for_project("research", config, workspace_override="acme")
)
assert ws.tenant_id == "team-tenant"
def test_get_workspace_for_project_override_no_match_raises(monkeypatch, config_manager):
"""An override that matches no accessible workspace is a clear error."""
module = importlib.import_module("basic_memory.cli.commands.cloud.project_sync")
config = config_manager.load_config()
monkeypatch.setattr(
module,
"get_available_workspaces",
lambda: _async_value(
[_workspace("personal-tenant", "personal", "personal", is_default=True)]
),
)
with pytest.raises(ValueError) as exc_info:
module.run_with_cleanup(
module._get_workspace_for_project("research", config, workspace_override="acme")
)
assert "No accessible workspace matches 'acme'" in str(exc_info.value)
def _stub_setup_env(monkeypatch, core, *, remote_exists=False, recorder=None):
"""Stub the `bm cloud setup` dependency chain for the acme workspace."""
monkeypatch.setattr(core, "install_rclone", lambda: None)
monkeypatch.setattr(
core,
"get_available_workspaces",
lambda: _async_value([_workspace("team-tenant", "organization", "acme")]),
)
monkeypatch.setattr(core, "rclone_remote_exists", lambda _remote: remote_exists)
def _mint(_tenant_id):
if recorder is not None:
recorder["minted"] = True
return _async_value(SimpleNamespace(access_key="ak", secret_key="sk"))
monkeypatch.setattr(
core,
"get_mount_info",
lambda **_kwargs: _async_value(
SimpleNamespace(tenant_id="team-tenant", bucket_name="acme-bucket")
),
)
monkeypatch.setattr(core, "generate_mount_credentials", _mint)
def _fake_configure(**kwargs):
if recorder is not None:
recorder.update(kwargs)
return kwargs.get("remote_name")
monkeypatch.setattr(core, "configure_rclone_remote", _fake_configure)
def test_cloud_setup_workspace_configures_named_remote(monkeypatch):
"""`bm cloud setup --workspace acme` provisions the acme tenant's own remote."""
core = importlib.import_module("basic_memory.cli.commands.cloud.core_commands")
recorder: dict = {}
_stub_setup_env(monkeypatch, core, remote_exists=False, recorder=recorder)
result = runner.invoke(app, ["cloud", "setup", "--workspace", "acme"])
assert result.exit_code == 0, result.output
assert recorder["remote_name"] == "basic-memory-cloud-acme"
def test_cloud_setup_aborts_when_remote_exists_without_force(monkeypatch):
"""Setup refuses to overwrite an existing remote, and mints nothing (the #922 footgun)."""
core = importlib.import_module("basic_memory.cli.commands.cloud.core_commands")
recorder: dict = {}
_stub_setup_env(monkeypatch, core, remote_exists=True, recorder=recorder)
result = runner.invoke(app, ["cloud", "setup", "--workspace", "acme"])
assert result.exit_code == 1, result.output
output = " ".join(result.output.split())
assert "basic-memory-cloud-acme' is already configured" in output
assert "--force" in output
# Aborted before minting credentials or touching the remote.
assert "minted" not in recorder
assert "remote_name" not in recorder
def test_cloud_setup_force_overwrites_existing_remote(monkeypatch):
"""--force reconfigures an existing remote."""
core = importlib.import_module("basic_memory.cli.commands.cloud.core_commands")
recorder: dict = {}
_stub_setup_env(monkeypatch, core, remote_exists=True, recorder=recorder)
result = runner.invoke(app, ["cloud", "setup", "--workspace", "acme", "--force"])
assert result.exit_code == 0, result.output
assert recorder["remote_name"] == "basic-memory-cloud-acme"
assert recorder.get("minted") is True
def test_cloud_setup_default_workspace_aborts_when_remote_exists(monkeypatch):
"""The original footgun: `bm cloud setup` (no --workspace) must not clobber
the shared basic-memory-cloud remote without --force."""
core = importlib.import_module("basic_memory.cli.commands.cloud.core_commands")
recorder: dict = {}
_stub_setup_env(monkeypatch, core, remote_exists=True, recorder=recorder)
result = runner.invoke(app, ["cloud", "setup"]) # no --workspace → basic-memory-cloud
assert result.exit_code == 1, result.output
output = " ".join(result.output.split())
assert "'basic-memory-cloud' is already configured" in output
assert "minted" not in recorder # nothing minted on abort
assert "remote_name" not in recorder
async def _async_value(value):
@@ -1,14 +1,9 @@
import time
import pytest
from basic_memory.cli.commands.cloud.bisync_commands import convert_bmignore_to_rclone_filters
from basic_memory.cli.commands.cloud.rclone_config import (
RcloneConfigError,
configure_rclone_remote,
get_rclone_config_path,
rclone_remote_exists,
remote_name_for_workspace,
)
from basic_memory.ignore_utils import get_bmignore_path
@@ -116,36 +111,3 @@ def test_configure_rclone_remote_writes_config_and_backs_up_existing(config_home
# Backup exists
backups = list(cfg_path.parent.glob("rclone.conf.backup-*"))
assert backups, "expected a backup of rclone.conf to be created"
def test_remote_name_for_workspace():
# Default workspace keeps the legacy remote name (back-compat)
assert remote_name_for_workspace("personal", is_default=True) == "basic-memory-cloud"
assert remote_name_for_workspace(None, is_default=False) == "basic-memory-cloud"
# Non-default workspaces get their own tenant-scoped remote
assert remote_name_for_workspace("acme", is_default=False) == "basic-memory-cloud-acme"
def test_remote_name_for_workspace_rejects_unsafe_slug():
# A slug from the API with characters invalid in an rclone remote name must
# fail fast rather than write a broken rclone.conf section.
for bad in ["a/b", "has space", "dot.dot", "weird:name"]:
with pytest.raises(RcloneConfigError):
remote_name_for_workspace(bad, is_default=False)
def test_configure_rclone_remote_named_workspace_remote(config_home):
remote = configure_rclone_remote(
access_key="ak", secret_key="sk", remote_name="basic-memory-cloud-acme"
)
assert remote == "basic-memory-cloud-acme"
text = get_rclone_config_path().read_text(encoding="utf-8")
assert "[basic-memory-cloud-acme]" in text
assert "access_key_id = ak" in text
def test_rclone_remote_exists(config_home):
assert rclone_remote_exists("basic-memory-cloud-acme") is False
configure_rclone_remote(access_key="ak", secret_key="sk", remote_name="basic-memory-cloud-acme")
assert rclone_remote_exists("basic-memory-cloud-acme") is True
-7
View File
@@ -292,13 +292,6 @@ def app_config(config_home, db_backend, postgres_container, monkeypatch) -> Basi
update_permalinks_on_move=True,
database_backend=backend,
database_url=database_url,
# Trigger: semantic_search_enabled defaults to True whenever fastembed/sqlite-vec
# are importable, which they are in dev and CI environments.
# Why: with it on, every test that syncs pays the ONNX embedding stack (~5-7s per
# sync) — embeddings are covered by the dedicated semantic suites, which
# configure semantic_search_enabled explicitly themselves.
# Outcome: non-semantic tests skip embedding work entirely.
semantic_search_enabled=False,
)
return app_config
-30
View File
@@ -133,36 +133,6 @@ 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."""
@@ -1,247 +0,0 @@
"""Tests for the basic_memory_diagnostics MCP tool."""
import json
import platform
import sys
from unittest.mock import MagicMock, patch
import basic_memory
from basic_memory.mcp.tools.basic_memory_diagnostics import (
_redact_config,
_redact_url,
basic_memory_diagnostics,
)
# ---------------------------------------------------------------------------
# Unit tests for _redact_config helper
# ---------------------------------------------------------------------------
def test_redact_config_removes_cloud_api_key():
raw = {"cloud_api_key": "bmc_secret", "default_project": "main", "projects": {}}
result = _redact_config(raw)
assert "cloud_api_key" not in result
assert result["default_project"] == "main"
assert "projects" in result
def test_redact_config_passes_through_safe_fields():
raw = {"default_project": "main", "log_level": "INFO", "env": "dev"}
result = _redact_config(raw)
assert result == raw
def test_redact_config_empty_dict():
assert _redact_config({}) == {}
# ---------------------------------------------------------------------------
# Tests for the basic_memory_diagnostics tool
# ---------------------------------------------------------------------------
def test_diagnostics_returns_string():
result = basic_memory_diagnostics()
assert isinstance(result, str)
def test_diagnostics_includes_version():
result = basic_memory_diagnostics()
assert basic_memory.__version__ in result
def test_diagnostics_includes_python_version():
result = basic_memory_diagnostics()
# sys.version can be multi-line; just check the version tuple prefix
major_minor = f"{sys.version_info.major}.{sys.version_info.minor}"
assert major_minor in result
def test_diagnostics_includes_platform():
result = basic_memory_diagnostics()
assert platform.machine() in result
def test_diagnostics_includes_config_path(tmp_path):
"""Config path section should appear in output."""
with patch("basic_memory.mcp.tools.basic_memory_diagnostics.ConfigManager") as MockMgr:
mock_mgr = MagicMock()
mock_mgr.config_dir = tmp_path
MockMgr.return_value = mock_mgr
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"default_project": "main", "projects": {}}))
result = basic_memory_diagnostics()
assert str(tmp_path) in result
assert "Config path:" in result
def test_diagnostics_config_exists_with_valid_json(tmp_path):
"""When config file exists, its safe contents should appear as JSON."""
config_data = {
"default_project": "research",
"projects": {"research": {"path": str(tmp_path / "research")}},
}
with patch("basic_memory.mcp.tools.basic_memory_diagnostics.ConfigManager") as MockMgr:
mock_mgr = MagicMock()
mock_mgr.config_dir = tmp_path
MockMgr.return_value = mock_mgr
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps(config_data))
result = basic_memory_diagnostics()
assert "research" in result
assert "```json" in result
def test_diagnostics_redacts_cloud_api_key(tmp_path):
"""cloud_api_key must never appear in diagnostic output."""
config_data = {
"default_project": "main",
"cloud_api_key": "bmc_super_secret_token",
"projects": {},
}
with patch("basic_memory.mcp.tools.basic_memory_diagnostics.ConfigManager") as MockMgr:
mock_mgr = MagicMock()
mock_mgr.config_dir = tmp_path
MockMgr.return_value = mock_mgr
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps(config_data))
result = basic_memory_diagnostics()
assert "bmc_super_secret_token" not in result
assert "cloud_api_key" not in result
def test_diagnostics_config_missing(tmp_path):
"""When config file does not exist, output should say so."""
with patch("basic_memory.mcp.tools.basic_memory_diagnostics.ConfigManager") as MockMgr:
mock_mgr = MagicMock()
mock_mgr.config_dir = tmp_path
MockMgr.return_value = mock_mgr
# Ensure no config.json is present
config_file = tmp_path / "config.json"
assert not config_file.exists()
result = basic_memory_diagnostics()
assert "Config exists: False" in result
assert "<config file not found>" in result
def test_diagnostics_output_sections():
"""All expected section headers should be present."""
result = basic_memory_diagnostics()
assert "# Basic Memory Diagnostics" in result
assert "## Version" in result
assert "## System" in result
assert "## Configuration" in result
# ---------------------------------------------------------------------------
# Unit tests for _redact_url helper
# ---------------------------------------------------------------------------
def test_redact_url_strips_password():
url = "postgresql://user:secret@localhost/mydb"
result = _redact_url(url)
assert "secret" not in result
assert "user" not in result
assert "localhost" in result
assert "mydb" in result
assert "***" in result
def test_redact_url_strips_only_password_when_no_username():
# password-only userinfo (unusual but valid per RFC)
url = "postgresql://:secret@db.example.com/app"
assert _redact_url(url) == "postgresql://***@db.example.com/app"
def test_redact_url_preserves_port():
url = "postgresql://admin:pw@db.internal:5432/prod"
assert _redact_url(url) == "postgresql://***@db.internal:5432/prod"
def test_redact_url_no_credentials_unchanged():
url = "postgresql://db.internal:5432/prod"
assert _redact_url(url) == url
def test_redact_url_non_url_string_unchanged():
# Bare file paths / non-URL values must not be mangled.
path = "/home/user/.local/share/basic-memory/main.db"
assert _redact_url(path) == path
# ---------------------------------------------------------------------------
# _redact_config tests for database_url
# ---------------------------------------------------------------------------
def test_redact_config_scrubs_database_url_credentials():
raw = {
"default_project": "main",
"database_url": "postgresql://dbuser:dbpass@host.example.com:5432/bm",
"projects": {},
}
result = _redact_config(raw)
# Exact match: credentials replaced, host/port/db preserved for diagnostics.
assert result["database_url"] == "postgresql://***@host.example.com:5432/bm"
def test_redact_config_leaves_database_url_without_credentials():
raw = {"database_url": "sqlite:////tmp/basic-memory/main.db"}
result = _redact_config(raw)
assert result["database_url"] == "sqlite:////tmp/basic-memory/main.db"
def test_redact_config_drops_secret_fields_independently():
raw = {
"cloud_api_key": "bmc_top_secret",
"database_url": "postgresql://dbuser:dbpassword@host/db",
"default_project": "main",
}
result = _redact_config(raw)
assert "cloud_api_key" not in result
assert "dbpassword" not in result["database_url"]
assert "dbuser" not in result["database_url"]
assert "main" == result["default_project"]
# ---------------------------------------------------------------------------
# Integration: database_url redaction surfaces in diagnostic output
# ---------------------------------------------------------------------------
def test_diagnostics_redacts_database_url_password(tmp_path):
"""Postgres password in database_url must not appear in diagnostic output."""
config_data = {
"default_project": "main",
"database_url": "postgresql://pguser:supersecret@db.internal:5432/basicmemory",
"projects": {},
}
with patch("basic_memory.mcp.tools.basic_memory_diagnostics.ConfigManager") as MockMgr:
mock_mgr = MagicMock()
mock_mgr.config_dir = tmp_path
MockMgr.return_value = mock_mgr
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps(config_data))
result = basic_memory_diagnostics()
assert "supersecret" not in result
assert "pguser" not in result
# Host and port remain visible for diagnostics.
assert "db.internal" in result
assert "5432" in result
-2
View File
@@ -62,8 +62,6 @@ EXPECTED_TOOL_SIGNATURES: dict[str, list[str]] = {
"identifier",
"project",
"project_id",
"page",
"page_size",
"output_format",
"include_frontmatter",
],
+1 -266
View File
@@ -1,14 +1,10 @@
"""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.clients import KnowledgeClient
from basic_memory.mcp.tools.edit_note import _resolve_after_disk_recovery, edit_note
from basic_memory.mcp.tools.edit_note import edit_note
from basic_memory.mcp.tools.read_note import read_note
from basic_memory.mcp.tools.write_note import write_note
@@ -1267,264 +1263,3 @@ 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
-300
View File
@@ -120,306 +120,6 @@ async def test_read_note_returns_related_results_when_text_search_finds_matches(
assert "## 2. Related Two" in result
@pytest.mark.asyncio
async def test_read_note_direct_match_returns_full_content_regardless_of_paging(app, test_project):
"""page/page_size never chunk note content — a direct match returns the whole note."""
content = "Line one of the note\nLine two of the note\nLine three of the note"
await write_note(
project=test_project.name,
title="Paging Direct Note",
directory="test",
content=content,
)
result = await read_note(
"test/paging-direct-note",
project=test_project.name,
page=3,
page_size=1,
)
assert "Line one of the note" in result
assert "Line three of the note" in result
@pytest.mark.asyncio
async def test_read_note_rejects_non_positive_pagination(app, test_project):
"""Fail fast on invalid pagination, matching search_notes/build_context."""
with pytest.raises(ValueError, match="page must be >= 1"):
await read_note("any-note", project=test_project.name, page=0)
with pytest.raises(ValueError, match="page_size must be >= 1"):
await read_note("any-note", project=test_project.name, page_size=0)
@pytest.mark.asyncio
async def test_read_note_forwards_pagination_to_fallback_search(monkeypatch, app, test_project):
"""page/page_size must reach the server-side fallback search, not be swallowed."""
import importlib
read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
OriginalKnowledgeClient = clients_mod.KnowledgeClient
captured_pages: list[tuple[str, int, int]] = []
async def fake_search_notes_fn(*, query, search_type, page, page_size, **kwargs):
captured_pages.append((search_type, page, page_size))
return {"results": [], "current_page": page, "page_size": page_size}
class FailingKnowledgeClient(OriginalKnowledgeClient):
async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
raise RuntimeError("force fallback")
monkeypatch.setattr(clients_mod, "KnowledgeClient", FailingKnowledgeClient)
monkeypatch.setattr(read_note_module, "search_notes", fake_search_notes_fn)
result = await read_note("missing-note", project=test_project.name, page=2, page_size=3)
# Title lookup is pinned to page 1 with a fixed lookup size (it exists to
# find THE note by exact title); caller page/page_size apply only to the
# text-search suggestions.
assert captured_pages == [("title", 1, 10), ("text", 2, 3)]
assert "Note Not Found" in result
@pytest.mark.asyncio
async def test_read_note_title_fallback_finds_exact_match_on_later_page(
monkeypatch, app, test_project
):
"""An exact title match is returned even when the caller asks for page > 1.
The title-match lookup is pinned to page 1 of title results; without the pin,
read_note("Exact Title", page=2) would page past the match and return
unrelated suggestions instead of the note.
"""
await write_note(
project=test_project.name,
title="Paged Title Note",
directory="test",
content="paged title content",
)
import importlib
from basic_memory.schemas.memory import memory_url_path
clients_mod = importlib.import_module("basic_memory.mcp.clients")
OriginalKnowledgeClient = clients_mod.KnowledgeClient
direct_identifier = memory_url_path("Paged Title Note")
class SelectiveKnowledgeClient(OriginalKnowledgeClient):
async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
# Fail on the direct identifier to force fallback to title search
if identifier == direct_identifier:
raise RuntimeError("force direct lookup failure")
return await super().resolve_entity(identifier, strict=strict)
monkeypatch.setattr(clients_mod, "KnowledgeClient", SelectiveKnowledgeClient)
content = await read_note("Paged Title Note", project=test_project.name, page=2)
assert "paged title content" in content
@pytest.mark.asyncio
async def test_read_note_title_fallback_finds_exact_match_with_small_page_size(
monkeypatch, app, test_project
):
"""An exact title match is returned even when the caller asks for a tiny page_size.
The title-match lookup uses a fixed lookup size; without it, a higher-ranked
fuzzy title ("Foo Bar Foo Bar") would displace the exact title ("Foo Bar")
out of a page_size=1 window and read_note("Foo Bar", page_size=1) would
return suggestions instead of the note.
"""
await write_note(
project=test_project.name,
title="Foo Bar Foo Bar",
directory="test",
content="fuzzy decoy content",
)
await write_note(
project=test_project.name,
title="Foo Bar",
directory="test",
content="exact title content",
)
import importlib
from basic_memory.schemas.memory import memory_url_path
clients_mod = importlib.import_module("basic_memory.mcp.clients")
OriginalKnowledgeClient = clients_mod.KnowledgeClient
direct_identifier = memory_url_path("Foo Bar")
class SelectiveKnowledgeClient(OriginalKnowledgeClient):
async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
# Fail on the direct identifier to force fallback to title search
if identifier == direct_identifier:
raise RuntimeError("force direct lookup failure")
return await super().resolve_entity(identifier, strict=strict)
monkeypatch.setattr(clients_mod, "KnowledgeClient", SelectiveKnowledgeClient)
content = await read_note("Foo Bar", project=test_project.name, page_size=1)
assert "exact title content" in content
@pytest.mark.asyncio
async def test_read_note_title_fallback_pages_past_higher_ranked_fuzzy_titles(
monkeypatch, app, test_project
):
"""An exact title match is found even when it ranks beyond the first lookup page.
bm25 ranks titles that repeat the queried phrase above the exact title, so with
more than _TITLE_LOOKUP_PAGE_SIZE such decoys the exact match lands on page 2
of title results. A single-page lookup would fall through to suggestions even
though the note exists; the lookup must page until the exact title is found.
"""
from basic_memory.mcp.tools.read_note import _TITLE_LOOKUP_PAGE_SIZE
from basic_memory.mcp.tools.search import search_notes
for index in range(1, _TITLE_LOOKUP_PAGE_SIZE + 2):
await write_note(
project=test_project.name,
title=f"Deep Page Note Deep Page Note Deep Page Note {index:02d}",
directory="test",
content=f"fuzzy decoy content {index}",
)
await write_note(
project=test_project.name,
title="Deep Page Note",
directory="test",
content="deep page exact content",
)
# Precondition: the exact title must rank beyond the first lookup page,
# otherwise this test would pass even with a single-page lookup.
first_page = await search_notes(
project=test_project.name,
query="Deep Page Note",
search_type="title",
page=1,
page_size=_TITLE_LOOKUP_PAGE_SIZE,
output_format="json",
)
assert isinstance(first_page, dict)
first_page_titles = [result["title"] for result in first_page["results"]]
assert "Deep Page Note" not in first_page_titles
assert first_page["has_more"] is True
import importlib
from basic_memory.schemas.memory import memory_url_path
clients_mod = importlib.import_module("basic_memory.mcp.clients")
OriginalKnowledgeClient = clients_mod.KnowledgeClient
direct_identifier = memory_url_path("Deep Page Note")
class SelectiveKnowledgeClient(OriginalKnowledgeClient):
async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
# Fail on the direct identifier to force fallback to title search
if identifier == direct_identifier:
raise RuntimeError("force direct lookup failure")
return await super().resolve_entity(identifier, strict=strict)
monkeypatch.setattr(clients_mod, "KnowledgeClient", SelectiveKnowledgeClient)
content = await read_note("Deep Page Note", project=test_project.name)
assert "deep page exact content" in content
@pytest.mark.asyncio
async def test_read_note_title_lookup_stops_at_page_cap(monkeypatch, app, test_project):
"""The title lookup is bounded: after the page cap it falls through to suggestions."""
import importlib
read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
OriginalKnowledgeClient = clients_mod.KnowledgeClient
captured_pages: list[tuple[str, int]] = []
async def fake_search_notes_fn(*, query, search_type, page, page_size, **kwargs):
captured_pages.append((search_type, page))
if search_type == "title":
# Endless fuzzy titles: every page is full and reports more available,
# simulating a pathological knowledge base that never yields the note.
return {
"results": [
{
"title": f"Fuzzy {page}-{index}",
"permalink": f"docs/fuzzy-{page}-{index}",
"content": "",
"type": "entity",
"score": 1.0,
"file_path": f"docs/fuzzy-{page}-{index}.md",
}
for index in range(page_size)
],
"current_page": page,
"page_size": page_size,
"has_more": True,
}
return {"results": [], "current_page": page, "page_size": page_size}
class FailingKnowledgeClient(OriginalKnowledgeClient):
async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
raise RuntimeError("force fallback")
monkeypatch.setattr(clients_mod, "KnowledgeClient", FailingKnowledgeClient)
monkeypatch.setattr(read_note_module, "search_notes", fake_search_notes_fn)
result = await read_note("Pathological Note", project=test_project.name)
title_pages = [page for search_type, page in captured_pages if search_type == "title"]
assert title_pages == list(range(1, read_note_module._TITLE_LOOKUP_MAX_PAGES + 1))
assert "Note Not Found" in result
@pytest.mark.asyncio
async def test_read_note_related_results_list_full_search_page(monkeypatch, app, test_project):
"""Suggestions list the whole returned search page instead of a hardcoded cap of 5."""
import importlib
read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note")
clients_mod = importlib.import_module("basic_memory.mcp.clients")
OriginalKnowledgeClient = clients_mod.KnowledgeClient
candidates = [
{
"title": f"Related {index}",
"permalink": f"docs/related-{index}",
"content": "",
"type": "entity",
"score": 1.0,
"file_path": f"docs/related-{index}.md",
}
for index in range(1, 7)
]
async def fake_search_notes_fn(*, query, search_type, **kwargs):
if search_type == "title":
return {"results": [], "current_page": 1, "page_size": 10}
return {"results": candidates, "current_page": 1, "page_size": 10}
class FailingKnowledgeClient(OriginalKnowledgeClient):
async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str:
raise RuntimeError("force fallback")
monkeypatch.setattr(clients_mod, "KnowledgeClient", FailingKnowledgeClient)
monkeypatch.setattr(read_note_module, "search_notes", fake_search_notes_fn)
text_result = await read_note("missing-note", project=test_project.name)
assert "## 6. Related 6" in text_result
json_result = await read_note(
"missing-note",
project=test_project.name,
output_format="json",
)
assert isinstance(json_result, dict)
assert len(json_result["related_results"]) == 6
@pytest.mark.asyncio
async def test_read_note_title_fallback_requires_exact_title_match(monkeypatch, app, test_project):
"""Do not fetch note content when title-search returns only fuzzy matches."""
-241
View File
@@ -1,15 +1,11 @@
"""Tests for search MCP tools."""
import inspect
import pytest
from contextlib import asynccontextmanager
from datetime import datetime, timedelta
from types import SimpleNamespace
from typing import cast
from pydantic import TypeAdapter
from basic_memory.mcp.tools import write_note
from basic_memory.mcp.tools.search import (
search_notes,
@@ -1626,243 +1622,6 @@ async def test_search_notes_tag_prefix_with_remaining_text(monkeypatch):
assert captured_payload["text"] == "authentication"
# --- Tests for comma-separated tags parameter (#910) ----------------------------
def test_search_notes_tags_annotation_splits_comma_strings():
"""The tags parameter annotation must parse every documented input form (#910).
Direct function calls bypass the BeforeValidator, so validate through the same
Annotated metadata pydantic applies on the MCP path. coerce_list wrapped a bare
comma string as the single literal tag ["a,b"]; parse_tags splits it like the
tag: query shorthand and write_note's tags convention.
"""
annotation = inspect.signature(search_notes).parameters["tags"].annotation
adapter = TypeAdapter(annotation)
real_list = adapter.validate_python(["a", "b"])
comma_string = adapter.validate_python("a,b")
json_string = adapter.validate_python('["a", "b"]')
single_string = adapter.validate_python("a")
assert real_list == ["a", "b"]
# The comma string and the real list must behave identically (the #910 bug).
assert comma_string == real_list
assert json_string == real_list
assert single_string == ["a"]
@pytest.mark.asyncio
async def test_search_notes_tags_comma_string_filters_via_mcp(mcp, client, test_project):
"""tags="alpha,beta" through the real MCP layer must match like a real list (#910)."""
from fastmcp import Client
async with Client(mcp) as mcp_client:
await mcp_client.call_tool(
"write_note",
{
"project": test_project.name,
"title": "Tag Split Note",
"directory": "test",
"content": "# Tag Split Note\nTagSplitToken body",
"tags": ["alpha", "beta"],
},
)
async def found(tags_value: object) -> bool:
result = await mcp_client.call_tool(
"search_notes",
{
"project": test_project.name,
"query": "TagSplitToken",
"search_type": "text",
"tags": tags_value,
},
)
return "Tag Split Note" in result.content[0].text
as_list = await found(["alpha", "beta"])
as_comma_string = await found("alpha,beta")
as_json_string = await found('["alpha", "beta"]')
as_single_string = await found("alpha")
assert as_list, "real-list tags must match (sanity)"
assert as_comma_string == as_list, "comma string must behave like the real list"
assert as_json_string == as_list
assert as_single_string == as_list
# Negative control: the filter is actually applied, not silently dropped.
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) -----------------------------------
-2
View File
@@ -109,8 +109,6 @@ async def test_read_note_emits_root_operation_and_project_context(
"tool_name": "read_note",
"requested_project": test_project.name,
"requested_project_id": None,
"page": 1,
"page_size": 10,
"output_format": "json",
"include_frontmatter": True,
},
@@ -466,64 +466,3 @@ async def test_observation_permalink_short_content_unchanged(
# Short content should be fully included (after permalink normalization)
# The generate_permalink function normalizes the content
assert "short-observation-content" in permalink.lower()
@pytest.mark.asyncio
async def test_observation_permalink_disambiguates_truncated_content(
session_maker: async_sessionmaker, repo, test_project: Project
):
"""Regression test for issue #909: shared 200-char prefixes must not collide.
Truncating content to 200 chars (PostgreSQL btree limit) made two distinct
observations with the same category and an identical 200-char prefix produce
the same synthetic permalink, silently dropping the second from the search
index. A digest of the full content now disambiguates them.
"""
async with db.scoped_session(session_maker) as session:
entity = Entity(
project_id=test_project.id,
title="test_entity",
note_type="test",
permalink="test/test-entity",
file_path="test/test_entity.md",
content_type="text/markdown",
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
session.add(entity)
await session.flush()
shared_prefix = "x" * 210 # identical beyond the 200-char truncation point
obs_alpha = Observation(
project_id=test_project.id,
entity_id=entity.id,
content=f"{shared_prefix} ALPHA_UNIQUE_MARKER",
category="note",
)
obs_beta = Observation(
project_id=test_project.id,
entity_id=entity.id,
content=f"{shared_prefix} BETA_UNIQUE_MARKER",
category="note",
)
session.add_all([obs_alpha, obs_beta])
await session.flush()
# Distinct content must yield distinct permalinks despite the shared prefix
assert obs_alpha.permalink != obs_beta.permalink
# The digest suffix must keep permalinks under the PostgreSQL btree budget
assert len(obs_alpha.permalink) < 300
assert len(obs_beta.permalink) < 300
# Truly identical long content must still produce the same permalink so
# exact duplicates continue to dedupe in the search index
obs_beta_dup = Observation(
project_id=test_project.id,
entity_id=entity.id,
content=f"{shared_prefix} BETA_UNIQUE_MARKER",
category="note",
)
session.add(obs_beta_dup)
await session.flush()
assert obs_beta_dup.permalink == obs_beta.permalink
-41
View File
@@ -212,47 +212,6 @@ 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."""
@@ -14,28 +14,6 @@ def _is_postgres() -> bool:
return os.environ.get("BASIC_MEMORY_TEST_POSTGRES", "").lower() in ("1", "true", "yes")
async def _create_embeddings_stub(project_service: ProjectService) -> None:
"""Create a minimal search_vector_embeddings stub so vector_tables_exist is True.
Test fixtures run with semantic search disabled, so the real vec0/pgvector
embeddings table is never created. get_embedding_status only probes table
existence and joins on chunk_id (rowid on SQLite), so a plain table suffices.
"""
await project_service.repository.execute_query(
text(
"CREATE TABLE IF NOT EXISTS search_vector_embeddings ( chunk_id INTEGER PRIMARY KEY)"
),
{},
)
async def _drop_embeddings_stub(project_service: ProjectService) -> None:
"""Drop the stub table to avoid polluting subsequent tests."""
await project_service.repository.execute_query(
text("DROP TABLE IF EXISTS search_vector_embeddings"), {}
)
@pytest.mark.asyncio
async def test_embedding_status_semantic_disabled(project_service: ProjectService, test_project):
"""When semantic search is disabled, return minimal status with zero counts."""
@@ -91,9 +69,7 @@ async def test_embedding_status_entities_without_chunks(
project_service: ProjectService, test_graph, test_project
):
"""When entities have search_index rows but no chunks, recommend reindex."""
# search_vector_chunks comes from Base.metadata; the embeddings table needs a stub
# because fixtures run with semantic search disabled.
await _create_embeddings_stub(project_service)
# search_vector_chunks table is created by the test fixture (empty)
with patch.object(
type(project_service),
"config_manager",
@@ -103,8 +79,6 @@ async def test_embedding_status_entities_without_chunks(
):
status = await project_service.get_embedding_status(test_project.id)
await _drop_embeddings_stub(project_service)
assert status.semantic_search_enabled is True
assert status.vector_tables_exist is True
# test_graph creates entities indexed in search_index, but no vector chunks
@@ -185,10 +159,6 @@ async def test_embedding_status_handles_sqlite_vec_unavailable(
if _is_postgres():
pytest.skip("sqlite-vec unavailable handling is SQLite-specific.")
# Both vector tables must exist so the status check reaches the vec query;
# fixtures run with semantic search disabled, so stub the embeddings table.
await _create_embeddings_stub(project_service)
# scalar_vec_query returns None when the extension can't be loaded on this
# Python build (e.g. the python.org macOS interpreter). Simulate that here.
async def _vec_query_unavailable(query, params=None):
@@ -208,8 +178,6 @@ async def test_embedding_status_handles_sqlite_vec_unavailable(
):
status = await project_service.get_embedding_status(test_project.id)
await _drop_embeddings_stub(project_service)
assert status.semantic_search_enabled is True
assert status.total_indexed_entities > 0
assert status.vector_tables_exist is False
@@ -304,9 +272,6 @@ async def test_embedding_status_excludes_stale_entity_ids(
# Include 'id' column — required NOT NULL on Postgres (regular table),
# ignored on SQLite (FTS5 virtual table where id is UNINDEXED).
stale_entity_id = 999999
# Both vector tables must exist to reach the stale-filtered count queries;
# fixtures run with semantic search disabled, so stub the embeddings table.
await _create_embeddings_stub(project_service)
await project_service.repository.execute_query(
text(
"INSERT INTO search_index "
-103
View File
@@ -1248,78 +1248,6 @@ async def test_index_entity_multiple_categories_same_content(
assert len(results) >= 2
@pytest.mark.asyncio
async def test_index_entity_long_observations_shared_prefix_both_searchable(
search_service, session_maker, test_project
):
"""Regression test for issue #909: truncated permalink collisions drop observations.
Observation permalinks truncate content to 200 chars (PostgreSQL btree limit),
so two distinct observations of the same category sharing a 200-char prefix
collided on the same synthetic permalink and the second was silently skipped
during indexing. Both must be independently searchable.
"""
from basic_memory.repository import EntityRepository, ObservationRepository
from datetime import datetime
entity_repo = EntityRepository(session_maker, project_id=test_project.id)
obs_repo = ObservationRepository(session_maker, project_id=test_project.id)
entity_data = {
"title": "Long Observation Collision Entity",
"note_type": "note",
"entity_metadata": {},
"content_type": "text/markdown",
"file_path": "test/long-obs-collision.md",
"permalink": "test/long-obs-collision",
"project_id": test_project.id,
"created_at": datetime.now(),
"updated_at": datetime.now(),
}
entity = await entity_repo.create(entity_data)
# Identical for the first 210 chars (beyond the 200-char truncation point),
# differing only in the trailing unique marker
shared_prefix = "x" * 210
await obs_repo.create(
{
"entity_id": entity.id,
"category": "note",
"content": f"{shared_prefix} ALPHA_UNIQUE_MARKER",
}
)
await obs_repo.create(
{
"entity_id": entity.id,
"category": "note",
"content": f"{shared_prefix} BETA_UNIQUE_MARKER",
}
)
# Reload entity with observations (get_by_permalink eagerly loads observations)
entity = await entity_repo.get_by_permalink("test/long-obs-collision")
assert entity is not None
assert len(entity.observations) == 2
# Distinct content must produce distinct permalinks despite the shared prefix
permalinks = {obs.permalink for obs in entity.observations}
assert len(permalinks) == 2
await search_service.index_entity(entity, content="")
# The second observation must be findable by its own unique marker
results = await search_service.search(
SearchQuery(text="BETA_UNIQUE_MARKER", entity_types=[SearchItemType.OBSERVATION])
)
assert any("BETA_UNIQUE_MARKER" in (r.content_snippet or "") for r in results)
# The first observation must remain findable as well
results = await search_service.search(
SearchQuery(text="ALPHA_UNIQUE_MARKER", entity_types=[SearchItemType.OBSERVATION])
)
assert any("ALPHA_UNIQUE_MARKER" in (r.content_snippet or "") for r in results)
# Tests for NUL byte stripping
@@ -1384,21 +1312,6 @@ async def test_reindex_vectors(search_service, session_maker, test_project, monk
entity_repo = EntityRepository(session_maker, project_id=test_project.id)
# Test fixtures disable semantic search, and delete_stale_vector_rows is the one call
# in this flow that requires the semantic stack — stub it so the test exercises the
# reindex wiring (id collection, batch call, stats mapping) without embeddings.
# raising=False: the method is SQLite-only; the Postgres purge path never calls it,
# so on Postgres this just attaches an unused attribute.
async def _noop_delete_stale_vector_rows() -> None:
return None
monkeypatch.setattr(
search_service.repository,
"delete_stale_vector_rows",
_noop_delete_stale_vector_rows,
raising=False,
)
# Create some entities
created_entity_ids: list[int] = []
for i in range(3):
@@ -1469,22 +1382,6 @@ async def test_reindex_vectors_no_callback(
from datetime import datetime
entity_repo = EntityRepository(session_maker, project_id=test_project.id)
# Test fixtures disable semantic search, and delete_stale_vector_rows is the one call
# in this flow that requires the semantic stack — stub it so the test exercises the
# reindex wiring without embeddings.
# raising=False: the method is SQLite-only; the Postgres purge path never calls it,
# so on Postgres this just attaches an unused attribute.
async def _noop_delete_stale_vector_rows() -> None:
return None
monkeypatch.setattr(
search_service.repository,
"delete_stale_vector_rows",
_noop_delete_stale_vector_rows,
raising=False,
)
entity = await entity_repo.create(
{
"title": "No Callback Entity",
-7
View File
@@ -6,8 +6,6 @@ from pathlib import Path
from textwrap import dedent
from typing import Any, cast
import os
import pytest
from basic_memory.config import ProjectConfig, BasicMemoryConfig
@@ -390,11 +388,6 @@ modified: 2024-01-01
@pytest.mark.asyncio
@pytest.mark.skipif(
os.environ.get("CI") == "true",
reason="#940: intermittent batch-indexing race leaves a relation unresolved under CI "
"concurrency; quarantined pending root-cause, still runs locally",
)
async def test_sync_entity_circular_relations(
sync_service: SyncService, project_config: ProjectConfig
):
-73
View File
@@ -55,76 +55,3 @@ def test_codex_plugin_docs_explain_global_install_and_repo_mapping() -> None:
assert "codex plugin add codex@basic-memory-local" in readme
assert "Plugin installation is user-level in Codex" in readme
assert "Each repository still needs its own `.codex/basic-memory.json`" in readme
def test_infographics_skill_keeps_weekly_contract_and_bm_style_pool() -> None:
repo_root = Path(__file__).resolve().parents[1]
skill = (repo_root / ".agents/skills/infographics/SKILL.md").read_text(encoding="utf-8")
style_balance = (
repo_root / ".agents/skills/infographics/references/style-balance.md"
).read_text(encoding="utf-8")
prompt_blueprint = (
repo_root / ".agents/skills/infographics/references/prompt-blueprint.md"
).read_text(encoding="utf-8")
skill_flat = " ".join(skill.split())
assert "Weekly image" in skill
assert "2-Week Retro window" in skill
assert "docs/assets/infographics/<year>-w<start-week>-w<end-week>.webp" in skill
assert (
"docs/assets/infographics/<start-year>-w<start-week>-<end-year>-w<end-week>.webp" in skill
)
assert "computer science college textbooks" in skill
assert "classic literature subjects" in skill
assert "Metal, Hard Rock, Punk, techno, soul, reggae" in skill
assert "Star Wars inspired knockoff" in skill
assert "WWII propaganda posters" in skill
assert "Italian movie posters" in skill
assert "space exploration and astronomy" in skill
assert "paintings" in skill
assert "abstract painting" in skill
assert "classical landscape" in skill
assert "Remington-inspired" in skill
assert "Rembrandt-inspired" in skill
assert "classic black-and-white photography" in skill
assert "documentary" in skill
assert "editorial photo essay" in skill
assert "80's action movies" in skill
assert "practical explosions" in skill
assert "no direct actor likenesses" in skill
assert "poster, scene, tableau, cover image" in skill_flat
assert "image-first" in skill
assert "--print-prompt" in skill
assert "--dry-run" in skill
assert "--visual-format" not in skill
assert "--provenance-output" in skill
assert "BM_INFOGRAPHIC_PROVENANCE:start" in skill
assert "Image prompt sent to" not in skill
assert "revised prompt" not in skill
assert "star charts" in style_balance
assert "editorial scene" in style_balance
assert "painting, photograph" in style_balance
assert "copyrighted characters, logos, or named fictional universes" in skill
assert "retro game or classic app aesthetic" not in skill
assert "BM style category" in style_balance
assert "Chosen image form" in prompt_blueprint
assert "Chosen BM style category" in prompt_blueprint
def test_pr_create_skill_documents_optional_infographic_theme_arg() -> None:
repo_root = Path(__file__).resolve().parents[1]
skill = (repo_root / ".agents/skills/pr-create/SKILL.md").read_text(encoding="utf-8")
skill_flat = " ".join(skill.split())
assert "## How To Use" in skill
assert "$pr-create" in skill
assert "<theme>" in skill
assert '$pr-create "Italian movie poster"' in skill
assert '$pr-create "80\'s action movies"' in skill
assert "<!-- BM_INFOGRAPHIC_THEME:start -->" in skill
assert "<!-- BM_INFOGRAPHIC_THEME:end -->" in skill
assert "<!-- BM_INFOGRAPHIC_PROVENANCE:start -->" in skill
assert "BM Bossbot Approval" in skill
assert "selected visual direction" in skill_flat
assert "never merges" in skill
assert "non-gating" in skill
+2 -60
View File
@@ -1,9 +1,9 @@
"""Tests for coerce_list, coerce_dict, and strict_search_tags utility functions.
"""Tests for coerce_list and coerce_dict utility functions.
These must fail until the helpers are implemented in utils.py.
"""
from basic_memory.utils import coerce_list, coerce_dict, strict_search_tags
from basic_memory.utils import coerce_list, coerce_dict
class TestCoerceList:
@@ -33,64 +33,6 @@ 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."""
-13
View File
@@ -87,19 +87,6 @@ def test_get_project_remote():
assert get_project_remote(project, "my-bucket") == "basic-memory-cloud:my-bucket/research"
def test_get_project_remote_uses_workspace_remote():
# Non-default/team workspaces route through their own tenant-scoped remote.
project = SyncProject(name="research", path="/research", remote_name="basic-memory-cloud-acme")
assert (
get_project_remote(project, "acme-bucket") == "basic-memory-cloud-acme:acme-bucket/research"
)
def test_sync_project_remote_name_defaults_to_legacy_remote():
project = SyncProject(name="research", path="/research")
assert project.remote_name == "basic-memory-cloud"
def test_get_project_remote_strips_app_data_prefix():
project = SyncProject(name="research", path="/app/data/research")
assert get_project_remote(project, "my-bucket") == "basic-memory-cloud:my-bucket/research"
Generated
-28
View File
@@ -322,9 +322,7 @@ dev = [
{ name = "pytest-asyncio" },
{ name = "pytest-cov" },
{ name = "pytest-mock" },
{ name = "pytest-split" },
{ name = "pytest-testmon" },
{ name = "pytest-timeout" },
{ name = "pytest-xdist" },
{ name = "ruff" },
{ name = "testcontainers" },
@@ -391,9 +389,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" },
{ name = "ruff", specifier = ">=0.1.6" },
{ name = "testcontainers", extras = ["postgres"], specifier = ">=4.0.0" },
@@ -3062,18 +3058,6 @@ 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"
@@ -3087,18 +3071,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/61/55/ebb3c2f59fb089f08d00f764830d35780fc4e4c41dffcadafa3264682b65/pytest_testmon-2.2.0-py3-none-any.whl", hash = "sha256:2604ca44a54d61a2e830d9ce828b41a837075e4ebc1f81b148add8e90d34815b", size = 25199, upload-time = "2025-12-01T07:30:23.623Z" },
]
[[package]]
name = "pytest-timeout"
version = "2.4.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pytest" },
]
sdist = { url = "https://files.pythonhosted.org/packages/ac/82/4c9ecabab13363e72d880f2fb504c5f750433b2b6f16e99f4ec21ada284c/pytest_timeout-2.4.0.tar.gz", hash = "sha256:7e68e90b01f9eff71332b25001f85c75495fc4e3a836701876183c4bcfd0540a", size = 17973, upload-time = "2025-05-05T19:44:34.99Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/fa/b6/3127540ecdf1464a00e5a01ee60a1b09175f6913f0644ac748494d9c4b21/pytest_timeout-2.4.0-py3-none-any.whl", hash = "sha256:c42667e5cdadb151aeb5b26d114aff6bdf5a907f176a007a30b940d3d865b5c2", size = 14382, upload-time = "2025-05-05T19:44:33.502Z" },
]
[[package]]
name = "pytest-xdist"
version = "3.8.0"