johno/beads
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: MCP plugin follows .beads/redirect files (bd-7t9a, gt-tnw)

Browse Source 
The _find_beads_db_in_tree() function now follows .beads/redirect files
to find shared beads databases. This is essential for polecat/crew
directories that use redirect files to share a central database.

Changes:
- Added _resolve_beads_redirect() helper function
- Updated _find_beads_db_in_tree() to check for redirect files before
  looking for local .db files
- Added comprehensive tests for redirect functionality

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Steve Yegge
2025-12-30 00:05:17 -08:00
parent 99692b52f0
commit e20de54cb2
3 changed files with 190 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
integrations/beads-mcp/src/beads_mcp/tools.py
View File

@@ -64,45 +64,103 @@ def _register_client_for_cleanup(client: BdClientBase) -> None:
pass
def _resolve_beads_redirect(beads_dir: str, workspace_root: str) -> str | None:
"""Follow a .beads/redirect file to the actual beads directory.
Args:
beads_dir: Path to the .beads directory that may contain a redirect
workspace_root: The workspace root directory (parent of beads_dir)
Returns:
Resolved workspace root if redirect is valid, None otherwise
"""
import glob
redirect_path = os.path.join(beads_dir, "redirect")
if not os.path.isfile(redirect_path):
return None
try:
with open(redirect_path, 'r') as f:
redirect_target = f.read().strip()
if not redirect_target:
return None
# Resolve relative to workspace_root (the redirect is written from the perspective
# of being inside workspace_root, not inside workspace_root/.beads)
# e.g., redirect contains "../../mayor/rig/.beads"
# from polecats/capable/, this resolves to mayor/rig/.beads
resolved = os.path.normpath(os.path.join(workspace_root, redirect_target))
if not os.path.isdir(resolved):
logger.debug(f"Redirect target {resolved} does not exist")
return None
# Verify the redirected location has a valid database
db_files = glob.glob(os.path.join(resolved, "*.db"))
valid_dbs = [f for f in db_files if ".backup" not in os.path.basename(f)]
if not valid_dbs:
logger.debug(f"Redirect target {resolved} has no valid .db files")
return None
# Return the workspace root of the redirected location (parent of .beads)
return os.path.dirname(resolved)
except Exception as e:
logger.debug(f"Failed to follow redirect: {e}")
return None
def _find_beads_db_in_tree(start_dir: str | None = None) -> str | None:
"""Walk up directory tree looking for .beads/*.db (matches Go CLI behavior).
Also follows .beads/redirect files to shared beads locations, which is
essential for polecat/crew directories that share a central database.
Args:
start_dir: Starting directory (default: current working directory)
Returns:
Absolute path to workspace root containing .beads/*.db, or None if not found
"""
import glob
try:
current = os.path.abspath(start_dir or os.getcwd())
# Resolve symlinks like Go CLI does
try:
current = os.path.realpath(current)
except Exception:
pass
# Walk up directory tree
while True:
beads_dir = os.path.join(current, ".beads")
if os.path.isdir(beads_dir):
# Find any .db file in .beads/ (excluding backups)
# First, check for redirect file (polecat/crew directories use this)
redirected = _resolve_beads_redirect(beads_dir, current)
if redirected:
logger.debug(f"Followed redirect from {current} to {redirected}")
return redirected
# No redirect, check for local .db files
db_files = glob.glob(os.path.join(beads_dir, "*.db"))
valid_dbs = [f for f in db_files if ".backup" not in os.path.basename(f)]
if valid_dbs:
# Return workspace root (parent of .beads), not the db path
return current
parent = os.path.dirname(current)
if parent == current: # Reached filesystem root
break
current = parent
return None
except Exception as e:
logger.debug(f"Failed to search for .beads in tree: {e}")
return None

125
integrations/beads-mcp/tests/test_workspace_auto_detect.py
View File

