integrations/agent-plugins/hermes/__init__.py
1
"""koment plugin for Hermes Agent.
3
Wires two hooks onto the koment CLI:
5
* ``pre_tool_call`` — before Hermes writes or patches a file, the content is
6
handed to ``koment agents hook pre-tool``. When the edit adds ordinary
7
explanatory comment intent the write is denied and the reason tells the agent
8
to record the rationale as an annotation instead.
10
* ``pre_verify`` — before a turn that edited code is allowed to finish,
11
``koment agents hook stop`` runs the three repository gates. A failure keeps
12
the agent working rather than stopping with the repository in a state its own
13
policy rejects.
15
Both hooks shell out to the ``koment`` binary, which must be on ``PATH``. The
16
decision logic lives in Go so that this plugin, the Claude Code hooks, the
17
OpenCode plugin and CI cannot disagree about what the policy is.
19
Set ``KOMENT_PLUGIN_DISABLED=1`` to make both hooks no-ops without uninstalling.
20
"""
22
from __future__ import annotations
24
import json
25
import logging
26
import os
27
import shutil
28
import subprocess
29
from typing import Any, Dict, Optional
31
logger = logging.getLogger(__name__)
33
KOMENT_BINARY = "koment"
34
PRE_TOOL_TIMEOUT_SECONDS = 10
35
VERIFY_TIMEOUT_SECONDS = 120
37
_WRITE_TOOLS: Dict[str, tuple] = {
38
"write_file": ("path", ("content",)),
39
"patch": ("path", ("new_string", "patch")),
40
}
43
def _disabled() -> bool:
44
return os.environ.get("KOMENT_PLUGIN_DISABLED", "").strip() not in ("", "0", "false")
47
def _koment_available() -> bool:
48
return shutil.which(KOMENT_BINARY) is not None
51
def _run(arguments, payload: str, timeout: int) -> Optional[subprocess.CompletedProcess]:
52
try:
53
return subprocess.run(
54
[KOMENT_BINARY, *arguments],
55
input=payload,
56
capture_output=True,
57
text=True,
58
timeout=timeout,
59
check=False,
60
)
61
except (OSError, subprocess.SubprocessError) as error:
62
logger.warning("koment %s did not run: %s", " ".join(arguments), error)
63
return None
66
def _edit_payload(tool_name: str, arguments: Any) -> Optional[str]:
67
mapping = _WRITE_TOOLS.get(tool_name)
68
if mapping is None or not isinstance(arguments, dict):
69
return None
70
path_key, content_keys = mapping
71
path = arguments.get(path_key)
72
if not isinstance(path, str) or not path:
73
return None
74
for key in content_keys:
75
content = arguments.get(key)
76
if isinstance(content, str) and content:
77
return json.dumps(
78
{
79
"tool_name": "opencode_edit",
80
"tool_input": {"filePath": path, "content": content},
81
}
82
)
83
return None
86
def _denial_reason(stdout: str) -> Optional[str]:
87
try:
88
decoded = json.loads(stdout or "{}")
89
except json.JSONDecodeError:
90
logger.warning("koment pre-tool returned output that is not JSON")
91
return None
92
specific = decoded.get("hookSpecificOutput")
93
if not isinstance(specific, dict):
94
return None
95
if specific.get("permissionDecision") != "deny":
96
return None
97
reason = specific.get("permissionDecisionReason")
98
return reason if isinstance(reason, str) and reason else "koment policy denied this edit"
101
def _on_pre_tool_call(tool_name: str = "", args: Any = None, **_: Any) -> Optional[Dict[str, Any]]:
102
if _disabled() or not _koment_available():
103
return None
104
payload = _edit_payload(tool_name, args)
105
if payload is None:
106
return None
107
finished = _run(["agents", "hook", "pre-tool"], payload, PRE_TOOL_TIMEOUT_SECONDS)
108
if finished is None:
109
return None
110
reason = _denial_reason(finished.stdout)
111
if reason is None:
112
return None
113
return {"action": "block", "message": reason}
116
def _verification_failure() -> Optional[str]:
117
finished = _run(["agents", "hook", "stop"], "{}", VERIFY_TIMEOUT_SECONDS)
118
if finished is None:
119
return None
120
try:
121
decoded = json.loads(finished.stdout or "{}")
122
except json.JSONDecodeError:
123
logger.warning("koment stop hook returned output that is not JSON")
124
return None
125
for key in ("reason", "stopReason", "systemMessage"):
126
value = decoded.get(key)
127
if isinstance(value, str) and value:
128
return value
129
return None
132
def _on_pre_verify(**_: Any) -> Optional[Dict[str, Any]]:
133
if _disabled() or not _koment_available():
134
return None
135
reason = _verification_failure()
136
if reason is None:
137
return None
138
return {"decision": "block", "reason": reason}
141
def register(ctx) -> None:
142
ctx.register_hook("pre_tool_call", _on_pre_tool_call)
143
ctx.register_hook("pre_verify", _on_pre_verify)