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

Fix bd-8zf2: MCP server auto-detects workspace from CWD

Browse Source 
- Add _find_beads_db_in_tree() to walk up looking for .beads/*.db
- Update _get_client() to auto-detect when workspace not set
- Matches CLI behavior (no manual set_context needed after restart)
- Add 8 comprehensive tests for auto-detection
- Update existing tests to mock auto-detection in error cases

Fixes silent failures after Amp restart.

Amp-Thread-ID: https://ampcode.com/threads/T-c47f524d-c101-40d5-839a-659f52b9be48
Co-authored-by: Amp <amp@ampcode.com>
This commit is contained in:
Steve Yegge
2025-11-07 23:59:57 -08:00
parent e6e458fc40
commit 44179d7326
4 changed files with 228 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -63,6 +63,50 @@ def _register_client_for_cleanup(client: BdClientBase) -> None:
pass
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).
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)
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
def _resolve_workspace_root(path: str) -> str:
"""Resolve workspace root to git repo root if inside a git repo.
@@ -181,8 +225,11 @@ async def _get_client() -> BdClientBase:
"""Get a BdClient instance for the current workspace.
Uses connection pool to manage per-project daemon sockets.
Workspace is determined by current_workspace ContextVar or BEADS_WORKING_DIR env.
Workspace is auto-detected using the same logic as CLI:
1. current_workspace ContextVar (from workspace_root parameter)
2. BEADS_WORKING_DIR environment variable
3. Walk up from CWD looking for .beads/*.db
Performs health check before returning cached client.
On failure, drops from pool and attempts reconnection with exponential backoff.
@@ -193,13 +240,23 @@ async def _get_client() -> BdClientBase:
Configured BdClientBase instance for the current workspace
Raises:
BdError: If no workspace is set, or bd is not installed, or version is incompatible
BdError: If no workspace found, or bd is not installed, or version is incompatible
"""
# Determine workspace from ContextVar or environment
# Determine workspace using standard search order (matches Go CLI)
workspace = current_workspace.get() or os.environ.get("BEADS_WORKING_DIR")
# Auto-detect from CWD if not explicitly set (NEW!)
if not workspace:
workspace = _find_beads_db_in_tree()
if workspace:
logger.debug(f"Auto-detected workspace from CWD: {workspace}")
if not workspace:
raise BdError(
"No workspace set. Either provide workspace_root parameter or call set_context() first."
"No beads workspace found. Either:\n"
" 1. Call set_context(workspace_root=\"/path/to/project\"), OR\n"
" 2. Run from a directory containing .beads/, OR\n"
" 3. Set BEADS_WORKING_DIR environment variable"
)
# Canonicalize path to handle symlinks and deduplicate connections

6
integrations/beads-mcp/tests/test_daemon_health_check.py
View File

@@ -299,5 +299,7 @@ async def test_get_client_no_workspace_error():
tools.current_workspace.set(None)
with patch.dict('os.environ', {}, clear=True):
with pytest.raises(BdError, match="No workspace set"):
await _get_client()
# Mock auto-detection to fail
with patch("beads_mcp.tools._find_beads_db_in_tree", return_value=None):
with pytest.raises(BdError, match="No beads workspace found"):
await _get_client()

10
integrations/beads-mcp/tests/test_multi_project_switching.py
View File

@@ -430,16 +430,18 @@ class TestEdgeCases:
async def test_no_workspace_raises_error(self):
"""Test calling without workspace raises helpful error."""
import os
from beads_mcp import tools
# Clear env
# Clear context and env
tools.current_workspace.set(None)
os.environ.pop("BEADS_WORKING_DIR", None)
# No ContextVar set, no env var
# No ContextVar set, no env var, and auto-detect fails
with pytest.raises(Exception) as exc_info:
with patch("beads_mcp.tools.create_bd_client") as mock_create:
with patch("beads_mcp.tools._find_beads_db_in_tree", return_value=None):
await beads_ready_work()
assert "No workspace set" in str(exc_info.value)
assert "No beads workspace found" in str(exc_info.value)
def test_canonicalize_path_cached(self, temp_projects):
"""Test path canonicalization is cached for performance."""

156
integrations/beads-mcp/tests/test_workspace_auto_detect.py Normal file
View File

@@ -0,0 +1,156 @@
"""Test workspace auto-detection from CWD (bd-8zf2)."""
import os
import pytest
import tempfile
from pathlib import Path
from unittest.mock import AsyncMock, patch
from beads_mcp.tools import _find_beads_db_in_tree, _get_client, current_workspace
from beads_mcp.bd_client import BdError
def test_find_beads_db_in_tree_direct():
"""Test finding .beads in current directory."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create .beads/beads.db
beads_dir = Path(tmpdir) / ".beads"
beads_dir.mkdir()
(beads_dir / "beads.db").touch()
# Should find workspace root (use realpath for macOS symlink resolution)
result = _find_beads_db_in_tree(tmpdir)
assert result == os.path.realpath(tmpdir)
def test_find_beads_db_in_tree_parent():
"""Test finding .beads in parent directory."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create .beads/beads.db in root
beads_dir = Path(tmpdir) / ".beads"
beads_dir.mkdir()
(beads_dir / "beads.db").touch()
# Create subdirectory
subdir = Path(tmpdir) / "subdir" / "deep"
subdir.mkdir(parents=True)
# Should find workspace root (walks up from subdir)
result = _find_beads_db_in_tree(str(subdir))
assert result == os.path.realpath(tmpdir)
def test_find_beads_db_in_tree_not_found():
"""Test when no .beads directory exists."""
with tempfile.TemporaryDirectory() as tmpdir:
# No .beads directory
result = _find_beads_db_in_tree(tmpdir)
assert result is None
def test_find_beads_db_excludes_backups():
"""Test that backup .db files are ignored."""
with tempfile.TemporaryDirectory() as tmpdir:
beads_dir = Path(tmpdir) / ".beads"
beads_dir.mkdir()
# Only backup file exists
(beads_dir / "beads.db.backup").touch()
result = _find_beads_db_in_tree(tmpdir)
assert result is None # Should not find backup files
# Add valid db file
(beads_dir / "beads.db").touch()
result = _find_beads_db_in_tree(tmpdir)
assert result == os.path.realpath(tmpdir)
@pytest.mark.asyncio
async def test_get_client_auto_detect_from_cwd():
"""Test that _get_client() auto-detects workspace from CWD."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create .beads/beads.db
beads_dir = Path(tmpdir) / ".beads"
beads_dir.mkdir()
(beads_dir / "beads.db").touch()
# Reset ContextVar for this test
token = current_workspace.set(None)
try:
with patch.dict(os.environ, {}, clear=True):
# Mock _find_beads_db_in_tree to return our tmpdir
with patch("beads_mcp.tools._find_beads_db_in_tree", return_value=tmpdir):
# Mock create_bd_client to avoid actual connection
mock_client = AsyncMock()
mock_client.ping = AsyncMock(return_value=None)
with patch("beads_mcp.tools.create_bd_client", return_value=mock_client):
# Should auto-detect and not raise error
client = await _get_client()
assert client is not None
finally:
current_workspace.reset(token)
@pytest.mark.asyncio
async def test_get_client_no_workspace_found():
"""Test that _get_client() raises helpful error when no workspace found."""
# Reset ContextVar for this test
token = current_workspace.set(None)
try:
with patch.dict(os.environ, {}, clear=True):
with patch("beads_mcp.tools._find_beads_db_in_tree", return_value=None):
with pytest.raises(BdError) as exc_info:
await _get_client()
# Verify error message is helpful
error_msg = str(exc_info.value)
assert "No beads workspace found" in error_msg
assert "set_context" in error_msg
assert ".beads/" in error_msg
finally:
current_workspace.reset(token)
@pytest.mark.asyncio
async def test_get_client_prefers_context_var_over_auto_detect():
"""Test that explicit workspace_root parameter takes precedence."""
explicit_workspace = "/explicit/path"
token = current_workspace.set(explicit_workspace)
try:
with patch("beads_mcp.tools._canonicalize_path", return_value=explicit_workspace):
mock_client = AsyncMock()
mock_client.ping = AsyncMock(return_value=None)
with patch("beads_mcp.tools.create_bd_client", return_value=mock_client) as mock_create:
client = await _get_client()
# Should use explicit workspace, not auto-detect
mock_create.assert_called_once()
# The working_dir parameter should be the canonicalized explicit path
assert mock_create.call_args[1]["working_dir"] == explicit_workspace
finally:
current_workspace.reset(token)
@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)