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
160 changes: 160 additions & 0 deletions memanto/cli/connect/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,6 +121,24 @@ def remove_agent(
except Exception as e:
errors.append(f"Skill removal: {e}")

# Remove hook configuration (only Claude Code currently)
if agent.supports_hooks and agent.hook_config:
try:
hook_result = _remove_hooks(agent, project_path, is_global)
if hook_result:
steps.append(hook_result)
except Exception as e:
errors.append(f"Hook removal: {e}")

# Remove permissions added by MEMANTO
if agent.permissions_file and agent.permissions_payload:
try:
perm_result = _remove_permissions(agent, project_path, is_global)
if perm_result:
steps.append(perm_result)
except Exception as e:
errors.append(f"Permission removal: {e}")

try:
ConfigManager().remove_connection(
agent_name, str(project_path) if not is_global else None, is_global
Expand Down Expand Up @@ -349,6 +367,83 @@ def _install_hooks(agent: AgentDef, project_path: Path, is_global: bool) -> str
return None # Already configured


def _remove_hooks(agent: AgentDef, project_path: Path, is_global: bool) -> str | None:
"""Remove MEMANTO hook configuration for agents that support hooks."""
if not agent.hook_config:
return None

if is_global:
if agent.config_global_dir:
config_dir = Path.home() / agent.config_global_dir.lstrip("~/")
else:
return None
else:
if agent.config_local_dir:
config_dir = project_path / agent.config_local_dir
else:
return None

settings_path = config_dir / agent.hook_config.settings_file
if not settings_path.exists():
return None

settings = json.loads(settings_path.read_text(encoding="utf-8"))
hooks = settings.get("hooks")
session_start = hooks.get("SessionStart") if isinstance(hooks, dict) else None
if not isinstance(session_start, list):
return None

expected_payload = agent.hook_config.hook_payload
expected_matcher = expected_payload.get("matcher")
expected_hooks = [
hook for hook in expected_payload.get("hooks", []) if isinstance(hook, dict)
]
expected_commands = {
hook.get("command") for hook in expected_hooks if hook.get("command")
}
if not expected_commands:
return None

changed = False
next_session_start = []
for group in session_start:
if not isinstance(group, dict) or not isinstance(group.get("hooks"), list):
next_session_start.append(group)
continue

remaining_hooks = []
for hook in group["hooks"]:
if (
group.get("matcher") == expected_matcher
and isinstance(hook, dict)
and hook.get("command") in expected_commands
and any(hook == expected_hook for expected_hook in expected_hooks)
):
changed = True
continue
Comment thread
coderabbitai[bot] marked this conversation as resolved.
remaining_hooks.append(hook)

if remaining_hooks:
updated_group = dict(group)
updated_group["hooks"] = remaining_hooks
next_session_start.append(updated_group)
else:
changed = True

if not changed:
return None

if next_session_start:
hooks["SessionStart"] = next_session_start
else:
hooks.pop("SessionStart", None)
if not hooks:
settings.pop("hooks", None)

_write_or_remove_json(settings_path, settings)
return "Removed SessionStart hook"


# Internal: Permission configuration


Expand Down Expand Up @@ -397,6 +492,56 @@ def _install_permissions(
return None # Already configured


def _remove_permissions(
agent: AgentDef, project_path: Path, is_global: bool
) -> str | None:
"""Remove permissions added by MEMANTO without disturbing user entries."""
if not agent.permissions_file or not agent.permissions_payload:
return None

if is_global:
if agent.config_global_dir:
config_dir = Path.home() / agent.config_global_dir.lstrip("~/")
else:
return None
perm_path = config_dir / agent.permissions_file
else:
if agent.config_local_dir:
config_dir = project_path / agent.config_local_dir
else:
return None
perm_path = config_dir / agent.permissions_file

if not perm_path.exists():
return None

existing = json.loads(perm_path.read_text(encoding="utf-8"))
permissions = existing.get("permissions")
allow_list = permissions.get("allow") if isinstance(permissions, dict) else None
if not isinstance(allow_list, list):
return None

expected_permissions = set(
agent.permissions_payload.get("permissions", {}).get("allow", [])
)
if not expected_permissions:
return None

next_allow = [perm for perm in allow_list if perm not in expected_permissions]
if len(next_allow) == len(allow_list):
return None

if next_allow:
permissions["allow"] = next_allow
else:
permissions.pop("allow", None)
if isinstance(permissions, dict) and not permissions:
existing.pop("permissions", None)

_write_or_remove_json(perm_path, existing)
return "Removed permissions"


# Utilities


Expand All @@ -408,3 +553,18 @@ def _display_path(path: Path, is_global: bool) -> str:
return str(path)
except ValueError:
return str(path)


def _write_or_remove_json(path: Path, data: dict[str, Any]) -> None:
"""Persist JSON data, or remove the file when the managed data was all it had."""
if data:
path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
return

path.unlink()
parent = path.parent
try:
if parent.exists() and not any(parent.iterdir()):
parent.rmdir()
except Exception:
pass
135 changes: 135 additions & 0 deletions tests/test_connect_engine.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,135 @@
import json

from memanto.cli.connect import engine


class DummyConfigManager:
def add_connection(self, *args, **kwargs):
return None

def remove_connection(self, *args, **kwargs):
return None


def stub_config_manager(monkeypatch):
monkeypatch.setattr(engine, "ConfigManager", lambda: DummyConfigManager())


def read_json(path):
return json.loads(path.read_text(encoding="utf-8"))


def test_remove_claude_code_removes_installed_hook_and_permissions(
tmp_path, monkeypatch
):
stub_config_manager(monkeypatch)

install_result = engine.install_agent("claude-code", str(tmp_path))

assert install_result["errors"] == []
assert (tmp_path / ".claude" / "settings.json").exists()
assert (tmp_path / ".claude" / "settings.local.json").exists()

remove_result = engine.remove_agent("claude-code", str(tmp_path))

assert remove_result["errors"] == []
assert any("SessionStart hook" in step for step in remove_result["steps"])
assert any("permissions" in step for step in remove_result["steps"])
assert not (tmp_path / ".claude" / "settings.json").exists()
assert not (tmp_path / ".claude" / "settings.local.json").exists()


def test_remove_claude_code_preserves_unrelated_hooks_and_permissions(
tmp_path, monkeypatch
):
stub_config_manager(monkeypatch)
claude_dir = tmp_path / ".claude"
claude_dir.mkdir()
settings_path = claude_dir / "settings.json"
permissions_path = claude_dir / "settings.local.json"

settings_path.write_text(
json.dumps(
{
"theme": "dark",
"hooks": {
"SessionStart": [
{
"matcher": "startup",
"hooks": [
{"type": "command", "command": "echo keep"},
{
"type": "command",
"command": "memanto memory sync --project-dir .",
},
{
"type": "command",
"command": "memanto memory sync --project-dir .",
"timeout": 30,
},
],
},
{
"matcher": "manual",
"hooks": [
{
"type": "command",
"command": "memanto memory sync --project-dir .",
"timeout": 30,
}
],
},
{"matcher": "other", "hooks": [{"command": "echo other"}]},
]
},
}
)
+ "\n",
encoding="utf-8",
)
permissions_path.write_text(
json.dumps(
{
"permissions": {
"allow": ["Bash(git status)", "Bash(memanto:*)"],
},
"env": {"KEEP": "1"},
}
)
+ "\n",
encoding="utf-8",
)

remove_result = engine.remove_agent("claude-code", str(tmp_path))

assert remove_result["errors"] == []
settings = read_json(settings_path)
permissions = read_json(permissions_path)
assert settings["theme"] == "dark"
assert settings["hooks"]["SessionStart"] == [
{
"matcher": "startup",
"hooks": [
{"type": "command", "command": "echo keep"},
{
"type": "command",
"command": "memanto memory sync --project-dir .",
},
],
},
{
"matcher": "manual",
"hooks": [
{
"type": "command",
"command": "memanto memory sync --project-dir .",
"timeout": 30,
}
],
},
{"matcher": "other", "hooks": [{"command": "echo other"}]},
]
assert permissions == {
"permissions": {"allow": ["Bash(git status)"]},
"env": {"KEEP": "1"},
}