113 lines
4.5 KiB
Python
113 lines
4.5 KiB
Python
"""Simbiot Harness — хук Claude Code для LLM-тика на движке claude_code (cc_backend.py).
|
||
|
||
Claude Code зовёт его на каждом PreToolUse/PostToolUse (stdin — JSON события).
|
||
PreToolUse: инструмент Claude → инструмент харнеса → trust.check_tool по кругу тика
|
||
(SIMBIOT_TICK_CIRCLE); запрет — exit 2 + причина в stderr (Claude видит её
|
||
как отказ инструмента). MCP портала — не больше MAX_CALLS_PER_TICK за тик.
|
||
PostToolUse: ToolResult в трассу.
|
||
Трасса — те же события, что у Strands (_wire_trace_hooks): ToolCalled/ToolResult в
|
||
агрегате trace "<trace_id>:<n>", n — общий счётчик тика (файл SIMBIOT_CC_STATE + flock).
|
||
Хук не должен валить тик: любая собственная ошибка → разрешить (кроме явного запрета).
|
||
"""
|
||
import fcntl
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, os.path.dirname(_HERE))
|
||
|
||
# инструмент Claude Code → (имя для trust.check_tool, ключ пути/команды)
|
||
_READ = {"Read", "Glob", "Grep", "LS", "NotebookRead"}
|
||
_WRITE = {"Write", "Edit", "MultiEdit", "NotebookEdit"}
|
||
_FREE = {"TodoWrite", "TodoRead", "ExitPlanMode", "ToolSearch"} # без внешних эффектов (ToolSearch — подгрузка схем MCP)
|
||
|
||
|
||
def _map(tool: str, inp: dict):
|
||
if tool == "Bash":
|
||
return "shell", {"command": inp.get("command") or ""}
|
||
if tool in _READ:
|
||
return "read", {"path": inp.get("file_path") or inp.get("path") or inp.get("notebook_path") or ""}
|
||
if tool in _WRITE:
|
||
return "write", {"path": inp.get("file_path") or inp.get("notebook_path") or ""}
|
||
if tool.startswith("mcp__portal__"):
|
||
return "portal_mcp", {"tool": tool[len("mcp__portal__"):]}
|
||
if tool in _FREE:
|
||
return "todo_write", {}
|
||
return tool.lower(), inp # WebFetch/WebSearch/Task… — только круг с полным shell
|
||
|
||
|
||
def _state(update):
|
||
"""Атомарно прочитать/изменить состояние тика (счётчики)."""
|
||
path = os.environ.get("SIMBIOT_CC_STATE", "")
|
||
if not path:
|
||
return update({})
|
||
with open(path, "a+") as f:
|
||
fcntl.flock(f, fcntl.LOCK_EX)
|
||
f.seek(0)
|
||
try:
|
||
st = json.loads(f.read() or "{}")
|
||
except ValueError:
|
||
st = {}
|
||
out = update(st)
|
||
f.seek(0)
|
||
f.truncate()
|
||
f.write(json.dumps(st))
|
||
return out
|
||
|
||
|
||
def _trace(kind: str, payload: dict):
|
||
trace_id = os.environ.get("SIMBIOT_TRACE_ID", "")
|
||
if not trace_id:
|
||
return
|
||
try:
|
||
from harness import es
|
||
|
||
def bump(st):
|
||
st["n"] = st.get("n", 0) + 1
|
||
return st["n"]
|
||
n = _state(bump)
|
||
es.append_event("trace", f"{trace_id}:cc{n}", kind, {
|
||
"trace_id": trace_id, "level": os.environ.get("SIMBIOT_TICK_LEVEL", ""),
|
||
"engine": "claude_code", **payload})
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def main():
|
||
try:
|
||
ev = json.loads(sys.stdin.read() or "{}")
|
||
except ValueError:
|
||
return 0
|
||
tool = ev.get("tool_name") or "?"
|
||
inp = ev.get("tool_input") or {}
|
||
if ev.get("hook_event_name") == "PostToolUse":
|
||
res = ev.get("tool_response")
|
||
_trace("ToolResult", {"tool": tool, "ok": True,
|
||
"result": (res if isinstance(res, str) else json.dumps(res, ensure_ascii=False, default=str))[:4000]})
|
||
return 0
|
||
name, hinp = _map(tool, inp)
|
||
ok, why = True, ""
|
||
try:
|
||
from harness import trust
|
||
ok, why = trust.check_tool(name, hinp, os.environ.get("SIMBIOT_TICK_CIRCLE"))
|
||
if ok and name == "portal_mcp":
|
||
from harness.portal_mcp import MAX_CALLS_PER_TICK
|
||
|
||
def take(st):
|
||
st["mcp"] = st.get("mcp", 0) + 1
|
||
return st["mcp"]
|
||
if _state(take) > MAX_CALLS_PER_TICK:
|
||
ok, why = False, f"лимит: {MAX_CALLS_PER_TICK} вызовов портала за тик"
|
||
except Exception:
|
||
ok, why = True, ""
|
||
_trace("ToolCalled", {"tool": tool, "input": json.dumps(inp, ensure_ascii=False, default=str)[:4000],
|
||
**({} if ok else {"blocked": why})})
|
||
if not ok:
|
||
print(f"Запрещено стражем харнеса: {why}", file=sys.stderr)
|
||
return 2
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|