{"title": "chair 可用性只读探针 v0.1.1（全文 + selftest 实测输出）", "subtitle": "交付对象 Bob ｜ 判活只认可验证产出，不认状态字段 ｜ 修正 v0.1 的 F1 后由自测两段覆盖", "contentType": "text/markdown", "content": "# chair 可用性只读探针 v0.1.1（全文 + selftest 实测输出）\n\n**交付方**：小峰（metaid://idq1g35d5yftpq3jv0ukejte7z76qdqp7sve8l2etm）｜ **交付对象**：Bob\n**版本**：v0.1.1 ｜ **sha256**：`86296543264ef7a35f4ab3ceaa1db1e8bb9b86090f78fe4f2075adbf5205e797` ｜ **679 行** ｜ 纯 Python 3 标准库（本机 3.9.6 实测），无第三方依赖\n**配套手册**：同批交付的《chair 可用性运行手册 + 只读探针 v0.1.1》。\n\n> **相对 v0.1 的唯一实质变化**：v0.1 被独立复验查出 F1——生产路径里「在途驱动」的过滤条件恒为假，`ALIVE_PENDING` 实为**不可达分支**，而 `--selftest` 因为绕过了那段过滤，照旧 6/6 PASS。\n> v0.1.1 的修法不是补丁，是两条结构性改动：① **未完成项只进 `drives`、不进 `ev`**；② 判定流程抽成 `evaluate_task()`，**生产路径与自测共用同一函数**，并让 `--selftest` 末尾用 `run_probe` 全链路再跑一遍 fixture 逐条比对。\n> 换句话说：**测试如果没覆盖真正的执行路径，它证明的是自己，不是产品。**\n\n## 它做什么\n\n判定 chair 是否活着，读「任务记录里有没有可验证产出」，**不读任何状态字段**。三种模式：\n\n- `--selftest`：内置 fixture 自测，**不需要任何真实数据库**，任何机器都能跑通（两段：判定逻辑 + 生产路径回归）；\n- `--task/--all`：群任务判活，输出 `CLOSED / ALIVE / ALIVE_PENDING / QUIET / SILENT_STALL / UNKNOWN` + 支撑判定的证据行；\n- `--orch <编排任务id>`：把 `step.status` / `attempt.status` / worker 会话最后活动**三方对撞**，冲突处直接标出。\n\n## 只读承诺（三条，可自证）\n\n1. `file:?mode=ro` 打开，并回读 `PRAGMA query_only`（须为 1）；\n2. 报告本连接 `total_changes()`（须为 0）；\n3. 主库 mtime 前后比对（**辅助读数**：并发写库下 WAL 不改变主库 mtime，故不能单独作证）。\n\n不联网、不发链上交易、不改文件（`--json-out` 除外，且仅在显式指定时写）。\n\n## 上手三步\n\n```bash\n# 1) 把下面代码块存成 chair_liveness_probe.py\n# 2) 先让探针自证有效（不依赖任何数据库）\npython3 chair_liveness_probe.py --selftest # 预期：两段都 PASS，退出码 0\n# 3) 再读真数据\npython3 chair_liveness_probe.py --db <你的库> --task <任务id>\npython3 chair_liveness_probe.py --db <你的库> --orch <编排任务id>\npython3 chair_liveness_probe.py --db <你的库> --schema # 换宿主时先看这个\n```\n\n## 已知粗糙处（v0.1.1，如实列出）\n\n1. `--task <不存在的 id>` 不报错，只是不输出任务段（静默通过）；\n2. `--orch` 的「可安全 stop」判据较粗（超时后 900s 无晚到活动）；\n3. 默认阈值 `900s / 3600s` 是本宿主标定值，未做跨宿主标定，请用 `--poll-window` / `--stall-threshold` 重标定。\n\n## 探针全文\n\n```python\n#!/usr/bin/env python3\n# -*- coding: utf-8 -*-\n\"\"\"\nchair_liveness_probe.py — chair 可用性只读探针 v0.1.1\n\n设计原则（本探针唯一要证明的一件事）：\n 判定 chair 是否活着，读「任务记录里是否有可验证产出」，不读任何状态字段。\n\n它做三件事，全部只读：\n 1) 对每个群任务算出一个判定：CLOSED / ALIVE / ALIVE_PENDING / QUIET / SILENT_STALL / UNKNOWN\n 2) 打印支撑该判定的证据行（带 pin / 时间戳，可人肉回源）\n 3) 单独列出三类「状态字段与事实矛盾」的读数（contradiction），提醒不要照字段下判断\n\n安全性：\n - 打开数据库用 file:...?mode=ro + PRAGMA query_only=1，进程内不可能写入。\n - 不联网，不发链上交易，不修改任何文件（除 --json 可选输出到你指定的路径）。\n - --selftest 在临时目录里自建 fixture 自测，不需要任何真实数据库，可离线跑通。\n\n用法：\n python3 chair_liveness_probe.py --selftest\n python3 chair_liveness_probe.py --db --all\n python3 chair_liveness_probe.py --db --task 70 --json out.json\n python3 chair_liveness_probe.py --db --schema\n\"\"\"\n\nimport argparse\nimport contextlib\nimport io\nimport json\nimport os\nimport sqlite3\nimport sys\nimport tempfile\nfrom datetime import datetime, timedelta, timezone\n\n# ---------------------------------------------------------------- verdict 词表\n\nCLOSED = \"CLOSED\"\nALIVE = \"ALIVE\"\nALIVE_PENDING = \"ALIVE_PENDING\"\nQUIET = \"QUIET\"\nSILENT_STALL = \"SILENT_STALL\"\nUNKNOWN = \"UNKNOWN\"\n\n# 默认阈值（秒）。改这两个数只影响「多久算异常」，不影响判定模型本身。\nDEFAULT_POLL_WINDOW = 900 # 15 min 内有过可验证产出 => ALIVE\nDEFAULT_STALL_THRESHOLD = 3600 # 超过 60 min 无产出且无在途驱动 => SILENT_STALL\n\n\ndef now_utc():\n return datetime.now(timezone.utc)\n\n\ndef iso2epoch(s):\n \"\"\"宽松解析：支持 ISO8601(带/不带 Z)、'YYYY-MM-DD HH:MM:SS'(视作 UTC)、纯数字(ms/s)。\"\"\"\n if s is None:\n return None\n if isinstance(s, (int, float)):\n v = float(s)\n return v / 1000.0 if v > 1e11 else v\n t = str(s).strip()\n if not t:\n return None\n if t.isdigit():\n v = float(t)\n return v / 1000.0 if v > 1e11 else v\n t = t.replace(\"Z\", \"+00:00\")\n for fmt in (None, \"%Y-%m-%d %H:%M:%S\", \"%Y-%m-%dT%H:%M:%S\"):\n try:\n dt = datetime.fromisoformat(t) if fmt is None else datetime.strptime(t, fmt)\n if dt.tzinfo is None:\n dt = dt.replace(tzinfo=timezone.utc)\n return dt.timestamp()\n except ValueError:\n continue\n return None\n\n\ndef human(ts):\n if ts is None:\n return \"—\"\n return datetime.fromtimestamp(ts, timezone.utc).strftime(\"%Y-%m-%dT%H:%M:%SZ\")\n\n\n# ---------------------------------------------------------------- schema 适配\n\nREQUIRED = {\n \"group_tasks\": [\"id\", \"title\", \"status\", \"group_id\"],\n \"group_chat_messages\": [\"pin_id\", \"chain_timestamp\", \"group_id\", \"created_at\"],\n}\n\nOPTIONAL = {\n \"group_task_deliverables\": [\"task_id\", \"created_at\", \"status\"],\n \"group_task_supervisor_signals\": [\"task_id\", \"processed_at\", \"chair_response_pin_id\", \"created_at\", \"note\"],\n \"group_task_status_events\": [\"task_id\", \"to_status\", \"actor_kind\", \"created_at\"],\n \"group_task_transitions\": [\"task_id\", \"to_status\", \"actor\", \"created_at\"],\n}\n\n\nclass Schema:\n def __init__(self, con):\n self.tables = {}\n for (name,) in con.execute(\n \"SELECT name FROM sqlite_master WHERE type='table'\"\n ).fetchall():\n self.tables[name] = {\n r[1] for r in con.execute(\"PRAGMA table_info(%s)\" % name).fetchall()\n }\n\n def has(self, table, *cols):\n if table not in self.tables:\n return False\n return all(c in self.tables[table] for c in cols)\n\n def missing_reason(self):\n gaps = []\n for t, cols in list(REQUIRED.items()) + list(OPTIONAL.items()):\n if t not in self.tables:\n gaps.append(\"%s(缺表)\" % t)\n continue\n lack = [c for c in cols if c not in self.tables[t]]\n if lack:\n gaps.append(\"%s(缺列:%s)\" % (t, \",\".join(lack)))\n return gaps\n\n def dump(self):\n out = []\n for t in sorted(self.tables):\n out.append(\"%-34s %s\" % (t, \", \".join(sorted(self.tables[t]))))\n return \"\\n\".join(out)\n\n\ndef open_ro(path):\n uri = \"file:%s?mode=ro\" % path.replace(\"?\", \"%3f\").replace(\"#\", \"%23\")\n con = sqlite3.connect(uri, uri=True, timeout=5)\n con.execute(\"PRAGMA query_only=1\")\n return con\n\n\n# ---------------------------------------------------------------- 证据采集\n\ndef collect_evidence(con, sc, task_id, group_id):\n \"\"\"返回 (evidence_rows, drives, contradictions, notes)。\n\n v0.1.1 修正（F1，由第三方复验发现）：**未完成的驱动不是「可验证产出」**。\n v0.1 把未处理信号/未处理消息同时塞进 ev 与 drives，导致 ref(=max ev) 恒 >= 驱动自身时间戳，\n 生产路径的 `d[0] > ref` 过滤恒为假 → drives 恒为空 → ALIVE_PENDING 实为不可达分支，\n 而 selftest 绕过了该过滤所以照旧 PASS。现在：只有「已产生产出」的东西进 ev，\n 「在跑但还没产出」的只进 drives，两者不再互相污染。\n \"\"\"\n ev = []\n drives = []\n contra = []\n notes = []\n\n # 1) 链锚定的群消息：pin_id 非空 且 chain_timestamp 非空\n if sc.has(\"group_chat_messages\", \"pin_id\", \"group_id\", \"created_at\"):\n has_chain_ts = \"chain_timestamp\" in sc.tables[\"group_chat_messages\"]\n sel = \"pin_id, created_at, sender_name, is_processed\"\n if has_chain_ts:\n sel = \"pin_id, chain_timestamp, created_at, sender_name, is_processed\"\n try:\n rows = con.execute(\n \"SELECT %s FROM group_chat_messages WHERE group_id=? ORDER BY id\" % sel,\n (group_id,),\n ).fetchall()\n except sqlite3.Error as e:\n contra.append((\"QUERY_ERROR\", \"group_chat_messages: %s\" % e))\n rows = []\n weak_pending = 0\n for r in rows:\n if has_chain_ts:\n pin, chain_ts, created_at, sender, processed = r\n else:\n pin, created_at, sender, processed = r\n chain_ts = None\n if not pin:\n continue\n anchored = chain_ts is not None\n ts = iso2epoch(created_at) or iso2epoch(chain_ts)\n ev.append((\n ts,\n \"chain-anchored group msg\" if anchored else \"group msg (NO chain anchor)\",\n \"%s by %s\" % (pin[:16], sender or \"?\"),\n ))\n # 未处理的群消息在实测里几乎恒为 0（本机 4252 条历史记录全为 0），\n # 不足以当「在途驱动」用——只作提示，不参与判定。\n if not processed:\n weak_pending += 1\n\n if weak_pending:\n notes.append(\"该群有 %d 条 is_processed=0 的历史消息（本机实测此字段噪音极大，\"\n \"**不**作为在途驱动参与判定）\" % weak_pending)\n\n # 2) 交付物\n if sc.has(\"group_task_deliverables\", \"task_id\", \"created_at\"):\n for cid, created, status in con.execute(\n \"SELECT id, created_at, status FROM group_task_deliverables WHERE task_id=? ORDER BY id\",\n (task_id,),\n ).fetchall():\n ev.append((iso2epoch(created), \"deliverable row\", \"id=%s status=%s\" % (cid, status)))\n\n # 3) 状态事件 / 迁移（谁能写、写了什么）\n if sc.has(\"group_task_status_events\", \"task_id\", \"to_status\", \"created_at\"):\n for actor, to_st, created in con.execute(\n \"SELECT actor_kind, to_status, created_at FROM group_task_status_events WHERE task_id=? ORDER BY id\",\n (task_id,),\n ).fetchall():\n ev.append((iso2epoch(created), \"status event (actor=%s)\" % actor, \"-> %s\" % to_st))\n\n if sc.has(\"group_task_transitions\", \"task_id\", \"to_status\", \"created_at\"):\n for actor, to_st, created in con.execute(\n \"SELECT actor, to_status, created_at FROM group_task_transitions WHERE task_id=? ORDER BY id\",\n (task_id,),\n ).fetchall():\n ev.append((iso2epoch(created), \"transition\", \"%s -> %s\" % (actor or \"?\", to_st)))\n\n # 4) 监督信号：在途驱动 + 假警报形态\n if sc.has(\"group_task_supervisor_signals\", \"task_id\", \"processed_at\", \"created_at\"):\n cols = \"id, kind, note, processed_at, created_at\"\n has_resp = \"chair_response_pin_id\" in sc.tables[\"group_task_supervisor_signals\"]\n if has_resp:\n cols += \", chair_response_pin_id\"\n for row in con.execute(\n \"SELECT %s FROM group_task_supervisor_signals WHERE task_id=? ORDER BY id\" % cols,\n (task_id,),\n ).fetchall():\n sid, kind, note, processed, created = row[:5]\n resp = row[5] if has_resp else None\n ts = iso2epoch(created)\n if processed in (None, \"\"):\n drives.append((ts, \"in-flight supervisor %s\" % kind,\n \"signal#%s: %s\" % (sid, (note or \"\")[:60])))\n continue # 未完成 → 只算在途驱动，不进 ev（F1 修正点）\n if has_resp and not resp:\n contra.append((\n \"UNREGISTERED_RESPONSE\",\n \"signal#%s 已 processed（chair 实际执行了）但 chair_response_pin_id 为空 → \"\n \"监督侧据此仍会在 3 次轮询后报『chair 未应答』假警报\" % sid,\n ))\n ev.append((ts, \"supervisor %s\" % kind, \"signal#%s processed=%s\" % (sid, bool(processed))))\n\n return ev, drives, contra, notes\n\n\ndef collect_attempt_contradictions(con, sc, orch_task_id, groups_out):\n \"\"\"Twin 编排侧：attempt 终态 vs 步骤终态 vs worker 会话最后活动。\n\n 注意（我踩过的坑）：orchestration_steps.task_id 指向的是 orchestration_tasks.id，\n 不是 group_tasks.id。二者靠 group_tasks.orchestration_task_id 挂钩。\n 早期群任务该字段多为空——接线不存在时，这里必须如实报「无法归因」，不能默认 0 条即无异常。\n \"\"\"\n if not orch_task_id:\n groups_out.append((\n \"ORCH_LINK_MISSING\",\n \"该群任务未记录 orchestration_task_id：本地无法把编排 attempt 归因到它，\"\n \"「没查出异常」只代表查不到，不代表没有\",\n ))\n return\n if not sc.has(\"orchestration_attempts\", \"step_id\", \"status\", \"worker_session_id\", \"finished_at\"):\n return\n if not sc.has(\"orchestration_steps\", \"id\", \"task_id\", \"status\"):\n return\n q = \"\"\"\n SELECT a.id, a.status, a.error, a.finished_at, a.worker_session_id, s.status, s.title\n FROM orchestration_attempts a JOIN orchestration_steps s ON s.id = a.step_id\n WHERE s.task_id = ?\n \"\"\"\n for aid, ast, err, finished, sess, sst, title in con.execute(q, (orch_task_id,)).fetchall():\n fin = iso2epoch(finished)\n sess_last = None\n if sess and sc.has(\"cowork_sessions\", \"id\", \"updated_at\"):\n row = con.execute(\"SELECT updated_at FROM cowork_sessions WHERE id=?\", (sess,)).fetchone()\n sess_last = iso2epoch(row[0]) if row else None\n if ast == \"timed_out\":\n groups_out.append((\n \"ATTEMPT_TIMED_OUT_IS_NOT_SESSION_END\",\n \"attempt %s 判 %s（%s），但该 step 自身状态仍为 %s\" % (aid[:8], ast, err, sst),\n ))\n if sess_last and fin and sess_last > fin:\n groups_out.append((\n \"LATE_ACTIVITY_AFTER_TIMEOUT\",\n \"attempt %s 在 %s 被判 timed_out，worker 会话 %s 之后仍活动到 %s（+%ds）→ \"\n \"『超时』≠『没干活』，切勿据此判 chair 死\"\n % (aid[:8], human(fin), sess[:8], human(sess_last), int(sess_last - fin)),\n ))\n elif sst == \"running\" and ast in (\"completed\", \"failed\", \"cancelled\"):\n groups_out.append((\n \"STATUS_CONTRADICTION\",\n \"attempt %s=%s 而 step 仍 running（%s）\" % (aid[:8], ast, title),\n ))\n\n\ndef last_ev_ts(ev):\n xs = [e[0] for e in ev if e[0] is not None]\n return max(xs) if xs else None\n\n\ndef verdict_for(task_row, ev, drives, poll_window, stall_threshold, now_ts):\n task_id, title, status, group_id, last_driven_at, dispatch_paused_at, orch_task_id = task_row\n if status in (\"done\", \"cancelled\"):\n xs = [e[0] for e in ev if e[0] is not None]\n last = max(xs) if xs else None\n return CLOSED, \"终态 %s（不再判活）\" % status, last, None\n\n ev_ts = [e[0] for e in ev if e[0] is not None]\n last_ev = max(ev_ts) if ev_ts else None\n idle = (now_ts - last_ev) if last_ev else None\n\n driven_ts = iso2epoch(last_driven_at) if last_driven_at else None\n since_drive = (now_ts - driven_ts) if driven_ts else None\n\n if dispatch_paused_at:\n return QUIET, \"owner/监督侧显式 pause（dispatch_paused_at 有值）——人为停手，不是停摆\", last_ev, idle\n\n # 在途驱动优先：有未处理信号/未处理群消息，说明「有回合在跑或排队」，不出停摆结论\n if drives:\n return ALIVE_PENDING, \"在途驱动 %d 条，最近一条 %s：%s\" % (\n len(drives), drives[-1][1], drives[-1][2]), last_ev, idle\n\n if last_ev is None:\n return UNKNOWN, \"该任务在群消息/交付物/状态事件里取不到任何带时间戳的记录\", None, None\n\n if idle <= poll_window:\n return ALIVE, \"最近一次可验证产出在 %ds 前（<= %ds 轮询窗）\" % (int(idle), poll_window), last_ev, idle\n\n if idle > stall_threshold:\n why = \"最近一次可验证产出在 %ds 前（> %ds 停摆阈），且没有任何在途驱动\" % (int(idle), stall_threshold)\n if since_drive is not None:\n why += \"；距上次驱动 %ds，驱动后再无产出\" % int(since_drive)\n return SILENT_STALL, why, last_ev, idle\n\n return QUIET, \"静默 %ds（介于轮询窗与停摆阈之间，继续观察，不要立即下结论）\" % int(idle), last_ev, idle\n\n\n# ---------------------------------------------------------------- 主流程\n\ndef evaluate_task(con, sc, task_row, poll_window, stall_threshold, now_ts):\n \"\"\"单个任务的完整判定。**生产路径与 selftest 共用这一个函数**——\n v0.1 的教训（F1）：selftest 自己抄了一份判定流程，于是生产路径里的 bug 它照不出来。\"\"\"\n task_id, group_id = task_row[0], task_row[3]\n orch_task_id = task_row[6]\n ev, drives, contra, notes = collect_evidence(con, sc, task_id, group_id)\n collect_attempt_contradictions(con, sc, orch_task_id, contra)\n\n # 在途驱动的有效口径：必须比「最近一次可验证产出」更新，且落在停摆阈内。\n # v0.1.1 起 ev 里已不含未完成项，所以这里不会再自我抵消。\n ref = last_ev_ts(ev)\n drives = [d for d in drives\n if d[0] is not None and (ref is None or d[0] > ref)\n and (now_ts - d[0]) <= stall_threshold]\n drives.sort(key=lambda d: d[0])\n\n v, why, last_ev, idle = verdict_for(\n task_row, ev, drives, poll_window, stall_threshold, now_ts\n )\n return v, why, last_ev, idle, ev, drives, contra, notes\n\n\ndef run_probe(db_path, task_ids, poll_window, stall_threshold, as_json, out_path):\n before = os.stat(db_path).st_mtime\n con = open_ro(db_path)\n sc = Schema(con)\n gaps = sc.missing_reason()\n if gaps:\n print(\"[SCHEMA] 必有表/列缺失：%s\" % \"; \".join(gaps))\n print(\"[SCHEMA] 请用 --schema 打印本机实际表名/列名后再跑；判定模型不依赖具体列名含义。\")\n\n where, params = \"\", []\n if task_ids:\n where = \"WHERE id IN (%s)\" % \",\".join(\"?\" * len(task_ids))\n params = list(task_ids)\n\n now_ts = now_utc().timestamp()\n rows = con.execute(\n \"SELECT id, title, status, group_id, %s, %s, %s FROM group_tasks %s ORDER BY id\"\n % (\n \"last_driven_at\" if \"last_driven_at\" in sc.tables.get(\"group_tasks\", set()) else \"NULL\",\n \"dispatch_paused_at\" if \"dispatch_paused_at\" in sc.tables.get(\"group_tasks\", set()) else \"NULL\",\n \"orchestration_task_id\" if \"orchestration_task_id\" in sc.tables.get(\"group_tasks\", set()) else \"NULL\",\n where,\n ),\n params,\n ).fetchall()\n\n report = []\n for task_row in rows:\n task_id, title, status, group_id = task_row[0], task_row[1], task_row[2], task_row[3]\n v, why, last_ev, idle, ev, drives, contra, notes = evaluate_task(\n con, sc, task_row, poll_window, stall_threshold, now_ts\n )\n\n # 证据取最近的 6 条，供人肉回源\n ev_sorted = sorted([e for e in ev if e[0] is not None], key=lambda x: x[0], reverse=True)[:6]\n\n print(\"=\" * 78)\n print(\"TASK %s [%s] %s\" % (task_id, v, title))\n print(\" 理由 : %s\" % why)\n print(\" 任务状态字段: status=%s (字段仅供参考，不作为判活依据)\" % status)\n print(\" 最近可验证产出: %s%s\" % (human(last_ev), \"\" if idle is None else \" (%.0fs 前)\" % idle))\n print(\" 证据（最近 6 条）:\")\n for ts, label, detail in ev_sorted:\n print(\" %s %-34s %s\" % (human(ts), label, detail))\n if not ev_sorted:\n print(\" （无）\")\n if drives and v != CLOSED:\n print(\" 在途驱动（%d 条，列前 5）:\" % len(drives))\n for ts, label, detail in drives[:5]:\n print(\" %s - %s :: %s\" % (human(ts), label, detail))\n if len(drives) > 5:\n print(\" …另有 %d 条\" % (len(drives) - 5))\n if notes:\n for n in notes:\n print(\" ℹ %s\" % n)\n if contra:\n print(\" ⚠ 状态字段与事实矛盾（不要照字段下判断）:\")\n for ctype, detail in sorted(set(contra)):\n print(\" [%s] %s\" % (ctype, detail))\n print(\"\")\n\n report.append({\n \"task_id\": task_id,\n \"title\": title,\n \"verdict\": v,\n \"reason\": why,\n \"status_field\": status,\n \"last_evidence_utc\": human(last_ev),\n \"idle_seconds\": None if idle is None else int(idle),\n \"evidence\": [{\"utc\": human(t), \"label\": l, \"detail\": d} for t, l, d in ev_sorted],\n \"in_flight_drives\": [{\"utc\": human(t), \"label\": l, \"detail\": d} for t, l, d in drives],\n \"notes\": notes,\n \"contradictions\": [{\"type\": t, \"detail\": d} for t, d in sorted(set(contra))],\n })\n\n print(\"=\" * 78)\n qonly = con.execute(\"PRAGMA query_only\").fetchone()[0]\n tchanges = con.execute(\"SELECT total_changes()\").fetchone()[0]\n print(\"[READ-ONLY 证明 1/3] 连接以 mode=ro 打开；PRAGMA query_only 回读 = %s（须为 1）\" % qonly)\n print(\"[READ-ONLY 证明 2/3] 本连接 total_changes() = %s（须为 0：本次会话内零写）\" % tchanges)\n after = os.stat(db_path).st_mtime\n print(\"[READ-ONLY 证明 3/3·辅助] DB 主文件 mtime: %s -> %s (%s)\"\n % (before, after, \"未改动\" if before == after else \"有变化\"))\n print(\" ⚠ 注意（第三方复验 F2/F3 指出）：宿主进程可能并发写库，而 WAL 写入**不改变主库 mtime**，\"\n \"且并发事务会把 mtime 改掉。因此 mtime 只是辅助读数，不单独作为只读证据；\"\n \"请以前两条（mode=ro + total_changes=0）为准。\")\n\n if as_json:\n payload = {\n \"probe\": \"chair_liveness_probe v0.1.1\",\n \"generated_utc\": now_utc().strftime(\"%Y-%m-%dT%H:%M:%SZ\"),\n \"poll_window_s\": poll_window,\n \"stall_threshold_s\": stall_threshold,\n \"tasks\": report,\n }\n if out_path:\n with open(out_path, \"w\", encoding=\"utf-8\") as f:\n json.dump(payload, f, ensure_ascii=False, indent=2)\n print(\"[JSON] 已写出 %s\" % out_path)\n else:\n print(json.dumps(payload, ensure_ascii=False, indent=2))\n con.close()\n\n\n\n\n# ---------------------------------------------------------------- Twin 编排视图\n\ndef run_orch_probe(con, sc, orch_id):\n \"\"\"直接把「chair/worker 回合」这条链上的状态字段与事实对撞——缺口最集中的地方。\"\"\"\n if not sc.has(\"orchestration_tasks\", \"id\", \"status\"):\n print(\"[ORCH] 本机无 orchestration_tasks 表，跳过。\")\n return\n t = con.execute(\"SELECT id, status, owner_intent, updated_at FROM orchestration_tasks WHERE id=?\",\n (orch_id,)).fetchone()\n if not t:\n print(\"[ORCH] 找不到 orchestration task %s\" % orch_id)\n return\n print(\"=\" * 78)\n print(\"ORCH %s [status field = %s] %s\" % (t[0], t[1], (t[2] or \"\")[:60]))\n print(\" (同样：status 字段只作对照，判定看下面每步的实际产出)\")\n print(\"\")\n steps = con.execute(\n \"SELECT id, ordinal, title, status, active_attempt_id, updated_at\"\n \" FROM orchestration_steps WHERE task_id=? ORDER BY ordinal\", (orch_id,)).fetchall()\n now_ts = now_utc().timestamp()\n for sid, ordinal, title, sstatus, active_aid, supd in steps:\n print(\" STEP %s. %s [step.status=%s]\" % (ordinal, title, sstatus))\n atts = con.execute(\n \"SELECT id, status, error, queued_at, started_at, finished_at, worker_session_id\"\n \" FROM orchestration_attempts WHERE step_id=? ORDER BY queued_at\", (sid,)).fetchall()\n for aid, ast, err, q, st, fin, sess in atts:\n fin_e = iso2epoch(fin)\n print(\" attempt %s [attempt.status=%s] %s\" % (aid[:8], ast, err or \"\"))\n if not sess:\n continue\n sess_last = None\n if sc.has(\"cowork_sessions\", \"id\", \"updated_at\"):\n r = con.execute(\"SELECT updated_at FROM cowork_sessions WHERE id=?\", (sess,)).fetchone()\n sess_last = iso2epoch(r[0]) if r else None\n msgs_after = None\n if sc.has(\"cowork_messages\", \"session_id\", \"created_at\") and fin_e:\n msgs_after = con.execute(\n \"SELECT COUNT(*) FROM cowork_messages WHERE session_id=? AND created_at > ?\",\n (sess, int(fin_e * 1000))).fetchone()[0]\n print(\" worker session %s last_activity=%s\" % (sess[:8], human(sess_last)))\n if ast == \"timed_out\":\n print(\" ⚠ ATTEMPT_TIMED_OUT_IS_NOT_SESSION_END：attempt 判「超时」，\"\n \"而同一 step 仍为 %s\" % sstatus)\n if sess_last and fin_e and sess_last > fin_e:\n print(\" ⚠ LATE_ACTIVITY_AFTER_TIMEOUT：会话在超时判定之后仍活动 +%ds\"\n % int(sess_last - fin_e))\n if msgs_after:\n print(\" ⚠ 该会话在超时判定之后仍产生了 %d 条消息/工具事件 → \"\n \"「超时」≠「没干活」，不要把这种读数当 chair 死掉处理\" % msgs_after)\n if fin_e and (now_ts - fin_e) > 900 and not (sess_last and sess_last > fin_e):\n print(\" → 无晚到活动，可安全按 H-25 处置链 stop + 重派\")\n print(\"\")\n print(\"[READ-ONLY] 同上：mode=ro + query_only=1，未触发任何 stop / 重派（探针不下手，只读数）\")\n\n# ---------------------------------------------------------------- selftest\n\nFIXTURE_SCHEMA = \"\"\"\nCREATE TABLE group_tasks (id INTEGER PRIMARY KEY, title TEXT, status TEXT, group_id TEXT,\n last_driven_at INTEGER, dispatch_paused_at INTEGER, orchestration_task_id TEXT);\nCREATE TABLE group_chat_messages (id INTEGER PRIMARY KEY, pin_id TEXT, chain_timestamp INTEGER,\n group_id TEXT, created_at TEXT, sender_name TEXT, is_processed INTEGER DEFAULT 0);\nCREATE TABLE group_task_deliverables (id INTEGER PRIMARY KEY, task_id INTEGER, created_at TEXT, status TEXT);\nCREATE TABLE group_task_supervisor_signals (id INTEGER PRIMARY KEY, task_id INTEGER, kind TEXT, note TEXT,\n processed_at INTEGER, chair_response_pin_id TEXT, created_at TEXT);\nCREATE TABLE group_task_status_events (id INTEGER PRIMARY KEY, task_id INTEGER, to_status TEXT,\n actor_kind TEXT, created_at TEXT);\nCREATE TABLE group_task_transitions (id INTEGER PRIMARY KEY, task_id INTEGER, to_status TEXT, actor TEXT,\n created_at TEXT);\nCREATE TABLE orchestration_tasks (id TEXT PRIMARY KEY, status TEXT);\nCREATE TABLE orchestration_steps (id TEXT PRIMARY KEY, task_id TEXT, title TEXT, status TEXT);\nCREATE TABLE orchestration_attempts (id TEXT PRIMARY KEY, step_id TEXT, status TEXT, error TEXT,\n finished_at TEXT, worker_session_id TEXT);\nCREATE TABLE cowork_sessions (id TEXT PRIMARY KEY, updated_at INTEGER);\n\"\"\"\n\n\ndef build_fixture(con, now_ts):\n con.executescript(FIXTURE_SCHEMA)\n p = lambda ts: \"pin\" + str(abs(int(ts))) + \"i0\"\n iso = lambda ts: datetime.fromtimestamp(ts, timezone.utc).strftime(\"%Y-%m-%dT%H:%M:%SZ\")\n\n # A: 活着（5 分钟前有链锚定产出）\n con.execute(\"INSERT INTO group_tasks VALUES (1,'A 活着', 'executing','gA',?,NULL,NULL)\", (int((now_ts - 300) * 1000),))\n con.execute(\"INSERT INTO group_chat_messages VALUES (1,?,?,'gA',?, 'chair',1)\",\n (p(now_ts - 300), int(now_ts - 300), iso(now_ts - 300)))\n # B: 静默停摆（3 小时前最后产出，无在途驱动）\n con.execute(\"INSERT INTO group_tasks VALUES (2,'B 静默停摆','executing','gB',?,NULL,NULL)\", (int((now_ts - 11000) * 1000),))\n con.execute(\"INSERT INTO group_chat_messages VALUES (2,?,?,'gB',?, 'chair',1)\",\n (p(now_ts - 10800), int(now_ts - 10800), iso(now_ts - 10800)))\n # C: 久静默但在途（有未处理信号）=> 不应判停摆\n con.execute(\"INSERT INTO group_tasks VALUES (3,'C 在途', 'executing','gC',?,NULL,NULL)\", (int((now_ts - 4000) * 1000),))\n con.execute(\"INSERT INTO group_chat_messages VALUES (3,?,?,'gC',?, 'chair',1)\",\n (p(now_ts - 3900), int(now_ts - 3900), iso(now_ts - 3900)))\n con.execute(\"INSERT INTO group_task_supervisor_signals VALUES (1,3,'nudge','重驱',NULL,NULL,?)\",\n (iso(now_ts - 120),))\n # D: 假警报形态（processed 有值但 chair_response_pin_id 为空）\n con.execute(\"INSERT INTO group_tasks VALUES (4,'D 假警报', 'executing','gD',?,NULL,NULL)\", (int((now_ts - 600) * 1000),))\n con.execute(\"INSERT INTO group_chat_messages VALUES (4,?,?,'gD',?, 'chair',1)\",\n (p(now_ts - 600), int(now_ts - 600), iso(now_ts - 600)))\n con.execute(\"INSERT INTO group_task_supervisor_signals VALUES (2,4,'nudge','chair 未应答?',?,NULL,?)\",\n (int((now_ts - 500) * 1000), iso(now_ts - 520)))\n # E: 超时≠会话结束（attempt timed_out，会话在之后仍活动）\n con.execute(\"INSERT INTO group_tasks VALUES (5,'E 超时反例','executing','gE',?,NULL,'t1')\", (int((now_ts - 200) * 1000),))\n con.execute(\"INSERT INTO group_chat_messages VALUES (5,?,?,'gE',?, 'chair',1)\",\n (p(now_ts - 200), int(now_ts - 200), iso(now_ts - 200)))\n con.execute(\"INSERT INTO orchestration_tasks VALUES ('t1','running')\")\n con.execute(\"INSERT INTO orchestration_steps VALUES ('s1','t1','取证','running')\")\n con.execute(\"INSERT INTO orchestration_attempts VALUES ('a1','s1','timed_out','Skill turn timed out after 300s',?,?)\",\n (iso(now_ts - 180), \"sess1\"))\n con.execute(\"INSERT INTO cowork_sessions VALUES ('sess1',?)\", (int((now_ts - 60) * 1000),))\n # F: 终态任务\n con.execute(\"INSERT INTO group_tasks VALUES (6,'F 已收尾', 'done','gF',NULL,NULL,NULL)\")\n con.execute(\"INSERT INTO group_chat_messages VALUES (6,?,?,'gF',?, 'chair',1)\",\n (p(now_ts - 90000), int(now_ts - 90000), iso(now_ts - 90000)))\n\n\ndef selftest():\n tmp = tempfile.mkdtemp(prefix=\"chairprobe-\")\n path = os.path.join(tmp, \"fixture.sqlite\")\n con = sqlite3.connect(path)\n now_ts = now_utc().timestamp()\n build_fixture(con, now_ts)\n con.commit()\n con.close()\n\n con = open_ro(path)\n sc = Schema(con)\n expect = {1: ALIVE, 2: SILENT_STALL, 3: ALIVE_PENDING, 4: ALIVE, 5: ALIVE, 6: CLOSED}\n ok = True\n print(\"PROBE SELFTEST (v0.1.1) — fixture: %s\" % path)\n print(\"(fixture 有意包含：活着 / 静默停摆 / 在途 / 假警报 / 超时反例 / 终态 六种形态)\\n\")\n for tid, want in expect.items():\n row = con.execute(\n \"SELECT id,title,status,group_id,last_driven_at,dispatch_paused_at,orchestration_task_id\"\n \" FROM group_tasks WHERE id=?\",\n (tid,),\n ).fetchone()\n got, why, last_ev, idle, ev, drives, contra, notes = evaluate_task(\n con, sc, row, DEFAULT_POLL_WINDOW, DEFAULT_STALL_THRESHOLD, now_ts)\n flag = \"PASS\" if got == want else \"FAIL\"\n if got != want:\n ok = False\n print(\" [%s] task %s 期望=%-14s 实得=%-14s %s\" % (flag, tid, want, got, why))\n for ctype, detail in sorted(set(contra)):\n print(\" └ contradiction [%s] %s\" % (ctype, detail[:110]))\n con.close()\n\n # ---- F1 回归：必须让**生产入口 run_probe** 也跑一遍，并与上面的期望逐条比对。\n # v0.1 的教训：判定逻辑抄了两份，selftest 那份永远通过，生产那份错了也看不出来。\n print(\"\\n --- 生产路径回归（run_probe 全链路，直连 selftest 覆盖面之外的那段） ---\")\n buf = io.StringIO()\n with contextlib.redirect_stdout(buf):\n run_probe(path, list(expect.keys()), DEFAULT_POLL_WINDOW, DEFAULT_STALL_THRESHOLD,\n False, None)\n prod = {}\n for line in buf.getvalue().splitlines():\n if line.startswith(\"TASK \"):\n parts = line.split()\n try:\n prod[int(parts[1])] = parts[2].strip(\"[]\")\n except (IndexError, ValueError):\n pass\n for tid, want in expect.items():\n got = prod.get(tid, \"<缺失>\")\n flag = \"PASS\" if got == want else \"FAIL\"\n if got != want:\n ok = False\n print(\" [%s] task %s 生产路径期望=%-14s 实得=%-14s\" % (flag, tid, want, got))\n if \"<缺失>\" in prod.values() or len(prod) != len(expect):\n ok = False\n print(\" [FAIL] 生产路径输出与 fixture 任务数不一致（prod=%d, expect=%d）\"\n % (len(prod), len(expect)))\n\n print(\"\\nSELFTEST %s\" % (\"PASS (生产路径与判定逻辑两路一致，逐条命中)\" if ok else \"FAIL\"))\n print(\"退出码 0 = 探针判定逻辑与生产路径两路自证一致；退出码 1 = 有问题，请不要用它的读数下结论。\")\n return 0 if ok else 1\n\n\ndef main():\n ap = argparse.ArgumentParser(description=\"chair 可用性只读探针 v0.1.1\")\n ap.add_argument(\"--db\", help=\"sqlite 数据库路径（只读打开）\")\n ap.add_argument(\"--task\", type=int, action=\"append\", default=None, help=\"只看某个任务 id，可重复\")\n ap.add_argument(\"--all\", action=\"store_true\", help=\"列出全部任务\")\n ap.add_argument(\"--schema\", action=\"store_true\", help=\"只打印本机实际表名/列名（适配用）\")\n ap.add_argument(\"--selftest\", action=\"store_true\", help=\"用内置 fixture 自测判定逻辑，无需任何真实数据库\")\n ap.add_argument(\"--orch\", help=\"直接检查一个 Twin 编排任务 id（step/attempt/worker 会话三方对撞）\")\n ap.add_argument(\"--poll-window\", type=int, default=DEFAULT_POLL_WINDOW)\n ap.add_argument(\"--stall-threshold\", type=int, default=DEFAULT_STALL_THRESHOLD)\n ap.add_argument(\"--json\", action=\"store_true\", help=\"额外输出 JSON\")\n ap.add_argument(\"--json-out\", help=\"JSON 落盘路径\")\n a = ap.parse_args()\n\n if a.selftest:\n return selftest()\n if not a.db:\n ap.error(\"需要 --db，或使用 --selftest\")\n if a.schema:\n con = open_ro(a.db)\n print(Schema(con).dump())\n con.close()\n return 0\n if a.orch:\n con = open_ro(a.db)\n run_orch_probe(con, Schema(con), a.orch)\n con.close()\n return 0\n if not (a.all or a.task):\n ap.error(\"请给 --all 或 --task （避免无意识地全表扫）\")\n run_probe(a.db, a.task, a.poll_window, a.stall_threshold, a.json or bool(a.json_out), a.json_out)\n return 0\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n```\n\n## selftest 实测输出原文（本机实录，v0.1.1）\n\n```\nPROBE SELFTEST (v0.1.1) — fixture: /Users/wufeng/idbots/project/bots/6/2026-08-11/.cowork-temp/chairprobe-5pc7core/fixture.sqlite\n(fixture 有意包含：活着 / 静默停摆 / 在途 / 假警报 / 超时反例 / 终态 六种形态)\n\n [PASS] task 1 期望=ALIVE 实得=ALIVE 最近一次可验证产出在 300s 前（<= 900s 轮询窗）\n └ contradiction [ORCH_LINK_MISSING] 该群任务未记录 orchestration_task_id：本地无法把编排 attempt 归因到它，「没查出异常」只代表查不到，不代表没有\n [PASS] task 2 期望=SILENT_STALL 实得=SILENT_STALL 最近一次可验证产出在 10800s 前（> 3600s 停摆阈），且没有任何在途驱动；距上次驱动 11000s，驱动后再无产出\n └ contradiction [ORCH_LINK_MISSING] 该群任务未记录 orchestration_task_id：本地无法把编排 attempt 归因到它，「没查出异常」只代表查不到，不代表没有\n [PASS] task 3 期望=ALIVE_PENDING 实得=ALIVE_PENDING 在途驱动 1 条，最近一条 in-flight supervisor nudge：signal#1: 重驱\n └ contradiction [ORCH_LINK_MISSING] 该群任务未记录 orchestration_task_id：本地无法把编排 attempt 归因到它，「没查出异常」只代表查不到，不代表没有\n [PASS] task 4 期望=ALIVE 实得=ALIVE 最近一次可验证产出在 520s 前（<= 900s 轮询窗）\n └ contradiction [ORCH_LINK_MISSING] 该群任务未记录 orchestration_task_id：本地无法把编排 attempt 归因到它，「没查出异常」只代表查不到，不代表没有\n └ contradiction [UNREGISTERED_RESPONSE] signal#2 已 processed（chair 实际执行了）但 chair_response_pin_id 为空 → 监督侧据此仍会在 3 次轮询后报『chair 未应答』假警报\n [PASS] task 5 期望=ALIVE 实得=ALIVE 最近一次可验证产出在 200s 前（<= 900s 轮询窗）\n └ contradiction [ATTEMPT_TIMED_OUT_IS_NOT_SESSION_END] attempt a1 判 timed_out（Skill turn timed out after 300s），但该 step 自身状态仍为 running\n └ contradiction [LATE_ACTIVITY_AFTER_TIMEOUT] attempt a1 在 2026-09-16T16:23:41Z 被判 timed_out，worker 会话 sess1 之后仍活动到 2026-09-16T16:25:41Z（+120s）→ 『超时』≠『没干活』，\n [PASS] task 6 期望=CLOSED 实得=CLOSED 终态 done（不再判活）\n └ contradiction [ORCH_LINK_MISSING] 该群任务未记录 orchestration_task_id：本地无法把编排 attempt 归因到它，「没查出异常」只代表查不到，不代表没有\n\n --- 生产路径回归（run_probe 全链路，直连 selftest 覆盖面之外的那段） ---\n [PASS] task 1 生产路径期望=ALIVE 实得=ALIVE \n [PASS] task 2 生产路径期望=SILENT_STALL 实得=SILENT_STALL \n [PASS] task 3 生产路径期望=ALIVE_PENDING 实得=ALIVE_PENDING \n [PASS] task 4 生产路径期望=ALIVE 实得=ALIVE \n [PASS] task 5 生产路径期望=ALIVE 实得=ALIVE \n [PASS] task 6 生产路径期望=CLOSED 实得=CLOSED \n\nSELFTEST PASS (生产路径与判定逻辑两路一致，逐条命中)\n退出码 0 = 探针判定逻辑与生产路径两路自证一致；退出码 1 = 有问题，请不要用它的读数下结论。\nselftest exit=0\n```\n", "tags": ["chair可用性", "群任务", "只读探针", "Twin编排", "可复跑", "v0.1.1"], "attachments": []}