72 lines
3.2 KiB
Python
72 lines
3.2 KiB
Python
"""Движок тика claude_code: выбор движка/модели по уровню, страж cc_guard по кругам. Без LLM."""
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
sys.path.insert(0, ROOT)
|
|
ok = bad = 0
|
|
|
|
|
|
def eq(c, msg):
|
|
global ok, bad
|
|
if c:
|
|
ok += 1
|
|
else:
|
|
bad += 1
|
|
print("FAIL", msg)
|
|
|
|
|
|
def with_env(env, fn):
|
|
keep = {k: os.environ.get(k) for k in env}
|
|
os.environ.update(env)
|
|
try:
|
|
return fn()
|
|
finally:
|
|
for k, v in keep.items():
|
|
if v is None:
|
|
os.environ.pop(k, None)
|
|
else:
|
|
os.environ[k] = v
|
|
|
|
|
|
from harness import cc_backend as cb # noqa: E402
|
|
|
|
for k in [k for k in os.environ if k.startswith(("SIMBIOT_LLM_BACKEND", "SIMBIOT_CC_MODEL"))]:
|
|
os.environ.pop(k)
|
|
os.environ["SIMBIOT_HOME"] = tempfile.mkdtemp() # пустой .env инстанса
|
|
eq(cb.backend_for("min") == "strands", "default strands")
|
|
eq(with_env({"SIMBIOT_LLM_BACKEND": "claude_code"}, lambda: cb.backend_for("hour")) == "claude_code", "global switch")
|
|
eq(with_env({"SIMBIOT_LLM_BACKEND": "claude_code", "SIMBIOT_LLM_BACKEND_DAY": "strands"},
|
|
lambda: cb.backend_for("day")) == "strands", "per-level override wins")
|
|
eq(cb.model_for("min") == "claude-haiku-5-5" and cb.model_for("hour") == "claude-sonnet-5-5"
|
|
and cb.model_for("day") == "claude-opus-5-5", "default models per level")
|
|
eq(with_env({"SIMBIOT_CC_MODEL_MIN": "claude-sonnet-5-5"}, lambda: cb.model_for("min")) == "claude-sonnet-5-5", "model override")
|
|
d = cb._usage_delta({"input_tokens": 2, "cache_creation_input_tokens": 10, "cache_read_input_tokens": 5, "output_tokens": 7})
|
|
eq(d == {"prompt_tokens": 17, "completion_tokens": 7, "model_calls": 1}, "usage delta")
|
|
|
|
|
|
def guard(circle, tool, inp, state):
|
|
ev = json.dumps({"hook_event_name": "PreToolUse", "tool_name": tool, "tool_input": inp})
|
|
r = subprocess.run([sys.executable, os.path.join(ROOT, "harness", "cc_guard.py")], input=ev, text=True,
|
|
capture_output=True, env=dict(os.environ, SIMBIOT_TICK_CIRCLE=str(circle), SIMBIOT_CC_STATE=state))
|
|
return r.returncode
|
|
|
|
|
|
st = tempfile.mktemp()
|
|
eq(guard(1, "Bash", {"command": "ls /"}, st) == 0, "owner: full shell")
|
|
eq(guard(4, "Bash", {"command": "rm -rf /tmp/x"}, st) == 2, "others: arbitrary shell blocked")
|
|
eq(guard(4, "Bash", {"command": "./venv/bin/python cli.py tg-reply m-1 -c hi"}, st) == 0, "others: tg-reply allowed")
|
|
eq(guard(3, "Write", {"file_path": "/root/x"}, st) == 2, "own: write blocked")
|
|
eq(guard(2, "Read", {"file_path": os.path.join(ROOT, ".env")}, st) == 2, "board: secrets blocked")
|
|
eq(guard(2, "Read", {"file_path": "docs/ARCHITECTURE.md"}, st) == 0, "board: docs readable")
|
|
eq(guard(4, "mcp__portal__list_members", {}, st) == 2, "others: portal registry blocked")
|
|
eq(guard(3, "ToolSearch", {"query": "x"}, st) == 0, "ToolSearch is internal")
|
|
eq(guard(3, "WebFetch", {"url": "x"}, st) == 2, "own: web blocked")
|
|
st2 = tempfile.mktemp()
|
|
codes = [guard(1, "mcp__portal__whoami", {}, st2) for _ in range(5)]
|
|
eq(codes == [0, 0, 0, 0, 2], f"portal MCP limit 4/tick: {codes}")
|
|
print(f"{ok} ok, {bad} fail")
|
|
sys.exit(1 if bad else 0)
|