@@ -135,22 +135,139 @@ async def test_get_client_prefers_context_var_over_auto_detect():
current_workspace.reset(token)
@pytest.mark.asyncio
@pytest.mark.asyncio
async def test_get_client_env_var_over_auto_detect():
"""Test that BEADS_WORKING_DIR env var takes precedence over auto-detect."""
env_workspace = "/env/path"
token = current_workspace.set(None)
try:
with patch.dict(os.environ, {"BEADS_WORKING_DIR": env_workspace}):
with patch("beads_mcp.tools._canonicalize_path", return_value=env_workspace):
mock_client = AsyncMock()
mock_client.ping = AsyncMock(return_value=None)
with patch("beads_mcp.tools.create_bd_client", return_value=mock_client):
client = await _get_client()
# Should use env var, not call auto-detect
assert client is not None
finally:
current_workspace.reset(token)
def test_find_beads_db_follows_redirect():
"""Test that _find_beads_db_in_tree follows .beads/redirect files (gt-tnw).
This is essential for polecat/crew directories that use redirect files
to share a central beads database.
"""
with tempfile.TemporaryDirectory() as tmpdir:
# Create main workspace with actual database
main_dir = Path(tmpdir) / "mayor" / "rig"
main_dir.mkdir(parents=True)
main_beads = main_dir / ".beads"
main_beads.mkdir()
(main_beads / "beads.db").touch()
# Create polecat directory with redirect
polecat_dir = Path(tmpdir) / "polecats" / "capable"
polecat_dir.mkdir(parents=True)
polecat_beads = polecat_dir / ".beads"
polecat_beads.mkdir()
# Write redirect file (relative path from polecat dir)
redirect_file = polecat_beads / "redirect"
redirect_file.write_text("../../mayor/rig/.beads")
# Should find workspace via redirect
result = _find_beads_db_in_tree(str(polecat_dir))
assert result == os.path.realpath(str(main_dir))
def test_find_beads_db_redirect_invalid_target():
"""Test that invalid redirect targets are handled gracefully."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create polecat directory with redirect to nonexistent location
polecat_dir = Path(tmpdir) / "polecats" / "capable"
polecat_dir.mkdir(parents=True)
polecat_beads = polecat_dir / ".beads"
polecat_beads.mkdir()
# Write redirect file pointing to nonexistent location
redirect_file = polecat_beads / "redirect"
redirect_file.write_text("../../nonexistent/.beads")
# Should return None (graceful failure)
result = _find_beads_db_in_tree(str(polecat_dir))
assert result is None
def test_find_beads_db_redirect_empty():
"""Test that empty redirect files are handled gracefully."""
with tempfile.TemporaryDirectory() as tmpdir:
polecat_dir = Path(tmpdir) / "polecats" / "capable"
polecat_dir.mkdir(parents=True)
polecat_beads = polecat_dir / ".beads"
polecat_beads.mkdir()
# Write empty redirect file
redirect_file = polecat_beads / "redirect"
redirect_file.write_text("")
# Should return None (graceful failure)
result = _find_beads_db_in_tree(str(polecat_dir))
assert result is None
def test_find_beads_db_redirect_no_db_at_target():
"""Test that redirects to directories without .db files are handled."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create main workspace WITHOUT database
main_dir = Path(tmpdir) / "mayor" / "rig"
main_dir.mkdir(parents=True)
main_beads = main_dir / ".beads"
main_beads.mkdir() # No .db file!
# Create polecat directory with redirect
polecat_dir = Path(tmpdir) / "polecats" / "capable"
polecat_dir.mkdir(parents=True)
polecat_beads = polecat_dir / ".beads"
polecat_beads.mkdir()
redirect_file = polecat_beads / "redirect"
redirect_file.write_text("../../mayor/rig/.beads")
# Should return None (redirect target has no valid db)
result = _find_beads_db_in_tree(str(polecat_dir))
assert result is None
def test_find_beads_db_prefers_redirect_over_parent():
"""Test that redirect in current dir is followed before walking up."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create two beads locations
# 1. Parent directory with its own database
parent_dir = Path(tmpdir)
parent_beads = parent_dir / ".beads"
parent_beads.mkdir()
(parent_beads / "beads.db").touch()
# 2. Remote directory that redirect points to
remote_dir = Path(tmpdir) / "remote"
remote_dir.mkdir()
remote_beads = remote_dir / ".beads"
remote_beads.mkdir()
(remote_beads / "beads.db").touch()
# Create child directory with redirect to remote
child_dir = Path(tmpdir) / "child"
child_dir.mkdir()
child_beads = child_dir / ".beads"
child_beads.mkdir()
redirect_file = child_beads / "redirect"
redirect_file.write_text("../remote/.beads")
# Should follow redirect (to remote), not walk up to parent
result = _find_beads_db_in_tree(str(child_dir))
assert result == os.path.realpath(str(remote_dir))