dda4852938
在此之前 36 支閘裡只有 2 支會記錄自己擋了什麼,六種逃生口 (NOT_MY_BRANCH_OK=1/no-ticket-needed/wiki-secret-ok/credential-ok/ solo-ok/kbdb-down)用了完全不留痕 ⇒ 擋對 100 次跟擋錯 100 次在資料上 一模一樣 ⇒ 規則只會累積不會被淘汰。 機制:hooks/lib/gate-log-wrap.sh 包在每一支閘外面(hooks.json 的 82 條 gate command 全部前綴這支 wrapper;2 條 scripts/ refresher 不是閘、不包)。 它從不 exit 2,只把真正那支閘的離開碼原樣傳出去 ⇒ 記錄壞掉最多漏記一筆, 不可能誤攔(票紅線:寧可漏記,不可擋錯)。每次留一筆中繼資料: 哪支閘/何時/對什麼工具/判成什麼/(若繞過)用了哪個逃生口。 - 只記中繼資料,完全不寫入指令原文或 payload ⇒ 金鑰在結構上進不了帳本(D36) - 純 POSIX shell 記錄,不另起 python ⇒ 不讓閘變慢變脆 - 紀錄長在既有閘動作上,零輪詢/零 cron(D20) - 帳本沿用 milestone-account 慣例(不寫進會被 plugin update 換掉的目錄) 報告:scripts/gate-report all 回答票上三題(誤擋嫌疑排序/最近沒擋過的規則/ 同一 session 反覆撞同一支閘)。基線與反面驗證見票留言。 測試:hooks/tests/gate-log.test.sh 22 條(透明/留痕/反面驗證/不變脆/不外洩) + scripts/test-gate-report.sh 8 條(三題數字對不對)。 hooks 檔數 61、hooks.json 註冊 84 均不變(wrapper 住 hooks/lib,非 gate .sh)。 版本:待總管定版(改了會被載入的東西需升版才傳得到,但 tag 由總管打)。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TNfNypeDaCVEZQW1Wqwfxq
207 lines
7.9 KiB
Python
Executable File
207 lines
7.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""gate-report — 把 gate-log.jsonl 變成「規則有沒有在運作」的答案(inkstone/InkStoneCo#48)。
|
||
|
||
這支不是文件、不是感覺,是**查資料**。紀錄由 hooks/lib/gate-log-wrap.sh 在每一次
|
||
閘動作當下留下(擋/放行/被逃生口繞過各一種 verdict)。本工具回答票上那三題:
|
||
|
||
q1 哪支閘擋最多?其中有多少最後被繞過? (=誤擋嫌疑最高的)
|
||
q2 哪條規則最近 N 天一次都沒擋過? (=可能已經沒有存在意義)
|
||
q3 同一個 session/agent 反覆撞同一支閘幾次? (=規則沒被理解,或閘寫錯)
|
||
|
||
用法:
|
||
scripts/gate-report [--log PATH] [--days N] [all|q1|q2|q3|summary|ledger-path]
|
||
|
||
--log 紀錄檔路徑(預設沿用 gate-log-wrap.sh 的解析順序)
|
||
--days N q2 的視窗,最近 N 天(預設 7)
|
||
(不給子命令 = all:先印基線數字,再依序印三題)
|
||
|
||
離線、唯讀:只讀那個 jsonl,不打任何網路,不寫任何東西。
|
||
"""
|
||
import argparse
|
||
import collections
|
||
import datetime as dt
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
ROOT = os.path.dirname(HERE) # scripts/ 的上一層 = plugin 根
|
||
|
||
|
||
def resolve_log():
|
||
"""跟 gate-log-wrap.sh 同一套順序找紀錄檔。"""
|
||
v = os.environ.get("ISEP_GATE_LOG")
|
||
if v:
|
||
return v
|
||
for base in (os.environ.get("CLAUDE_PROJECT_DIR") or "", ROOT):
|
||
if base and os.path.isdir(os.path.join(base, "system-dev")):
|
||
return os.path.join(base, "system-dev", "gate-log", "gate-log.jsonl")
|
||
return os.path.join(os.path.expanduser("~"), ".claude", "isep-gate-log", "gate-log.jsonl")
|
||
|
||
|
||
def registered_gates():
|
||
"""從 hooks.json 讀出「目前真的被記錄層包著的閘」清單——q2 要拿它跟紀錄比對,
|
||
才分得出『掛著但從沒擋過』與『根本沒掛』。
|
||
|
||
判準是「這條 command 有沒有被 gate-log-wrap.sh 包住」:包住的才是我們在計數的閘,
|
||
scripts/ 底下的 refresher(不是閘、刻意沒包)就不會被誤算進來。
|
||
讀不到 hooks.json 就回空集合(q2 會退回只用紀錄裡看過的閘,不會因此爆掉)。"""
|
||
path = os.path.join(ROOT, "hooks", "hooks.json")
|
||
names = set()
|
||
try:
|
||
d = json.load(open(path))
|
||
for ev in d.get("hooks", {}).values():
|
||
for group in ev:
|
||
for h in group.get("hooks", []):
|
||
cmd = h.get("command", "")
|
||
if "gate-log-wrap" not in cmd:
|
||
continue # 沒被記錄層包住 = 不是我們在計數的閘
|
||
last = cmd.split()[-1] if cmd.split() else "" # 閘路徑一律是最後一段
|
||
base = last.rsplit("/", 1)[-1]
|
||
if base.endswith(".sh"):
|
||
names.add(base[:-3])
|
||
except Exception:
|
||
pass
|
||
return names
|
||
|
||
|
||
def load(path):
|
||
rows = []
|
||
try:
|
||
with open(path) as f:
|
||
for ln in f:
|
||
ln = ln.strip()
|
||
if not ln:
|
||
continue
|
||
try:
|
||
rows.append(json.loads(ln))
|
||
except Exception:
|
||
continue # 壞行跳過,不讓一行毀掉整份報告
|
||
except FileNotFoundError:
|
||
pass
|
||
return rows
|
||
|
||
|
||
def parse_ts(s):
|
||
try:
|
||
return dt.datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=dt.timezone.utc)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def summary(rows):
|
||
v = collections.Counter(r.get("verdict") for r in rows)
|
||
gates = {r.get("gate") for r in rows if r.get("gate")}
|
||
print("── 基線數字(拿今天當基線)───────────────────────────")
|
||
print(" 紀錄總筆數 : %d" % len(rows))
|
||
print(" 出現過的閘 : %d 支" % len(gates))
|
||
print(" 擋下 (block) : %d" % v.get("block", 0))
|
||
print(" 被繞過 (bypass) : %d" % v.get("bypass", 0))
|
||
print(" 放行 (pass) : %d" % v.get("pass", 0))
|
||
print(" 其他/錯 (error) : %d" % v.get("error", 0))
|
||
if not rows:
|
||
print(" (帳本還是空的——這一層剛裝上,還沒有任何閘動作被記錄。)")
|
||
|
||
|
||
def q1(rows):
|
||
print("── Q1 哪支閘擋最多?其中多少被繞過?(誤擋嫌疑排序)──────")
|
||
block = collections.Counter()
|
||
byp = collections.Counter()
|
||
for r in rows:
|
||
g = r.get("gate")
|
||
if not g:
|
||
continue
|
||
if r.get("verdict") == "block":
|
||
block[g] += 1
|
||
elif r.get("verdict") == "bypass":
|
||
byp[g] += 1
|
||
gates = set(block) | set(byp)
|
||
if not gates:
|
||
print(" (還沒有任何 block/bypass 紀錄。)")
|
||
return
|
||
# 「擋」= block + bypass(兩者都是「這支閘本來要擋」的動作);被繞過率高 = 誤擋嫌疑高
|
||
print(" %-30s %6s %7s %7s %s" % ("閘", "擋下", "被繞過", "本要擋", "繞過率"))
|
||
for g in sorted(gates, key=lambda x: -(block[x] + byp[x])):
|
||
would = block[g] + byp[g]
|
||
rate = (byp[g] / would * 100) if would else 0.0
|
||
print(" %-30s %6d %7d %7d %5.1f%%" % (g, block[g], byp[g], would, rate))
|
||
|
||
|
||
def q2(rows, days, reg):
|
||
print("── Q2 最近 %d 天一次都沒擋過的規則(可能已無存在意義)──" % days)
|
||
now = dt.datetime.now(dt.timezone.utc)
|
||
cutoff = now - dt.timedelta(days=days)
|
||
blocked_recent = set()
|
||
seen = set()
|
||
for r in rows:
|
||
g = r.get("gate")
|
||
if not g:
|
||
continue
|
||
seen.add(g)
|
||
ts = parse_ts(r.get("ts", ""))
|
||
if r.get("verdict") in ("block", "bypass") and ts and ts >= cutoff:
|
||
blocked_recent.add(g)
|
||
universe = (reg | seen) if reg else seen
|
||
idle = sorted(g for g in universe if g not in blocked_recent)
|
||
if not universe:
|
||
print(" (沒有可比對的閘清單。)")
|
||
return
|
||
for g in idle:
|
||
tag = "(掛著,但這視窗內沒有任何紀錄)" if g not in seen else "(有動作,但都是放行,沒擋過)"
|
||
print(" · %-30s %s" % (g, tag))
|
||
if not idle:
|
||
print(" (這視窗內每一支掛著的閘都至少擋過一次。)")
|
||
|
||
|
||
def q3(rows):
|
||
print("── Q3 同一個 session/agent 反覆撞同一支閘(≥2 次)─────")
|
||
hit = collections.Counter()
|
||
child = {}
|
||
for r in rows:
|
||
if r.get("verdict") not in ("block", "bypass"):
|
||
continue
|
||
g, sid = r.get("gate"), r.get("sid") or "(no-sid)"
|
||
if not g:
|
||
continue
|
||
hit[(sid, g)] += 1
|
||
child[(sid, g)] = r.get("child", 0)
|
||
repeat = [(k, n) for k, n in hit.items() if n >= 2]
|
||
if not repeat:
|
||
print(" (還沒有『同一 session 撞同一支閘 2 次以上』的紀錄。)")
|
||
return
|
||
print(" %6s %-4s %-30s %s" % ("撞幾次", "子?", "閘", "session"))
|
||
for (sid, g), n in sorted(repeat, key=lambda x: -x[1]):
|
||
print(" %6d %-4s %-30s %s" % (n, "子" if child[(sid, g)] else "主", g, sid))
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(add_help=True)
|
||
ap.add_argument("cmd", nargs="?", default="all",
|
||
choices=["all", "q1", "q2", "q3", "summary", "ledger-path"])
|
||
ap.add_argument("--log", default=None)
|
||
ap.add_argument("--days", type=int, default=7)
|
||
a = ap.parse_args()
|
||
|
||
path = a.log or resolve_log()
|
||
if a.cmd == "ledger-path":
|
||
print(path)
|
||
return
|
||
|
||
rows = load(path)
|
||
reg = registered_gates()
|
||
if a.cmd in ("all", "summary"):
|
||
summary(rows)
|
||
if a.cmd == "summary":
|
||
return
|
||
print()
|
||
if a.cmd in ("all", "q1"):
|
||
q1(rows); print() if a.cmd == "all" else None
|
||
if a.cmd in ("all", "q2"):
|
||
q2(rows, a.days, reg); print() if a.cmd == "all" else None
|
||
if a.cmd in ("all", "q3"):
|
||
q3(rows)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|