admin
c1b5e4d966
- 变更规模：新增 514 / 修改 62 / 重命名 155 / 删除 4（归档重组与文档轮次） - .gitignore 修：`归档/**/db-cwd归一-备份-*/` —— 原规则写绝对层级（归档/db-cwd归一-…）， 目录搬进 归档/配置与备份/ 后**静默失效**，43 MB 的 DB 备份又变成未跟踪 - .gitignore 补：嵌套 git 内部数据（归档/内嵌git-20261008/、归档/skills-git-旧线-20261007/dotgit-原样移出/） - .gitignore 补：运行态与部署副本（.workbuddy/collab/、.workbuddy/tools/、.workbuddy/.load-pending、.workbuddy/tmp-*） - .gitignore 补：备份件（*.bak-*） - 未跟踪文件从 2190 降到 890（其余为 归档/ 归档件与 .workbuddy/memory/ 知识文件，按口径入库）
554 lines
25 KiB
Python
554 lines
25 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
派活监管 · 三线执行状态巡检（**只读**）
|
||
==================================================================
|
||
用途：派活之后**不是结束** —— 本脚本把「各会话到底动没动、动了到哪、有没有卡住、
|
||
该不该跟进」变成一次可复跑的读数。⛔ 不起服务、⛔ 不抢锁、⛔ 不改任何文件。
|
||
|
||
用法：
|
||
"E:/ProgramData/.workbuddy/binaries/python/versions/3.13.12/python.exe" \
|
||
"E:/ProgramData/AIProject/ai1net-dsh-server/交付物/派活监管-三线巡检-20260928.py"
|
||
加 `--json` 出机读结果。
|
||
|
||
退出码：0 = 全部已收官 · 1 = 有棒在跑/待跑（正常等） · 2 = **有需跟进的**（未产出/异常/锁未释放）
|
||
|
||
判据源（全部只读）：
|
||
① `workbuddy.db` → `automations`（有没有排上、排的时刻）
|
||
② `workbuddy.db` → `automation_runs`（**跑没跑 ＋ 收官结论原文** ＋ result_success）
|
||
③ `workbuddy.db` → `automation_runtime_state`（此刻**是不是在跑**）
|
||
④ `workbuddy.db` → `sessions`（该工作区的会话活动）
|
||
⑤ 全局执行锁 `dsh-server-docs/05-交接单/.exec-lock`（**三线共用一把** ⇒ 谁占着、放没放）
|
||
⑥ 各线「收件目录」里**在派活时刻之后**新增/改动的文件（＝有没有产出）
|
||
|
||
⚠️ 本机实测两条硬事实：
|
||
- `workbuddy.db` 是 **WAL 三件套**、app 正在运行 ⇒ 一律以 `mode=ro` 打开，⛔ 不写、⛔ 不 cp。
|
||
- 在受限沙箱里 **Node 不能 spawn 系统命令**；Python 可以 ⇒ 本脚本用 Python。
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import time
|
||
from datetime import datetime
|
||
|
||
DB = r"E:\ProgramData\.workbuddy\workbuddy.db"
|
||
LOCKROOT = r"D:\github\dsh_shenxian\dsh-server-docs\05-交接单"
|
||
EXECLOCK = os.path.join(LOCKROOT, ".exec-lock")
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 派活登记（**派一条就加一条**；这是本脚本唯一的"配置"）
|
||
#
|
||
# 🔴 2026-09-28 实测教训（排期）：**三条棒挤在 5 分钟内会互相挡死**。
|
||
# 实测：21:19:12 客户端线抢到**全局独占锁**（未声明 `--domains`）⇒
|
||
# 21:20:02 手机接入线 **抢不到 ⇒ 棒 3 未开工**（守规矩停手，但白跑一趟）。
|
||
# ⇒ 排期两条铁律：① **错开 ≥ 一次跑的时长**（实测单棒约 5 分钟 ⇒ 建议 ≥6 分钟）；
|
||
# ② 并行棒一律让各线 `--claim-exec … --domains <域键>`（域不重叠才真并行）。
|
||
# ⚠️ 另注意：若别人持的是**旧式全局锁**，域锁也照样抢不到（guard 会明确报这点）。
|
||
# ---------------------------------------------------------------------------
|
||
DISPATCHES = [
|
||
{
|
||
"line": "客户端线",
|
||
"ws": "ai1net-dsh-desktop",
|
||
"auto_id": "b5c82852-04af-4213-936d-60958643029e",
|
||
"scheduled_at": "2026-09-28T21:19",
|
||
"task": "垫片改指 WorkBuddy 前置结算棒（只出 D-1 定论 ＋ 执行交接单）",
|
||
"expect": "docs/ 下新增执行交接单（含 D-1 唯一结论）",
|
||
"watch": ["docs"],
|
||
"precondition": None,
|
||
},
|
||
{
|
||
"line": "手机接入线",
|
||
"ws": "ai1net-dsh-anywhere",
|
||
# 21:20 那次（7d50229a）**抢不到全局锁 ⇒ 未开工**（已消耗）；
|
||
# 22:49 重排（da9a0b9e）—— 同一件、加了 --domains 提示。
|
||
"auto_id": "da9a0b9e-5171-47db-b198-1940edb4be2f",
|
||
"scheduled_at": "2026-09-28T22:49",
|
||
"task": "棒 3 开工（Android MVP · 按 sessions 一套实现）· **重排**",
|
||
"expect": "docs/交接单/ 回执 ＋（可能）dsh-client 壳内改动 / APK",
|
||
"watch": ["docs/交接单", "docs"],
|
||
"precondition": "首次（21:20）被全局锁挡下 ⇒ 只要锁空闲即可开工",
|
||
},
|
||
{
|
||
"line": "ai1net_ui",
|
||
"ws": "ai1net_ui",
|
||
"auto_id": "e5faa1ec-6d12-48d2-94dc-1efd829a2c2d",
|
||
"scheduled_at": "2026-09-28T21:24",
|
||
"task": "新接手外来件（ai1net 并入 M6 设备接入 ＋ 规格修订）",
|
||
"expect": "规格修订成稿 ＋ 归并方案（docs/）",
|
||
"watch": ["docs/交接单", "docs"],
|
||
"precondition": "该线此前未参与本功能 ⇒ 件里已备补课段 §1–§4",
|
||
},
|
||
]
|
||
|
||
WSROOT = r"E:\ProgramData\AIProject"
|
||
NOW_MS = int(time.time() * 1000)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 🔴 2026-09-28 实测教训（本脚本最重要的一个判据）
|
||
#
|
||
# 事故：手机接入线棒 3 在 21:20 **抢不到全局锁 ⇒ 未开工**（合规停手），
|
||
# 但它的运行记录 `result_success = 1`（它确实"跑完了"）⇒
|
||
# 旧版判据把它归到「收官·疑未产出」，而旧版**处置规则对这一类只报告、不重排**
|
||
# ⇒ 两条巡检棒（21:45 / 22:18）**连续 2 轮都发现了、都只报告** ⇒
|
||
# **那条活一直躺着死**，直到人手动发现。
|
||
#
|
||
# ⇒ 两条铁律（改这里之前先读）：
|
||
# ① **「跑完了」≠「干了活」** —— 结论原文里自述"未开工/未做事"的，判**异常**，不是"疑未产出"。
|
||
# ② **只报告的监管 = 没监管**（甚至更糟：制造"已经在管了"的假象）
|
||
# ⇒ 本脚本对每条线给出**显式 `action`**；巡检棒照 action 办事，
|
||
# ⛔ 不再由巡检棒自己解释状态分类（那正是这次误判的入口）。
|
||
#
|
||
# 🔴 2026-09-28 第二次教训：**第一版判据有假阳性，已收窄**。
|
||
# 原判据含 `未完成` / `阻塞` / `中止` / `放弃` / `被挡` —— 这些词**同时是报告里的字段名**
|
||
# （例：「## ⑥ 未完成项」「- 未完成：…」「阻塞项」）⇒ 会把**正常收官**的线误判成
|
||
# "未实际执行" ⇒ **凭空重排已完成的活**（＝让它们白跑一遍）。
|
||
# ⇒ 只保留"**明确没干活/干不下去**"的强信号；⛔ 凡可能是**字段名**的词一律不收。
|
||
# ⚠️ 同一份判据在后台监控 `wb-supervisor-watch.py` 里也有一份，**改一处必须改另一处**。
|
||
PENDING_MARKERS = (
|
||
"未开工", "未做事", "没开工",
|
||
"抢不到执行锁", "抢不到锁", "抢不到",
|
||
"无法继续", "未能继续",
|
||
"停手并报告", "停手并回报", "已停手",
|
||
)
|
||
|
||
# action 四类（巡检棒只认这四个，⛔ 不自创）
|
||
ACTION_RESCHEDULE = "reschedule" # 🔴 活没干成 ⇒ **必须当场新建自动化重排**（⛔ 不许只报告）
|
||
ACTION_REPORT_LOCK = "report-lock" # 锁未释放 ⇒ **只报告用户**（R9：⛔ 不代删、不接管）
|
||
ACTION_REPORT = "report" # 需人判读（如"跑完但产出缺失"）⇒ 报告 ＋ 列期望路径
|
||
ACTION_WAIT = "wait" # 正常等待，无事不打扰
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
def out(text):
|
||
"""⛔ 不走 print：文案含 ⛔ 等字符时，裸 print 在非 UTF-8 控制台会 UnicodeEncodeError
|
||
⇒ stdout 空 ⇒ **静默丢结论**（本项目已踩过）。一律写 bytes。"""
|
||
sys.stdout.buffer.write(text.encode("utf-8", "replace"))
|
||
sys.stdout.buffer.write(b"\n")
|
||
|
||
|
||
def ro_conn():
|
||
con = sqlite3.connect("file:%s?mode=ro" % DB, uri=True, timeout=5)
|
||
con.execute("PRAGMA busy_timeout=5000")
|
||
return con
|
||
|
||
|
||
def parse_iso_ms(s):
|
||
try:
|
||
return int(datetime.strptime(s, "%Y-%m-%dT%H:%M").timestamp() * 1000)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def fmt_ms(ms):
|
||
if not ms:
|
||
return "-"
|
||
return datetime.fromtimestamp(ms / 1000).strftime("%H:%M:%S")
|
||
|
||
|
||
def lock_state():
|
||
"""全局执行锁：三线共用一把 ⇒ 监管上这是最强的互斥信号。"""
|
||
if not os.path.isdir(EXECLOCK):
|
||
# 兼容：也有可能是文件式
|
||
if os.path.exists(EXECLOCK):
|
||
try:
|
||
with open(EXECLOCK, "r", encoding="utf-8", errors="replace") as f:
|
||
return {"held": True, "owner": f.read().strip()[:120]}
|
||
except Exception:
|
||
return {"held": True, "owner": "(文件式锁，读不出)"}
|
||
return {"held": False, "owner": None}
|
||
owner_file = os.path.join(EXECLOCK, "OWNER")
|
||
owner = None
|
||
if os.path.isfile(owner_file):
|
||
try:
|
||
with open(owner_file, "r", encoding="utf-8", errors="replace") as f:
|
||
txt = f.read()
|
||
lines = [ln for ln in txt.splitlines() if ln.strip()]
|
||
owner = lines[0][:120] if lines else "(OWNER 存在但为空)"
|
||
except Exception:
|
||
owner = "(OWNER 读不出)"
|
||
else:
|
||
try:
|
||
owner = "(目录式锁，无 OWNER；内容=%s)" % ",".join(os.listdir(EXECLOCK))[:80]
|
||
except Exception:
|
||
owner = "(目录式锁)"
|
||
return {"held": True, "owner": owner or "(未署名)"}
|
||
|
||
|
||
def domain_locks():
|
||
root = os.path.join(LOCKROOT, ".locks")
|
||
if not os.path.isdir(root):
|
||
return []
|
||
got = []
|
||
for name in os.listdir(root):
|
||
if name.startswith("."):
|
||
continue
|
||
got.append(name)
|
||
return got
|
||
|
||
|
||
def new_files(ws, dirs, since_ms, limit=8):
|
||
"""列出收件目录里 mtime > since_ms 的文件（＝派活之后新产出的东西）。"""
|
||
hits = []
|
||
for d in dirs:
|
||
base = os.path.join(WSROOT, ws, d.replace("/", os.sep))
|
||
if not os.path.isdir(base):
|
||
continue
|
||
for dirpath, _dirnames, filenames in os.walk(base):
|
||
for fn in filenames:
|
||
fp = os.path.join(dirpath, fn)
|
||
try:
|
||
st = os.stat(fp)
|
||
except Exception:
|
||
continue
|
||
if st.st_mtime * 1000 > since_ms:
|
||
hits.append({
|
||
"path": os.path.relpath(fp, os.path.join(WSROOT, ws)).replace("\\", "/"),
|
||
"mtime": fmt_ms(st.st_mtime * 1000),
|
||
"size": st.st_size,
|
||
})
|
||
hits.sort(key=lambda x: x["mtime"], reverse=True)
|
||
return hits[:limit]
|
||
|
||
|
||
def session_stats(con, ws, since_ms):
|
||
cwd = "%s/%s" % (WSROOT.replace("\\", "/"), ws)
|
||
rows = con.execute(
|
||
"select id,title,status,created_at,last_activity_at,is_background_automation "
|
||
"from sessions where lower(cwd)=lower(?) and coalesce(created_at,0)>? "
|
||
"order by created_at desc", (cwd, since_ms)).fetchall()
|
||
out_rows = []
|
||
for r in rows:
|
||
out_rows.append({
|
||
"id": (r[0] or "")[:8],
|
||
"title": (r[1] or "")[:44],
|
||
"status": r[2],
|
||
"created": fmt_ms(r[3]),
|
||
"last_activity": fmt_ms(r[4]),
|
||
"bg": bool(r[5]),
|
||
})
|
||
return out_rows
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 读「结果钩子」台账 —— 这条链的另一半
|
||
#
|
||
# 设计（用户 2026-09-28 提出）：hook 把执行结果交到本地台账 ⇒ 后台任务（本脚本）
|
||
# 读台账 ⇒ 结合登记表**决定下一步**。
|
||
#
|
||
# 🔴 这里必须有的一个自检：**钩子本身可能没挂上/没生效（静默失败）**。
|
||
# ⇒ 台账为空**不等于**"没有会话结束过"，只等于"钩子没工作"。
|
||
# ⇒ 所以台账为空时**报出来**（action=report），⛔ 不要当成"一切正常"。
|
||
# ---------------------------------------------------------------------------
|
||
LEDGER = os.path.join(WSROOT, "ai1net-dsh-server", "tmp", "supervise-inbox", "ledger.jsonl")
|
||
HOOK_SCRIPT = os.path.join(WSROOT, "ai1net-dsh-server", ".workbuddy", "tools", "wb-result-hook.py")
|
||
HOOK_LOG = os.path.join(WSROOT, "ai1net-dsh-server", "tmp", "supervise-inbox", "hook.log")
|
||
|
||
# 钩子判定的"没问题"值；其余都值得看一眼
|
||
HOOK_OK = ("ok", "short")
|
||
|
||
|
||
def read_ledger(limit=40):
|
||
"""读台账尾部。⛔ 只读，坏行跳过。"""
|
||
recs = []
|
||
try:
|
||
with open(LEDGER, encoding="utf-8", errors="replace") as f:
|
||
lines = f.read().splitlines()
|
||
except Exception:
|
||
return recs
|
||
for ln in lines[-limit:]:
|
||
ln = ln.strip()
|
||
if not ln:
|
||
continue
|
||
try:
|
||
recs.append(json.loads(ln))
|
||
except Exception:
|
||
continue
|
||
return recs
|
||
|
||
|
||
def hook_health(ledger):
|
||
"""钩子自身是否在工作 —— 这条必须是显式的，否则会静默假绿。"""
|
||
h = {"installed_script": os.path.isfile(HOOK_SCRIPT),
|
||
"ledger_exists": os.path.isfile(LEDGER),
|
||
"records": len(ledger),
|
||
"last_ts": ledger[-1].get("ts") if ledger else None,
|
||
"log_exists": os.path.isfile(HOOK_LOG)}
|
||
h["ok"] = bool(h["installed_script"] and h["records"])
|
||
if not h["installed_script"]:
|
||
h["problem"] = "钩子脚本不在位 ⇒ 台账永远不会写"
|
||
elif not h["ledger_exists"] or not h["records"]:
|
||
h["problem"] = ("台账为空 ⇒ 钩子**从未触发**（可能：还没等到下一个会话结束 / 钩子没挂上 / "
|
||
"Windows 钩子在会话启动时快照 ⇒ 装完后**必须先重启客户端**）")
|
||
else:
|
||
h["problem"] = None
|
||
return h
|
||
|
||
|
||
def ledger_for(ledger, ws, since_ms):
|
||
"""挑出属于某条线的台账记录（按 cwd 尾段匹配）+ 非 ok 的判定。"""
|
||
tail = "/" + ws.lower()
|
||
mine = [r for r in ledger if str(r.get("cwd", "")).lower().endswith(tail)]
|
||
bad = [r for r in mine if r.get("verdict") not in HOOK_OK]
|
||
return mine, bad
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
def check_one(con, d, ledger=None):
|
||
sched_ms = parse_iso_ms(d["scheduled_at"]) or 0
|
||
rec = {"line": d["line"], "ws": d["ws"], "task": d["task"],
|
||
"at": d["scheduled_at"], "expect": d["expect"]}
|
||
|
||
# ① 自动化本体
|
||
# 🔴 幂等待办：`auto_id` 为空 = **这条活还没排上**（不是数据缺失）。
|
||
# 用途（2026-09-28 实测教训）：一轮里若"读到一半被打断"（如工具瞬时不可用），
|
||
# 那把「该排的活」**提前写进登记表**（auto_id 留空）⇒ 中断也只丢最后一步，
|
||
# 任何后来的巡检都会看到它并 `reschedule` ⇒ 那条活**不会静默消失**。
|
||
if not d.get("auto_id"):
|
||
rec.update(state="待排（未登记）",
|
||
why="登记表里这条还没排上自动化号 ⇒ 必须新建（幂等待办）",
|
||
need_follow=True, action=ACTION_RESCHEDULE)
|
||
rec["new_files"] = new_files(d["ws"], d["watch"], parse_iso_ms(d["scheduled_at"]) or 0)
|
||
return rec
|
||
|
||
arow = con.execute(
|
||
"select status,scheduled_at,next_run_at,last_run_at from automations where id=?",
|
||
(d["auto_id"],)).fetchone()
|
||
if not arow:
|
||
rec.update(state="异常",
|
||
why="登记里这条自动化**查不到**（被删了？）⇒ 必须新建",
|
||
need_follow=True, action=ACTION_RESCHEDULE)
|
||
return rec
|
||
rec["auto_status"] = arow[0]
|
||
rec["auto_next"] = fmt_ms(arow[2])
|
||
rec["auto_last"] = fmt_ms(arow[3])
|
||
|
||
# ③ 此刻在不在跑
|
||
rrow = con.execute(
|
||
"select running,running_started_at,last_error,running_conversation_id "
|
||
"from automation_runtime_state where automation_id=?", (d["auto_id"],)).fetchone()
|
||
running = bool(rrow[0]) if rrow else False
|
||
running_since = fmt_ms(rrow[1]) if rrow and rrow[1] else None
|
||
last_err = (rrow[2] if rrow else None)
|
||
conv = (rrow[3] if rrow else None)
|
||
rec["running"] = running
|
||
rec["running_since"] = running_since
|
||
if last_err:
|
||
rec["last_error"] = str(last_err)[:120]
|
||
|
||
# ② 跑没跑 ＋ 结论原文
|
||
runs = con.execute(
|
||
"select created_at,status,result_success,thread_title,failure_code,reason_code "
|
||
"from automation_runs where automation_id=? and coalesce(created_at,0)>? "
|
||
"order by created_at desc", (d["auto_id"], sched_ms)).fetchall()
|
||
rec["run_count"] = len(runs)
|
||
if runs:
|
||
r0 = runs[0]
|
||
rec["run_at"] = fmt_ms(r0[0])
|
||
rec["run_status"] = r0[1]
|
||
rec["success"] = (r0[2] == 1)
|
||
rec["conclusion"] = (r0[3] or "")[:300]
|
||
if r0[4] or r0[5]:
|
||
rec["failure"] = "%s/%s" % (r0[4], r0[5])
|
||
if conv:
|
||
rec["conversation"] = str(conv)[:12]
|
||
|
||
# ④ 会话
|
||
rec["sessions"] = session_stats(con, d["ws"], sched_ms)
|
||
|
||
# ⑥ 产出
|
||
rec["new_files"] = new_files(d["ws"], d["watch"], sched_ms)
|
||
|
||
# ⑦ 结果钩子的台账（这条链的"接收"端）
|
||
mine, bad = ledger_for(ledger or [], d["ws"], sched_ms)
|
||
rec["hook"] = {"records": len(mine), "bad": [r.get("verdict") for r in bad],
|
||
"last": (mine[-1].get("ts") if mine else None)}
|
||
|
||
# ---- 判定 ----
|
||
if not runs and not running:
|
||
if NOW_MS < sched_ms:
|
||
rec.update(state="待跑", why="还没到点", need_follow=False, action=ACTION_WAIT)
|
||
else:
|
||
overdue = (NOW_MS - sched_ms) / 60000.0
|
||
if overdue < 3:
|
||
rec.update(state="待跑", why="到点 %.1f 分钟，尚未见运行记录" % overdue,
|
||
need_follow=False, action=ACTION_WAIT)
|
||
else:
|
||
rec.update(state="异常",
|
||
why="已过点 %.0f 分钟仍**无任何运行记录** ⇒ 自动化没触发" % overdue,
|
||
need_follow=True, action=ACTION_RESCHEDULE)
|
||
return rec
|
||
|
||
if running:
|
||
stuck = ""
|
||
if running_since:
|
||
try:
|
||
start = int(datetime.strptime(
|
||
"%s %s" % (datetime.now().strftime("%Y-%m-%d"), running_since),
|
||
"%Y-%m-%d %H:%M:%S").timestamp() * 1000)
|
||
mins = (NOW_MS - start) / 60000.0
|
||
if mins > 60:
|
||
stuck = "（已跑 %.0f 分钟 ⇒ **疑似卡住**）" % mins
|
||
except Exception:
|
||
pass
|
||
rec.update(state="进行中", why="正在跑" + stuck,
|
||
need_follow=bool(stuck),
|
||
action=(ACTION_REPORT if stuck else ACTION_WAIT))
|
||
return rec
|
||
|
||
if runs:
|
||
if rec.get("success"):
|
||
# 🔴 先判「跑完了但**没干活**」—— 这一条必须排在"有没有产出"之前。
|
||
# 判据 = 结论原文自述未执行（机器可读）。判错这里 = 那条活静默死掉（2026-09-28 实测）。
|
||
con_text = rec.get("conclusion") or ""
|
||
hit = [m for m in PENDING_MARKERS if m in con_text]
|
||
if hit:
|
||
rec.update(state="跑完·未实际执行",
|
||
why="运行成功，但结论自述未执行（命中：%s）⇒ **等同异常**，必须重排" % "、".join(hit[:4]),
|
||
need_follow=True, action=ACTION_RESCHEDULE)
|
||
elif rec["new_files"]:
|
||
# 🔴 交叉校验：**自动化说成功 ＋ 有产出，但钩子台账里这条线有"非正常收尾"**
|
||
# ⇒ 两个信号打架 ⇒ 以**证据**为准，交给下一棒看一眼（⛔ 不自动重排，
|
||
# 因为产出文件确实在，重排可能重复做同一件事）。
|
||
if rec["hook"]["bad"]:
|
||
rec.update(state="已收官·但钩子有异常收尾",
|
||
why="有产出，但结果钩子记录到非正常收尾：%s ⇒ 需人判读"
|
||
% "、".join(rec["hook"]["bad"][:4]),
|
||
need_follow=True, action=ACTION_REPORT)
|
||
else:
|
||
rec.update(state="已收官", why="有结论 ＋ 有产出",
|
||
need_follow=False, action=ACTION_WAIT)
|
||
else:
|
||
rec.update(state="收官·疑未产出",
|
||
why="跑完了但收件目录里**没有新文件** ⇒ 核实是否只是「只在对话里答了」",
|
||
need_follow=True, action=ACTION_REPORT)
|
||
else:
|
||
rec.update(state="失败",
|
||
why="运行记录 result_success != 1" + ("（%s）" % rec.get("failure", "") if rec.get("failure") else ""),
|
||
need_follow=True, action=ACTION_RESCHEDULE)
|
||
return rec
|
||
|
||
rec.update(state="未知", why="信号不足", need_follow=True, action=ACTION_REPORT)
|
||
return rec
|
||
|
||
|
||
def main():
|
||
as_json = "--json" in sys.argv
|
||
con = ro_conn()
|
||
|
||
lk = lock_state()
|
||
dl = domain_locks()
|
||
ledger = read_ledger()
|
||
hh = hook_health(ledger)
|
||
results = [check_one(con, d, ledger) for d in DISPATCHES]
|
||
|
||
# 未收官且有在跑的 ⇒ 锁被占属正常；全部停跑但锁还占着 ⇒ 异常（本项目红线：锁只能持有者释放）
|
||
lock_orphan = bool(lk["held"]) and all(
|
||
(not r.get("running")) and r.get("state") in ("已收官", "失败", "异常") for r in results)
|
||
|
||
follow = [r for r in results if r.get("need_follow")]
|
||
rc = 0
|
||
if follow or lock_orphan:
|
||
rc = 2
|
||
elif any(r["state"] in ("待跑", "进行中") for r in results):
|
||
rc = 1
|
||
|
||
if as_json:
|
||
out(json.dumps({"now": fmt_ms(NOW_MS), "lock": lk, "domain_locks": dl,
|
||
"hook_health": hh, "lines": results,
|
||
"lock_orphan": lock_orphan, "rc": rc},
|
||
ensure_ascii=False, indent=2))
|
||
return rc
|
||
|
||
out("═══ 派活监管 · 三线执行巡检 @ %s ═══" % fmt_ms(NOW_MS))
|
||
out("")
|
||
out("【全局执行锁】%s%s" % (
|
||
("🔴 被占：%s" % lk["owner"]) if lk["held"] else "✅ 空闲",
|
||
" ＋ 域锁：%s" % ", ".join(dl) if dl else ""))
|
||
if lock_orphan:
|
||
out(" ⛔ **锁被占但三线都已停跑** ⇒ 疑似未释放（红线：锁只能持有者自己释放，⛔ 不许代为删锁）")
|
||
out("")
|
||
out("【结果钩子】%s 台账记录=%d 最近=%s" % (
|
||
"🟢 在工作" if hh["ok"] else "🔴 **未工作**", hh["records"], hh["last_ts"] or "-"))
|
||
if hh["problem"]:
|
||
out(" ⚠️ %s" % hh["problem"])
|
||
out("")
|
||
|
||
icon = {"已收官": "🟢", "进行中": "🔵", "待跑": "⚪", "失败": "🔴",
|
||
"异常": "🔴", "收官·疑未产出": "🟡", "跑完·未实际执行": "🔴",
|
||
"待排（未登记）": "🔴", "未知": "🟡"}
|
||
for r in results:
|
||
out("%s 【%s】%s (排在 %s)" % (icon.get(r["state"], "·"), r["line"], r["state"], r["at"]))
|
||
out(" 任务：%s" % r["task"])
|
||
if r.get("running"):
|
||
out(" 此刻：**正在跑**" + ("（起于 %s）" % r["running_since"] if r.get("running_since") else ""))
|
||
if r.get("run_count"):
|
||
out(" 运行：%s 次 · 最近 %s · 成功=%s" % (
|
||
r["run_count"], r.get("run_at", "-"), "是" if r.get("success") else "否/未定"))
|
||
con_txt = (r.get("conclusion") or "").replace("\n", " ").strip()
|
||
if con_txt:
|
||
out(" 结论：%s" % con_txt[:160])
|
||
if r.get("last_error") or r.get("failure"):
|
||
out(" 🔴 错误：%s %s" % (r.get("last_error", ""), r.get("failure", "")))
|
||
for s in r["sessions"][:3]:
|
||
out(" 会话：%s status=%s 创建=%s 最后活动=%s" % (
|
||
s["id"], s["status"], s["created"], s["last_activity"]))
|
||
if r["new_files"]:
|
||
out(" 产出（新）：")
|
||
for f in r["new_files"][:4]:
|
||
out(" · %s (%s, %s B)" % (f["path"], f["mtime"], f["size"]))
|
||
else:
|
||
out(" 产出（新）：**无**")
|
||
hk = r.get("hook") or {}
|
||
if hk.get("records"):
|
||
out(" 钩子台账：%d 条 · 最近 %s%s" % (
|
||
hk["records"], hk.get("last") or "-",
|
||
(" · 🔴 非正常收尾：" + "、".join(hk["bad"])) if hk.get("bad") else ""))
|
||
out(" 判读：%s" % r["why"])
|
||
out(" ➡️ **action = %s**%s" % (r.get("action", "?"),
|
||
ACTIONS_HELP.get(r.get("action"), "")))
|
||
out("")
|
||
|
||
# ---- 汇总：按 action 分组（巡检棒照 action 办事，⛔ 别自己解释状态）----
|
||
by_action = {}
|
||
for r in results:
|
||
by_action.setdefault(r.get("action", "?"), []).append(r)
|
||
if lock_orphan:
|
||
by_action.setdefault(ACTION_REPORT_LOCK, []).append({"line": "（全局）", "why": "锁被占但三线都已停跑 ⇒ 疑似未释放"})
|
||
if not hh["ok"]:
|
||
# 钩子没工作 ⇒ 必须报出来（⛔ 否则"台账为空"会被读成"一切正常"）
|
||
by_action.setdefault(ACTION_REPORT, []).append(
|
||
{"line": "（结果钩子）", "why": hh["problem"] or "钩子未工作"})
|
||
|
||
if not by_action or set(by_action) == {ACTION_WAIT}:
|
||
out("⇒ ✅ 无需动作（wait）：%s" % (
|
||
"全部收官" if all(r["state"] == "已收官" for r in results) else "正常等待"))
|
||
return rc
|
||
|
||
out("⇒ 本轮动作清单（**照 action 办事**）：")
|
||
for act in (ACTION_RESCHEDULE, ACTION_REPORT_LOCK, ACTION_REPORT, ACTION_WAIT):
|
||
items = by_action.get(act)
|
||
if not items:
|
||
continue
|
||
out(" 【%s】%s" % (act, ACTIONS_HELP.get(act, "")))
|
||
for r in items:
|
||
out(" · 【%s】%s" % (r["line"], r["why"]))
|
||
return rc
|
||
|
||
|
||
ACTIONS_HELP = {
|
||
ACTION_RESCHEDULE: "⇒ 🔴 **必须当场新建一条一次性自动化**给该线（⛔ 改时间不触发；⛔ 不许只报告），并把新 id 写回脚本顶部 DISPATCHES",
|
||
ACTION_REPORT_LOCK: "⇒ ⛔ **只报告用户**：不代删锁、不接管（红线）",
|
||
ACTION_REPORT: "⇒ 报告 ＋ 列期望产出路径（须人判读，别自建自动化）",
|
||
ACTION_WAIT: "⇒ 不做任何事（正常等待，无事不打扰）",
|
||
}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
sys.exit(main())
|
||
except Exception as e:
|
||
out("⛔ 巡检脚本自身出错：%s: %s" % (type(e).__name__, e))
|
||
sys.exit(3)
|