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
81 changes: 68 additions & 13 deletions src/telegram_codex_bot/hook.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import logging
import os
import re
import shlex
import shutil
import subprocess
import sys
Expand Down Expand Up @@ -87,11 +88,21 @@ def _find_cli_path() -> str:
return "telegram-codex-bot"


def _is_hook_installed(settings: dict) -> bool:
"""Check if telegram-codex-bot hook is already installed in hooks.json.
def _is_telegram_codex_bot_hook_command(command: str) -> bool:
"""Return True if command invokes ``telegram-codex-bot hook``."""
try:
parts = shlex.split(command)
except ValueError:
return command == _HOOK_COMMAND_SUFFIX or command.endswith(
"/" + _HOOK_COMMAND_SUFFIX
)
if len(parts) < 2 or parts[-1] != "hook":
return False
return Path(parts[-2]).name == "telegram-codex-bot"

Detects both 'telegram-codex-bot hook' and full paths like '/path/to/telegram-codex-bot hook'.
"""

def _find_installed_hook(settings: dict) -> dict[str, Any] | None:
"""Find the first installed telegram-codex-bot hook command, if any."""
hooks = settings.get("hooks", {})
session_start = hooks.get("SessionStart", [])

Expand All @@ -103,10 +114,26 @@ def _is_hook_installed(settings: dict) -> bool:
if not isinstance(h, dict):
continue
cmd = h.get("command", "")
# Match 'telegram-codex-bot hook' or paths ending with 'telegram-codex-bot hook'
if cmd == _HOOK_COMMAND_SUFFIX or cmd.endswith("/" + _HOOK_COMMAND_SUFFIX):
return True
return False
if isinstance(cmd, str) and _is_telegram_codex_bot_hook_command(cmd):
return h
return None


def _is_hook_installed(settings: dict) -> bool:
"""Check if telegram-codex-bot hook is already installed in hooks.json."""
return _find_installed_hook(settings) is not None


def _hook_command_has_missing_absolute_executable(command: str) -> bool:
"""Return True when an installed absolute hook path no longer exists."""
try:
parts = shlex.split(command)
except ValueError:
return False
if len(parts) < 2 or parts[-1] != "hook":
return False
executable = Path(parts[-2]).expanduser()
return executable.is_absolute() and not executable.exists()


def _read_json_file(path: Path) -> dict[str, Any]:
Expand Down Expand Up @@ -219,17 +246,45 @@ def _install_hook() -> int:
print(message, file=sys.stderr)
return 1

# Check if already installed
if _is_hook_installed(settings):
# Find the full path to telegram-codex-bot
cli_path = _find_cli_path()
hook_command = f"{cli_path} hook"

# Check if already installed. Older installs may point at a deleted venv or
# checkout; repair that in place so hook --install remains self-healing.
installed_hook = _find_installed_hook(settings)
if installed_hook is not None:
installed_command = str(installed_hook.get("command") or "")
if _hook_command_has_missing_absolute_executable(installed_command):
installed_hook["command"] = hook_command
installed_hook.setdefault("type", "command")
installed_hook.setdefault("statusMessage", _HOOK_STATUS_MESSAGE)
installed_hook.setdefault("timeout", _HOOK_TIMEOUT_SECONDS)
try:
hooks_file.parent.mkdir(parents=True, exist_ok=True)
_write_json_file(hooks_file, settings)
except OSError as e:
logger.error("Error writing %s: %s", hooks_file, e)
print(f"Error writing {hooks_file}: {e}", file=sys.stderr)
return 1
logger.info(
"Repaired stale hook command in %s: %s -> %s",
hooks_file,
installed_command,
hook_command,
)
print(
"Hook command repaired in "
f"{hooks_file} (Codex hooks enabled in {config_file})"
)
return 0

logger.info("Hook already installed in %s", hooks_file)
print(
f"Hook already installed in {hooks_file} (Codex hooks enabled in {config_file})"
)
return 0

# Find the full path to telegram-codex-bot
cli_path = _find_cli_path()
hook_command = f"{cli_path} hook"
hook_config = {
"type": "command",
"command": hook_command,
Expand Down
44 changes: 44 additions & 0 deletions tests/telegram_codex_bot/test_hook.py
Original file line number Diff line number Diff line change
Expand Up @@ -340,6 +340,50 @@ def test_install_is_idempotent_and_enables_feature(
assert len(hooks_payload["hooks"]["SessionStart"]) == 1
assert "Hook already installed" in capsys.readouterr().out

def test_install_repairs_stale_absolute_hook_path(
self,
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
) -> None:
config_file, hooks_file = self._patch_codex_paths(monkeypatch, tmp_path)
monkeypatch.setattr(
hook_module.shutil, "which", lambda _: "/opt/bin/telegram-codex-bot"
)
stale_cli = tmp_path / "deleted" / ".venv" / "bin" / "telegram-codex-bot"
hooks_file.parent.mkdir(parents=True, exist_ok=True)
hooks_file.write_text(
json.dumps(
{
"hooks": {
"SessionStart": [
{
"matcher": "startup|resume",
"hooks": [
{
"type": "command",
"command": f"{stale_cli} hook",
"statusMessage": "Registering Codex session",
"timeout": 5,
}
],
}
]
}
}
),
encoding="utf-8",
)

assert _install_hook() == 0

assert config_file.read_text(encoding="utf-8") == "[features]\nhooks = true\n"
hooks_payload = json.loads(hooks_file.read_text(encoding="utf-8"))
installed_hooks = hooks_payload["hooks"]["SessionStart"][0]["hooks"]
assert len(installed_hooks) == 1
assert installed_hooks[0]["command"] == "/opt/bin/telegram-codex-bot hook"
assert "Hook command repaired" in capsys.readouterr().out

def test_install_preserves_existing_hooks(
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
Expand Down
Loading