Skip to content
Open
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
20 changes: 15 additions & 5 deletions src/dippy/core/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,11 @@
# Cache home directory at module load - fails fast if HOME is unset
_HOME = Path.home()


def _posix_str(path: str) -> str:
"""Normalize path separators to forward slashes for consistent matching."""
return path.replace("\\", "/")

USER_CONFIG = _HOME / ".dippy" / "config"
PROJECT_CONFIG_NAME = ".dippy"
ENV_CONFIG = "DIPPY_CONFIG"
Expand Down Expand Up @@ -458,6 +463,9 @@ def _classify_token(token: str) -> str:
return _VARIABLE
if token.startswith("/"):
return _ABSOLUTE
# Windows drive letter paths: C:\, C:/, D:\, etc.
if len(token) >= 3 and token[0].isalpha() and token[1] == ":" and token[2] in ("/", "\\"):
return _ABSOLUTE
if token == "~" or token.startswith("~/"):
return _HOME
if token.startswith("~"):
Expand Down Expand Up @@ -490,17 +498,18 @@ def _expand_token(token: str, cwd: Path, *, force_path: bool = False) -> str:
if kind == _VARIABLE:
return token
if kind == _ABSOLUTE:
return token
return _posix_str(token)
if kind == _HOME:
# ~ → /home/user, ~/foo → /home/user/foo
return str(home) + token[1:] if len(token) > 1 else str(home)
h = _posix_str(str(home))
return h + token[1:] if len(token) > 1 else h
if kind == _USER_HOME:
return token
if kind == _RELATIVE:
return str((cwd / token).resolve())
return _posix_str(str((cwd / token).resolve()))
# BARE
if force_path:
return str((cwd / token).resolve())
return _posix_str(str((cwd / token).resolve()))
return token


