Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 12 additions & 23 deletions src/ai_memory_protocol/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,15 +21,22 @@

import argparse
import json
import subprocess
import sys
import textwrap
from datetime import date
from pathlib import Path

from . import __version__
from .config import TYPE_FILES
from .engine import expand_graph, find_workspace, load_needs, resolve_id, tag_match, text_match
from .engine import (
expand_graph,
find_workspace,
load_needs,
resolve_id,
run_rebuild,
tag_match,
text_match,
)
from .formatter import format_brief, format_compact, format_context_pack, format_full
from .rst import (
add_tags_in_rst,
Expand Down Expand Up @@ -327,29 +334,11 @@ def cmd_stale(args: argparse.Namespace) -> None:
def cmd_rebuild(args: argparse.Namespace) -> None:
"""Rebuild needs.json by running Sphinx build."""
workspace = find_workspace(args.dir)
venv_sphinx = workspace / ".venv" / "bin" / "sphinx-build"
sphinx_cmd = str(venv_sphinx) if venv_sphinx.exists() else "sphinx-build"
cmd = [sphinx_cmd, "-b", "html", "-q", str(workspace), str(workspace / "_build" / "html")]
print(f"Building: {' '.join(cmd)}")
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
print(f"Build failed:\n{result.stderr}", file=sys.stderr)
success, message = run_rebuild(workspace)
print(message)
if not success:
Comment on lines +338 to +339
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

On failure, the rebuild message is printed to stdout and then the process exits non-zero. Previously failures were sent to stderr; it would be better to keep errors on stderr (and only print success output on stdout) for scripting/pipe use.

Suggested change
print(message)
if not success:
if success:
print(message)
else:
print(message, file=sys.stderr)

Copilot uses AI. Check for mistakes.
sys.exit(1)

from .engine import find_needs_json

print(f"needs.json updated at {find_needs_json(workspace)}")

needs = load_needs(workspace)
by_type: dict[str, int] = {}
by_status: dict[str, int] = {}
for n in needs.values():
by_type[n.get("type", "?")] = by_type.get(n.get("type", "?"), 0) + 1
by_status[n.get("status", "?")] = by_status.get(n.get("status", "?"), 0) + 1
print(f"Total: {len(needs)} memories")
print(f" Types: {', '.join(f'{k}={v}' for k, v in sorted(by_type.items()))}")
print(f" Statuses: {', '.join(f'{k}={v}' for k, v in sorted(by_status.items()))}")


# ---------------------------------------------------------------------------
# Helpers
Expand Down
82 changes: 82 additions & 0 deletions src/ai_memory_protocol/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@

import json
import os
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Any
Expand Down Expand Up @@ -167,3 +169,83 @@ def expand_graph(
frontier = next_frontier

return {nid: needs[nid] for nid in collected if nid in needs}


# ---------------------------------------------------------------------------
# Sphinx-build discovery and rebuild
# ---------------------------------------------------------------------------


def find_sphinx_build(workspace: Path) -> str:
"""Locate the sphinx-build executable.

Search order:
1. ``workspace/.venv/bin/sphinx-build``
2. Walk parent directories for ``.venv/bin/sphinx-build``
3. ``shutil.which("sphinx-build")`` (system PATH)

Returns the path string, or raises ``FileNotFoundError``.
"""
# 1. Workspace venv
candidate = workspace / ".venv" / "bin" / "sphinx-build"
if candidate.exists():
return str(candidate)

# 2. Walk parent directories
for parent in workspace.parents:
candidate = parent / ".venv" / "bin" / "sphinx-build"
if candidate.exists():
return str(candidate)
# Also check sibling directories (e.g., ../ros2_medkit/.venv/)
if parent.is_dir():
for sibling in parent.iterdir():
if sibling.is_dir() and sibling != workspace:
candidate = sibling / ".venv" / "bin" / "sphinx-build"
if candidate.exists():
return str(candidate)
Comment on lines +183 to +205
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The venv candidate path is hard-coded to .venv/bin/sphinx-build, which will not exist on Windows venvs (typically .venv/Scripts/sphinx-build.exe). If this package is intended to be cross-platform, consider checking both bin/ and Scripts/ (and .exe) before falling back to shutil.which.

Suggested change
1. ``workspace/.venv/bin/sphinx-build``
2. Walk parent directories for ``.venv/bin/sphinx-build``
3. ``shutil.which("sphinx-build")`` (system PATH)
Returns the path string, or raises ``FileNotFoundError``.
"""
# 1. Workspace venv
candidate = workspace / ".venv" / "bin" / "sphinx-build"
if candidate.exists():
return str(candidate)
# 2. Walk parent directories
for parent in workspace.parents:
candidate = parent / ".venv" / "bin" / "sphinx-build"
if candidate.exists():
return str(candidate)
# Also check sibling directories (e.g., ../ros2_medkit/.venv/)
if parent.is_dir():
for sibling in parent.iterdir():
if sibling.is_dir() and sibling != workspace:
candidate = sibling / ".venv" / "bin" / "sphinx-build"
if candidate.exists():
return str(candidate)
1. ``workspace/.venv/bin/sphinx-build`` (and Windows equivalents)
2. Walk parent directories for ``.venv/bin/sphinx-build`` (and Windows equivalents)
3. ``shutil.which("sphinx-build")`` (system PATH)
Returns the path string, or raises ``FileNotFoundError``.
"""
def _venv_sphinx_candidate(base: Path) -> str | None:
"""Return the first existing sphinx-build path under base/.venv, or None.
Handles common POSIX and Windows virtualenv layouts.
"""
relative_candidates = [
Path(".venv") / "bin" / "sphinx-build",
Path(".venv") / "Scripts" / "sphinx-build",
Path(".venv") / "Scripts" / "sphinx-build.exe",
]
for rel in relative_candidates:
candidate = base / rel
if candidate.exists():
return str(candidate)
return None
# 1. Workspace venv
venv_path = _venv_sphinx_candidate(workspace)
if venv_path is not None:
return venv_path
# 2. Walk parent directories
for parent in workspace.parents:
venv_path = _venv_sphinx_candidate(parent)
if venv_path is not None:
return venv_path
# Also check sibling directories (e.g., ../ros2_medkit/.venv/)
if parent.is_dir():
for sibling in parent.iterdir():
if sibling.is_dir() and sibling != workspace:
venv_path = _venv_sphinx_candidate(sibling)
if venv_path is not None:
return venv_path

Copilot uses AI. Check for mistakes.
break # Only check immediate parent's siblings

Comment on lines +195 to +207
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

find_sphinx_build() claims to “walk parent directories”, but the unconditional break after checking the immediate parent’s siblings stops the loop after the first parent. This means venvs in higher ancestors (e.g., grandparent /.venv/) will never be discovered; remove the loop break (or gate sibling scanning to only the first parent) so the parent walk actually continues.

Suggested change
for parent in workspace.parents:
candidate = parent / ".venv" / "bin" / "sphinx-build"
if candidate.exists():
return str(candidate)
# Also check sibling directories (e.g., ../ros2_medkit/.venv/)
if parent.is_dir():
for sibling in parent.iterdir():
if sibling.is_dir() and sibling != workspace:
candidate = sibling / ".venv" / "bin" / "sphinx-build"
if candidate.exists():
return str(candidate)
break # Only check immediate parent's siblings
for i, parent in enumerate(workspace.parents):
candidate = parent / ".venv" / "bin" / "sphinx-build"
if candidate.exists():
return str(candidate)
# Also check sibling directories of the immediate parent
if i == 0 and parent.is_dir():
for sibling in parent.iterdir():
if sibling.is_dir() and sibling != workspace:
candidate = sibling / ".venv" / "bin" / "sphinx-build"
if candidate.exists():
return str(candidate)

Copilot uses AI. Check for mistakes.
# 3. System PATH
system = shutil.which("sphinx-build")
if system:
return system

raise FileNotFoundError(
"sphinx-build not found. Install it with: pip install sphinx sphinx-needs\n"
"Or create a venv in your memory workspace: memory init --install <dir>"
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The FileNotFoundError guidance says memory init --install <dir>, but the CLI syntax elsewhere is memory init <dir> --install (dir is the positional argument). Updating this message would prevent users from copy/pasting an invalid command.

Suggested change
"Or create a venv in your memory workspace: memory init --install <dir>"
"Or create a venv in your memory workspace: memory init <dir> --install"

Copilot uses AI. Check for mistakes.
)


def run_rebuild(workspace: Path) -> tuple[bool, str]:
"""Run Sphinx build to regenerate needs.json.

Returns ``(success, message)`` — never raises on build failure.
"""
try:
sphinx_cmd = find_sphinx_build(workspace)
except FileNotFoundError as e:
return False, f"Rebuild skipped: {e}"

cmd = [sphinx_cmd, "-b", "html", "-q", str(workspace), str(workspace / "_build" / "html")]
try:
result = subprocess.run(cmd, capture_output=True, text=True)
except OSError as e:
return False, f"Rebuild failed: {e}"

if result.returncode != 0:
return False, f"Rebuild failed:\n{result.stderr}"

needs = load_needs(workspace)
by_type: dict[str, int] = {}
by_status: dict[str, int] = {}
for n in needs.values():
by_type[n.get("type", "?")] = by_type.get(n.get("type", "?"), 0) + 1
by_status[n.get("status", "?")] = by_status.get(n.get("status", "?"), 0) + 1

Comment on lines +238 to +244
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

run_rebuild() documents that it “never raises on build failure”, but after a successful Sphinx run it calls load_needs(), which can sys.exit(1) (SystemExit) if needs.json is missing/relocated. To match the contract and avoid crashing CLI/MCP callers, catch SystemExit around load_needs() (and return (False, <message>)) or refactor load_needs() to optionally raise/return an error instead of exiting.

Copilot uses AI. Check for mistakes.
lines = [
f"needs.json updated at {find_needs_json(workspace)}",
f"Total: {len(needs)} memories",
f" Types: {', '.join(f'{k}={v}' for k, v in sorted(by_type.items()))}",
f" Statuses: {', '.join(f'{k}={v}' for k, v in sorted(by_status.items()))}",
]
return True, "\n".join(lines)
33 changes: 17 additions & 16 deletions src/ai_memory_protocol/mcp_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,15 +15,22 @@

import json
import logging
import subprocess
from datetime import date
from pathlib import Path
from typing import Any

from mcp.server import Server
from mcp.types import TextContent, Tool

from .engine import expand_graph, find_workspace, load_needs, resolve_id, tag_match, text_match
from .engine import (
expand_graph,
find_workspace,
load_needs,
resolve_id,
run_rebuild,
tag_match,
text_match,
)
from .formatter import format_brief, format_compact, format_context_pack, format_full
from .rst import (
add_tags_in_rst,
Expand Down Expand Up @@ -385,18 +392,8 @@ def _format_output(

def _do_rebuild(workspace: Path) -> str:
"""Run Sphinx build to regenerate needs.json."""
venv_sphinx = workspace / ".venv" / "bin" / "sphinx-build"
sphinx_cmd = str(venv_sphinx) if venv_sphinx.exists() else "sphinx-build"
cmd = [sphinx_cmd, "-b", "html", "-q", str(workspace), str(workspace / "_build" / "html")]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
return f"Rebuild failed: {result.stderr}"

needs = load_needs(workspace)
by_type: dict[str, int] = {}
for n in needs.values():
by_type[n.get("type", "?")] = by_type.get(n.get("type", "?"), 0) + 1
return f"Rebuilt successfully. {len(needs)} memories: {', '.join(f'{k}={v}' for k, v in sorted(by_type.items()))}"
success, message = run_rebuild(workspace)
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

success is assigned but never used here. With ruff/pyflakes enabled (F841), this will fail lint; use _/_success or return based on it.

Suggested change
success, message = run_rebuild(workspace)
_, message = run_rebuild(workspace)

Copilot uses AI. Check for mistakes.
return message


def _text_response(text: str) -> list[TextContent]:
Expand Down Expand Up @@ -534,8 +531,12 @@ def _handle_add(args: dict[str, Any]) -> list[TextContent]:
result_lines = [f"Added {nid} → {target.name}"]

if args.get("rebuild", True):
rebuild_result = _do_rebuild(workspace)
result_lines.append(rebuild_result)
success, rebuild_msg = run_rebuild(workspace)
if success:
result_lines.append(rebuild_msg)
else:
result_lines.append(f"Warning: Memory was added but rebuild failed: {rebuild_msg}")
result_lines.append("Run memory_rebuild manually when sphinx-build is available.")
Comment on lines +538 to +539
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When run_rebuild() returns False it may return messages like Rebuild skipped: ... (e.g., sphinx-build not found). Prefixing it with rebuild failed makes the result misleading/confusing ("failed: skipped"). Consider using a neutral prefix (e.g., "rebuild did not run"), or just append rebuild_msg without re-labeling it.

Suggested change
result_lines.append(f"Warning: Memory was added but rebuild failed: {rebuild_msg}")
result_lines.append("Run memory_rebuild manually when sphinx-build is available.")
result_lines.append(
f"Warning: Memory was added but rebuild did not complete automatically: {rebuild_msg}"
)
result_lines.append("You can run memory_rebuild manually when sphinx-build is available.")

Copilot uses AI. Check for mistakes.

return _text_response("\n".join(result_lines))

Expand Down
Loading