Files
ISEP/hooks/lib/dispatch_parse.py
T
Leo 5ac06abc95 工人有名字+回覆也是派工+未經調查不寫診斷(inkstone/ISEP#86/#87/#88)
三張票同一族(誰在派、派給誰、派的內容住哪裡),做在同一條分支:

#86 工人名單:agents/ 七位有名字的工人+scripts/roster+hooks/roster-guard.sh
    派工用 Task 的 subagent_type 指名,派工單格式一個字都沒改;
    指對名字就把那位的檔案原文注入(你是誰/先讀什麼/你的紅線)。
    【身份】欄同時吃得下工人名字(原本三個角色照舊)。

#87 未經調查不寫診斷:hooks/diagnosis-evidence-guard.sh + investigate-first-stamp.sh
    三個結構訊號(派過人查沒/有沒有走得過去的出處/有沒有份量),
    一個關鍵字比對都沒有;轉述有出處不會被誤擋。

#88 回覆也是派工:不另造閘,把攔截點加掛上去。
    hooks/lib/dispatch_parse.py 的 tool_channel() 一次列全所有通往 subagent 的路
    (SendMessage/雲端 session・trigger/claude -p);擋下來時把那段內容原文印出來。
    subagent 往上回報(to: "main")=交件不是派工,刻意不管。

順手修掉一個真的會咬人的 flake:dispatch-format-guard 原本開四支 python 各讀一個欄位,
機器忙的時候某個欄位會靜靜變空字串(實測連跑 10 次有 1 次「豁免了卻還是被擋」)。
四個欄位改成一次讀完。

版本號待總管定(plugin.json 只更新了描述裡的數字,版本沒動)。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016ZBu4Sa1cGntKFRBYNZ6xs
2026-08-28 01:15:09 +00:00

420 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""hooks/lib/dispatch_parse.py — 派工單與交件回覆的**結構解析器**(不是閘)。
依 inkstone/ISEP#40 S7 的慣例:`lib/` 裡的東西是共用零件,本身不決定 allow/block
判決由呼叫它的閘做(`hooks/dispatch-format-guard.sh`、`hooks/reply-identity-guard.sh`、
`scripts/ticket`)。
━━ 規則本身(leo 2026-08-27inkstone/ISEP#30 comment 4322 / 4325 / 4327)━━
派工單 【工單】owner/repo#N。就這樣。
三則合起來是一條規則:
4322 任務要在票上,不是在 prompt 裡 —— 而閘只驗了票號這個殼
4325 「你用一個 **output parser** 把你給 subagent 的指令規範,分作幾點,
每一點規定格式,**照這種散文寫法根本無法迭代**」「**警察也不能抓**」
4327 「**交件方式不需要寫,定義在原則裡**⋯⋯每次都一樣提取出來變成共通規定」
「(那些 session 事實)**為什麼不寫到票裡?**」
⇒ 派工單裡的東西只有兩種,**兩種都不該留在派工單**:
每次都一樣 交件方式、不要 push main、org 是 inkstone、先讀該 repo 的 CLAUDE.md
=> 共通規定(由閘自動注入,見 dispatch-format-guard.sh
這次才知道 main 是哪顆 commit、今天撞過什麼、另一條線正在動什麼
=> **寫進那張票**(「票上還沒有」是「去把它寫上票」的指令,
不是「把它寫進 prompt」的理由)
判準一句話:**「這句話換一張票還成立嗎?」**
還成立 => 共通規定;只有這次成立 => 寫進這張票。兩種都不進派工單。
為什麼連「這次才知道的事」也不能留在 prompt(4327 實害):
那個 agent 被停掉/換人接手,那段事實就隨 prompt 消失。
2026-08-27 總管停掉重派 3 次,前兩次的任務與 session 事實全部蒸發。
━━ 為什麼這支過得了「封動作不封文字」的檢驗(leo 2026-08-17)━━━━━━━━
「你在**文字層**封路⋯⋯**自然語言的變體是無限的,blacklist 永遠追不完。**
封路哲學之所以有效,是因為它封的是**動作**——動作有限且可枚舉,文字不是。」
當日實證:文字層的閘 8 次誤攔、0 次正確攔截,**紅線寫得越細,命中關鍵字的機率越高
⇒ 那些閘在懲罰謹慎**。
本檔**沒有「命中某個詞就違規」的比對**。判準是「這一行是不是【工單】欄位」——
**在不在**,不是**寫什麼**。用到的正則有三種,全部在認形狀:
- `【某某】` 欄位標記本身
- `owner/repo#N`(可選外掛全形/半形括號的 `→ comment M`)票號形狀
- 唯一例外(ISEP#65):`_COURTESY_CLOSERS` 是一份**白名單**,只用來認出
「這一行是不是零資訊的禮貌收尾」,作用方向與黑名單相反——認出的是
「不算違規」,不是「算違規」;且要求整行清乾淨標點後完全相等,不是
「包含」,所以不會反過來放走任何真正的內容。
措辭多謹慎、有沒有提到「紅線」「部署」「main」,本檔一律不看,仍然不會被
「寫得越小心越容易被罰」的病拖下水。
━━ 派工單格式 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
【工單】inkstone/ISEP#30 → comment 4322
就這一行。要帶兩張票就兩行【工單】。**沒有第二個欄位。**
━━ 交件回覆格式(任何人 → 票)━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
【身份】subagentinkstone/ISEPfeat/ticket-carries-the-task
第一行就要認得出是誰寫的。角色是三選一的允許清單:`總管`/`subagent``leo`。
2026-08-27 實害:總管寫的診斷被當成 subagent 的結論,而其中一則是錯的。
━━ 通往 subagent 的路不只一條(inkstone/ISEP#88)━━━━━━━━━━━━━━━━
第一版的閘只掛在 `PreToolUse(Task|Agent)`——那只涵蓋「**新開**一個 subagent」。
**回覆一個正在跑的 subagent 走的是別的工具,根本不經過那個攔截點。**
實況(2026-08-27 本 session):第一次派工乾乾淨淨只有票號,
中途回覆時又把一長串修改要求直接丟過去,**那些話票上一個字都沒有**。
那條線被停掉或換人接手,它們就消失——正是壓縮派工單原本要解決的問題。
⇒ 本檔不再只認一種 payload 形狀。`tool_channel()` 把所有通往 subagent 的路
收成一張表(見該函式),一次列全:
Task / Agent 新開一個 subagent → prompt
SendMessage 回覆正在跑的那個 → message
*Claude_Code_Remote__create_session 開一個新的雲端 session → prompt
*Claude_Code_Remote__send_message 送訊息進某個 session → text / message
*Claude_Code_Remote__create_trigger 排程派工(未來會醒) → prompt
*Claude_Code_Remote__update_trigger 改排程派工的內容 → prompt
*Claude_Code_Remote__fire_trigger 當場點燃排程派工 → text
*Claude_Code_Remote__send_later 排一則訊息給自己 → message
Bash `claude -p <prompt>` 用 CLI 開一個新 session → 指令裡那段 prompt
🔴 這張表就是 `ticket-api-bypass-guard.sh` 檔頭記過的那一課的同款:
「認動作的方式漏了一條路」是這一族的通病(那支第一版只認大寫裸字 `POST`,
於是 `requests.post()` 與 urllib 的隱式 POST 全部漏掉)。
⇒ 新增一條路要加在 `tool_channel()`**不要另造一支平行的閘**。
兩條路的分界只有一個(其餘判準完全共用,這是刻意的):
· `dispatch`TaskAgent)沒有【工單】 ⇒ **skip**,那是 no-ticket-no-dispatch 的地盤
· `reply`(其餘全部)沒有【工單】 ⇒ **違規**,因為那支閘沒有掛在這些路上,
不擋就等於這條路整條裸奔
用法:
import dispatch_parse
r = dispatch_parse.parse_dispatch(prompt_text)
ok, detail = dispatch_parse.parse_identity(comment_body)
CLI:吃 hook 的 JSON,吐一份事實 JSON
printf '%s' "$INPUT" | python3 hooks/lib/dispatch_parse.py dispatch
"""
import json
import re
import shlex
import sys
# ── 派工單的字彙表:**只有一個欄位**。這是允許清單,不是黑名單 ──────────────
DISPATCH_ALLOWED = ("工單",)
# 曾經合法、現已收回的欄位。**不參與判決**(判決是「不是【工單】就是多的」),
# 只用來讓訊息講得出「你這一欄的內容該搬去哪」。
RETIRED_FIELDS = {
"人格": "共通規定(收工方從票的 repo 就知道自己戴誰的人格)",
"交件": "共通規定(每張票都一樣:貼回原票)",
"就地": "**寫進那張票**(這正是「票上還沒有」的解法)",
"身份": "那是**交件回覆**的欄位,不是派工單的",
}
IDENTITY_ROLES = ("總管", "subagent", "leo")
_MARKER_RE = re.compile(r"^【([^】]{1,12})】\s*(.*)$")
_REF_RE = re.compile(r"^([A-Za-z0-9_.-]+)/([A-Za-z0-9_.-]+)#(\d+)\s*$")
# 🔴 inkstone/ISEP#65:頂層 CLAUDE.md 規定的合格格式帶著全形括號——
# 「【工單】owner/repo#N(→ comment M)」。舊版正則只吃「→ comment M」本體,
# 兩側的括號(全形/半形都要吃)沒被算進去,殘留的括號讓 _REF_RE 比不過,
# 於是**規則自己定義的合格範例會被自己的閘擋下**。這裡把左右括號都收進來,
# 兩種都設成可選(沒有括號的舊寫法、純 #issuecomment- 連結都還要能過)。
_COMMENT_RE = re.compile(r"[(]?\s*(?:#issuecomment-|→\s*comment\s*)(\d+)\s*[)]?")
_FENCE_RE = re.compile(r"^\s*(```|~~~)")
# 🔴 inkstone/ISEP#65:票號後面只掛一句純禮貌收尾(例:「謝謝」),不該被當成
# 「派工單不只有票號」的違規——它不含任何要搬去票上或共通規定的實質內容。
# 這是**白名單,不是黑名單**:判準仍是結構(整行清乾淨標點後「等於」清單裡的
# 固定字串,不是「包含」),漏收某個禮貌講法頂多誤鎖一次(代價小,見 ISEP#65
# 的「優先做成低誤鎖」),但不會反過來放走任何真正的內容——真內容不會剛好
# 打出一個字不差的空話。全檔仍然沒有「命中某個詞就違規」的比對,這裡是唯一
# 一處字面比對,且作用方向相反:認出「不是內容」,不是認出「是違規」。
_COURTESY_CLOSERS = frozenset({
"謝謝", "謝謝你", "謝謝您", "謝謝囉", "多謝", "感謝", "感恩", "辛苦了", "收到",
"thanks", "thank you", "thx",
})
_COURTESY_STRIP_RE = re.compile(r"[\s。!!.,~,、]+$")
def _is_courtesy_closer(line):
s = _COURTESY_STRIP_RE.sub("", line.strip())
return s.lower() in _COURTESY_CLOSERS
def _split_lines(text):
"""切成 (是否在圍欄內, 原始行)。
圍欄(``` / ~~~)裡的東西是**資料**不是欄位標記。同一個道理已經在
hooks/lib/strip_heredoc.py 落地過(heredoc body 是資料不是指令,
inkstone/InkStoneCo#23、#56 的根因就是把資料當指令掃)。
這裡的效果是:貼一段指令進派工單,裡面的【】不會被誤認成欄位——
但整段圍欄本身仍然算「多出來的內容」,照樣會被擋。
"""
out, in_fence = [], False
for line in text.split("\n"):
if _FENCE_RE.match(line):
out.append((in_fence, line)) # 圍欄那一行歸屬於「切換前」的狀態
in_fence = not in_fence
continue
out.append((in_fence, line))
return out
def parse_dispatch(prompt):
"""把派工單拆成「【工單】欄位」與「其餘一切」。**只回事實,不下判決**。
回傳 dict
has_ticket_marker 有沒有出現過【工單】——沒有就不是本格式的東西
refs 解析成功的票號清單 [{'owner','repo','num','comment'}]
bad_ticket_lines 有【工單】標記但形狀不對的那幾行
extra 多出來的實質內容 [(來源欄位或 None, 原始行)]
retired_seen 出現過的、已收回的欄位名(給訊息用)
"""
fields_seen, refs, bad_ticket_lines = [], [], []
extra, retired_seen = [], []
cur = None
for in_fence, raw in _split_lines(prompt or ""):
if not in_fence:
m = _MARKER_RE.match(raw)
if m:
name, rest = m.group(1), m.group(2)
cur = name
fields_seen.append(name)
if name in RETIRED_FIELDS and name not in retired_seen:
retired_seen.append(name)
if name in DISPATCH_ALLOWED:
text = rest.strip()
head = _COMMENT_RE.sub("", text).strip()
rm = _REF_RE.match(head)
if rm:
cm = _COMMENT_RE.search(text)
refs.append({"owner": rm.group(1), "repo": rm.group(2),
"num": int(rm.group(3)),
"comment": int(cm.group(1)) if cm else None})
else:
bad_ticket_lines.append(raw.strip())
else:
if rest.strip():
extra.append((name, raw))
continue
if raw.strip():
# 【工單】欄位本身只准佔它自己那一行;它下面的續行也是多的。
# 白名單豁免只在圍欄外生效——圍欄裡的東西本來就該一律算數(test ⑮),
# 這裡不額外放寬,避免有人把真內容包在圍欄裡再貼一句「謝謝」蒙混。
if not in_fence and _is_courtesy_closer(raw):
continue
extra.append((cur, raw))
return {
"has_ticket_marker": "工單" in fields_seen,
"refs": refs,
"bad_ticket_lines": bad_ticket_lines,
"extra": extra,
"extra_lines": len(extra),
"extra_chars": sum(len(r.strip()) for _, r in extra),
"retired_seen": retired_seen,
"fields_seen": fields_seen,
}
def dispatch_violations(parsed):
"""換算成違規清單。每一條都是**結構**事實,可以單獨迭代。"""
v = []
for line in parsed["bad_ticket_lines"]:
v.append(("票號形狀不對",
"「%s」不是 owner/repo#N(可加 → comment M);org 是 inkstone 不是 Leo"
% line[:60]))
if parsed["extra"]:
first = parsed["extra"][0][1].strip()
v.append(("派工單不只有票號",
"【工單】以外還有 %d 行/%d 字,第一行是:%s"
% (parsed["extra_lines"], parsed["extra_chars"], first[:50])))
return v
def identity_roles():
"""合法的身份角色 = 三個固定角色 + **名單上的工人名字**inkstone/ISEP#86)。
leo:「票上的紀錄要看得出是哪個工人做的。」`subagent` 這個字回答不了「是誰」——
多條線並行時,三個 subagent 的留言長得一模一樣。
名單讀不到就退回三個固定角色(fail-open:名單壞掉不該讓人貼不了留言)。
"""
try:
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import roster
return tuple(IDENTITY_ROLES) + tuple(roster.names())
except Exception:
return tuple(IDENTITY_ROLES)
def parse_identity(body):
"""交件回覆的【身份】欄。回 (ok, detail)。"""
for raw in (body or "").split("\n"):
if not raw.strip():
continue
m = _MARKER_RE.match(raw)
if not m or m.group(1) != "身份":
return False, "第一行不是【身份】"
val = m.group(2).strip()
if not val:
return False, "【身份】後面是空的"
role = re.split(r"[/]", val)[0].strip()
allowed = identity_roles()
if role not in allowed:
return False, "角色「%s」不在 %s 之內" % (role or "(空)", "".join(allowed))
return True, val
return False, "內文是空的"
# ── 通往 subagent 的路:一張表,一次列全(inkstone/ISEP#88)────────────────
# 新增一條路加在這裡。**不要另造一支平行的閘**——票上寫死了:
# 「要嘛把攔截點加掛上去,要嘛讓兩條路共用同一個判斷函式。」
#
# 值是「這個工具的哪幾個欄位裝著要送給 subagent 的話」,依序取第一個有內容的。
_DISPATCH_TOOLS = {
"Task": ("prompt",),
"Agent": ("prompt",),
}
_REPLY_TOOLS = {
"SendMessage": ("message",),
"create_session": ("prompt",),
"send_message": ("text", "message"),
"create_trigger": ("prompt",),
"update_trigger": ("prompt",),
"fire_trigger": ("text",),
"send_later": ("message",),
}
# `claude -p <prompt>` 用 CLI 開一個新 session。判準是**指令名**(結構),
# 不是指令內容裡有沒有某個詞——所以一段剛好含有「【工單】」字樣的 echo 不會被誤認。
_CLAUDE_CLI_RE = re.compile(r"(?:^|[;&|]\s*|\s)claude\s")
_CLI_PRINT_FLAGS = ("-p", "--print")
# 這些旗標後面跟的是它自己的值,不是 prompt
_CLI_VALUE_FLAGS = {"--model", "-m", "--append-system-prompt", "--system-prompt",
"--allowedTools", "--permission-mode", "--output-format",
"--input-format", "--session-id", "--resume", "--add-dir",
"--mcp-config", "--settings", "--agents"}
def _claude_cli_prompt(command):
"""從一段 shell 指令裡把 `claude -p <prompt>` 的 prompt 取出來。取不到回 ""。"""
if not command or not _CLAUDE_CLI_RE.search(command):
return ""
try:
toks = shlex.split(command)
except Exception:
return ""
for i, t in enumerate(toks):
if t.rsplit("/", 1)[-1] != "claude":
continue
rest = toks[i + 1:]
if not any(f in rest for f in _CLI_PRINT_FLAGS):
return "" # 沒有 -p ⇒ 不是一次性的 prompt 執行
skip = False
for j, tok in enumerate(rest):
if skip:
skip = False
continue
if tok in _CLI_VALUE_FLAGS:
skip = True
continue
if tok.startswith("-"):
continue
return tok # 第一個位置參數 prompt
return ""
return ""
def tool_channel(payload):
"""回 (channel, text)。channel ∈ {"dispatch", "reply", ""}。
"" = 這個工具呼叫不是在驅動任何 subagent,本閘管不到。
"""
name = payload.get("tool_name") or ""
ti = payload.get("tool_input") or {}
if not isinstance(ti, dict):
return "", ""
short = name.rsplit("__", 1)[-1]
if name == "Bash" or short == "Bash":
return ("reply", _claude_cli_prompt(ti.get("command") or ""))
# 🔴 方向很重要:本規則管的是「**派工的人**送出去的話」。
# subagent 往上回報(`to: "main"`)是**交件**,不是派工——
# 擋它等於擋掉交件本身。交件的規矩由 baton-handback-guard
# reply-identity-guard 管,不在這裡。
if (name == "SendMessage" or short == "SendMessage") and \
str(ti.get("to") or "").strip().lower() == "main":
return "", ""
for table, channel in ((_DISPATCH_TOOLS, "dispatch"), (_REPLY_TOOLS, "reply")):
keys = table.get(name) or table.get(short)
if not keys:
continue
for k in keys:
v = ti.get(k)
if isinstance(v, str) and v.strip():
return channel, v
return channel, ""
return "", ""
# ── CLI ──────────────────────────────────────────────────────────────────
def _main():
mode = sys.argv[1] if len(sys.argv) > 1 else "dispatch"
try:
payload = json.load(sys.stdin)
except Exception:
print(json.dumps({"status": "skip", "why": "payload 不是 JSON"}))
return
if mode != "dispatch":
print(json.dumps({"status": "skip", "why": "未知模式 %s" % mode}))
return
channel, text = tool_channel(payload)
if not channel:
print(json.dumps({"status": "skip", "why": "不是通往 subagent 的路"}))
return
if channel == "reply" and not (text or "").strip():
print(json.dumps({"status": "skip", "why": "這條路這次沒有帶任何話"}))
return
parsed = parse_dispatch(text)
if not parsed["has_ticket_marker"]:
if channel == "dispatch":
# 連【工單】都沒有 => 那是 no-ticket-no-dispatch.sh 的地盤,本閘閉嘴。
# 兩支閘同時對同一件事開口,收工方會拿到兩份互相打架的教學。
print(json.dumps({"status": "skip",
"why": "沒有【工單】,交給 no-ticket-no-dispatch"}))
return
# reply 這條路上**沒有**那支閘(它只掛 Task|Agent)⇒ 不擋就整條裸奔。
violations = [{"code": "回覆沒有票號",
"detail": "回覆也是派工:內容寫進票,訊息只給【工單】owner/repo#N"}]
extra_text = (text or "").strip()
else:
violations = [{"code": c, "detail": d} for c, d in dispatch_violations(parsed)]
extra_text = "\n".join(r.rstrip() for _, r in parsed["extra"]).strip()
print(json.dumps({
"status": "ok",
"channel": channel,
"violations": violations,
"refs": parsed["refs"],
"retired_seen": parsed["retired_seen"],
"extra_lines": parsed["extra_lines"],
"extra_text": extra_text,
"session_id": payload.get("session_id") or "",
}, ensure_ascii=False))
if __name__ == "__main__":
_main()