Expand All @@ -512,7 +521,8 @@ def _expand_home_only(token: str) -> str:
"""
if _classify_token(token) == _HOME:
home = Path.home()
return str(home) + token[1:] if len(token) > 1 else str(home)
h = _posix_str(str(home))
return h + token[1:] if len(token) > 1 else h
return token


Expand Down
63 changes: 54 additions & 9 deletions src/dippy/dippy_statusline.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import shutil
import subprocess
import sys
import threading
import time
import traceback
from datetime import datetime, timezone
Expand Down Expand Up @@ -216,6 +217,35 @@ def get_local_mcp_servers() -> list[str]:
return []


def _refresh_mcp_cache_python(tmp, conn_r, conn_g, conn_b, disc_r, disc_g, disc_b):
"""Refresh MCP cache using pure Python (Windows-compatible)."""
try:
result = subprocess.run(
["claude", "mcp", "list"],
capture_output=True, text=True, timeout=10,
)
lines = []
for line in result.stdout.splitlines():
if ":" not in line:
continue
name = line.split(":")[0].strip()
if not name:
continue
if "Connected" in line:
lines.append(f"\033[38;2;{conn_r};{conn_g};{conn_b}m{name}\033[0m")
else:
lines.append(f"\033[38;2;{disc_r};{disc_g};{disc_b}m!{name}\033[0m")
with open(tmp, "w") as f:
f.write(", ".join(lines))
os.replace(tmp, MCP_CACHE_PATH)
except Exception:
log.error("mcp_cache_refresh_python_failed")
try:
os.unlink(tmp)
except OSError:
pass


def get_mcp_servers() -> str | None:
"""Read MCP servers from local config and cached global list."""
local_servers = get_local_mcp_servers()
Expand All @@ -242,15 +272,23 @@ def get_mcp_servers() -> str | None:
os.makedirs(CACHE_DIR, exist_ok=True)
tmp = f"{MCP_CACHE_PATH}.tmp.{os.getpid()}"
disc_r, disc_g, disc_b = hex_to_rgb(MOLOKAI[STYLES["mcp_disconnected"][0]])
cmd = f'timeout 10 claude mcp list 2>/dev/null | awk -F: \'NF>1 {{if (/Connected/) print "\\033[38;2;{conn_r};{conn_g};{conn_b}m" $1 "\\033[0m"; else print "\\033[38;2;{disc_r};{disc_g};{disc_b}m!" $1 "\\033[0m"}}\' | paste -sd, | sed \'s/,/, /g\' > {tmp} && mv {tmp} {MCP_CACHE_PATH}'
subprocess.Popen(
cmd,
shell=True,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
if sys.platform == "win32":
t = threading.Thread(
target=_refresh_mcp_cache_python,
args=(tmp, conn_r, conn_g, conn_b, disc_r, disc_g, disc_b),
daemon=True,
)
t.start()
else:
cmd = f'timeout 10 claude mcp list 2>/dev/null | awk -F: \'NF>1 {{if (/Connected/) print "\\033[38;2;{conn_r};{conn_g};{conn_b}m" $1 "\\033[0m"; else print "\\033[38;2;{disc_r};{disc_g};{disc_b}m!" $1 "\\033[0m"}}\' | paste -sd, | sed \'s/,/, /g\' > {tmp} && mv {tmp} {MCP_CACHE_PATH}'
subprocess.Popen(
cmd,
shell=True,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
log.debug("mcp_cache_refresh_spawned", age=age, ttl=MCP_CACHE_TTL)
except Exception:
log.error("mcp_cache_refresh_failed")
Expand Down Expand Up @@ -462,6 +500,13 @@ def build_statusline(data: dict) -> str:

def main():
log.info("main_start")
# Ensure stdout can handle Unicode (emoji) on Windows where
# subprocess stdout defaults to cp1252
if hasattr(sys.stdout, "reconfigure"):
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
try:
data = json.load(sys.stdin)
log.debug("main_input_parsed", session_id=data.get("session_id", ""))
Expand Down
10 changes: 7 additions & 3 deletions tests/test_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import json
import os
import stat
import sys
from pathlib import Path

import pytest
Expand Down Expand Up @@ -217,6 +218,9 @@ def test_env_config_tilde_expansion(self, tmp_path, monkeypatch):
fake_home.mkdir()
(fake_home / "my.cfg").write_text("allow tilde")
monkeypatch.setenv("HOME", str(fake_home))
# On Windows, Path.expanduser() uses USERPROFILE, not HOME
if sys.platform == "win32":
monkeypatch.setenv("USERPROFILE", str(fake_home))
monkeypatch.setenv("DIPPY_CONFIG", "~/my.cfg")

def mock_parse(text, source=None):
Expand Down Expand Up @@ -1305,8 +1309,8 @@ def test_tilde_expansion_consistency(self, tmp_path):
home = Path.home()
# Settings expand tilde correctly
assert cfg.log == home / "audit.log"
# Rule patterns also expand tilde consistently
assert cfg.rules[0].pattern == f"cd {home}"
# Rule patterns also expand tilde consistently (forward slashes for matching)
assert cfg.rules[0].pattern == f"cd {str(home).replace(chr(92), '/')}"

def test_url_not_mangled(self, tmp_path):
"""URLs should not be turned into paths under cwd."""
Expand Down Expand Up @@ -1821,7 +1825,7 @@ class TestAlias:

def test_alias_tilde_path(self, tmp_path):
"""alias ~/bin/gh gh + allow gh matches ~/bin/gh pr list."""
home = str(Path.home())
home = str(Path.home()).replace("\\", "/")
cfg = parse_config("alias ~/bin/gh gh\nallow gh")
assert cfg.aliases == {f"{home}/bin/gh": "gh"}
c = SimpleCommand(words=["~/bin/gh", "pr", "list"])
Expand Down
10 changes: 7 additions & 3 deletions tests/test_entrypoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,11 @@
import tempfile
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent
DIPPY_HOOK = REPO_ROOT / "bin" / "dippy-hook"
# Use system Python to avoid venv masking import issues
SYSTEM_PYTHON = "/usr/bin/python3"
IS_WINDOWS = sys.platform == "win32"


def get_decision(output: dict) -> str | None:
Expand Down Expand Up @@ -46,7 +46,9 @@ def _run(
else:
stdin_bytes = input_data.encode()

python = SYSTEM_PYTHON if use_system_python else sys.executable
# use_system_python is for testing path resolution without venv;
# on Windows there's no /usr/bin/python3, so always use sys.executable
python = sys.executable
return subprocess.run(
[python, str(script)],
input=stdin_bytes,
Expand All @@ -66,6 +68,7 @@ def test_direct_invocation(self):
output = json.loads(result.stdout)
assert get_decision(output) == "allow"

@pytest.mark.skipif(IS_WINDOWS, reason="symlinks require elevated privileges on Windows")
def test_symlink_invocation(self):
"""Critical: invocation via symlink must also work.

Expand All @@ -78,6 +81,7 @@ def test_symlink_invocation(self):
output = json.loads(result.stdout)
assert get_decision(output) == "allow"

@pytest.mark.skipif(IS_WINDOWS, reason="symlinks require elevated privileges on Windows")
def test_nested_symlink_invocation(self):
"""Symlink in deeply nested unrelated path (simulates Homebrew Cellar)."""
input_data = {"tool_name": "Bash", "tool_input": {"command": "ls"}}
Expand Down
9 changes: 7 additions & 2 deletions tests/test_statusline.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,14 +3,18 @@
from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
import uuid
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent
DIPPY_STATUSLINE = REPO_ROOT / "bin" / "dippy-statusline"
SYSTEM_PYTHON = "/usr/bin/python3"
IS_WINDOWS = sys.platform == "win32"


def unique_session_id() -> str:
Expand Down Expand Up @@ -39,7 +43,7 @@ def run_statusline(
script = symlink_path

return subprocess.run(
[SYSTEM_PYTHON, str(script)],
[sys.executable, str(script)],
input=stdin_bytes,
capture_output=True,
timeout=10,
Expand All @@ -63,6 +67,7 @@ def test_valid_input(self):
# Should contain model name
assert "Opus" in output

@pytest.mark.skipif(IS_WINDOWS, reason="symlinks require elevated privileges on Windows")
def test_symlink_invocation(self):
"""Works when invoked via symlink (Homebrew scenario)."""
input_data = {
Expand Down
Loading