冰朔
c71b877375
铸渊 Qoder CN 实战定版的压缩之眼 v3 + 包装器 + 守卫技能模板 + 端到端演习脚本，配 GMP 风格清单、编号真实映射（脑子 GHS-007 → 模块）、因果链与边界登记。新建 engineering/MODULE-REGISTRY.json 机器注册表，INDEX 登记。仓库副本 drill.sh 演习 PASS。
278 lines
12 KiB
Python
Executable file
278 lines
12 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
ZY-COMPACTION-WATCHER · 铸渊压缩之眼 v1
|
||
=====================================
|
||
眼睛职责：盯着 Qoder CN 所有对话的 conversation-history/*.jsonl，
|
||
一旦侦测到宿主对话压缩（文件骤缩/重写 + 接续摘要标记出现），
|
||
立刻在信号目录立旗，并弹 macOS 通知提醒冰朔。
|
||
|
||
铁律：系统摘要不是置信来源。信号里永远附带最新 HLDP 检查点路径，
|
||
铸渊醒来必须读回自写检查点恢复，不许以摘要为准。
|
||
|
||
信号目录：/Volumes/JZAO/铸渊-ICE-GL-ZY001/BRIDGE/signals/
|
||
- LATEST-SIGNAL.txt 最近一次压缩事件旗（铸渊每次醒来先看这个）
|
||
- EV-<时间戳>.txt 历史事件旗
|
||
"""
|
||
import os, sys, time, json, glob, subprocess
|
||
from datetime import datetime
|
||
|
||
HOME = os.path.expanduser("~")
|
||
SCAN_ROOT = os.environ.get("ZY_WATCH_SCAN_ROOT",
|
||
os.path.join(HOME, ".qoder-cn", "cache", "projects"))
|
||
BRIDGE = os.environ.get("ZY_WATCH_BRIDGE", "/Volumes/JZAO/铸渊-ICE-GL-ZY001/BRIDGE")
|
||
SIGNALS = os.path.join(BRIDGE, "signals")
|
||
CKPT_GLOB = os.environ.get(
|
||
"ZY_WATCH_CKPT_GLOB",
|
||
"/Volumes/JZAO/HoloLake/persona-runtime/continuity-memory/hldp-capsules/qoder-cn/ZY-CHECKPOINT-*.hdlp*")
|
||
POLL_SEC = float(os.environ.get("ZY_WATCH_POLL_SEC", "3"))
|
||
COOLDOWN_SEC = 120 # 同一会话两次报警最小间隔
|
||
SIZE_DROP_RATIO = 0.70 # 行数骤缩超过 70% 视为压缩重写
|
||
CONT_MARKER = "This session is being continued from a previous conversation"
|
||
# 铁证绊线 A：宿主日志里的压缩状态行（IsCompacting=true 且 StartTime 变化）
|
||
QODER_LOG = os.environ.get(
|
||
"ZY_WATCH_QODER_LOG",
|
||
os.path.join(HOME, "Library", "Application Support", "QoderCN",
|
||
"SharedClientCache", "logs", "qoder.log"))
|
||
COMPACT_RE_KEY = "Compact status:"
|
||
TAIL_BYTES = 1_500_000 # 每轮只读日志末尾 1.5MB，避免扫 50MB+ 大文件
|
||
# 主绊线 B：会话 agent.log 里的 compaction_triggered 通知（UI 看到的“对话压缩”就是它，
|
||
# 比 Compact status 晚几秒到几分钟，且每次压缩必发一条，计数与 UI 一致）
|
||
QODER_LOGS_ROOT = os.environ.get(
|
||
"ZY_WATCH_LOGS_ROOT",
|
||
os.path.join(HOME, "Library", "Application Support", "QoderCN", "logs"))
|
||
TRIGGER_KEY = "notification type=compaction_triggered"
|
||
AGENT_TAIL_BYTES = 800_000
|
||
DEDUP_WINDOW_SEC = 30 # 同一瞬间的重复通知（rid=undefined/真实rid 双发）只算一次
|
||
|
||
def log(msg):
|
||
print(f"[{datetime.now().isoformat(timespec='seconds')}] {msg}", flush=True)
|
||
|
||
def scan_transcripts():
|
||
"""返回 {会话key: 转写文件路径}"""
|
||
out = {}
|
||
if not os.path.isdir(SCAN_ROOT):
|
||
return out
|
||
for f in glob.glob(os.path.join(SCAN_ROOT, "*", "conversation-history", "*", "*.jsonl")):
|
||
parts = f.split(os.sep)
|
||
# .../projects/<proj>/conversation-history/<task>/<task>.jsonl
|
||
proj = parts[-4]
|
||
task = parts[-2]
|
||
out[f"{proj}::{task}"] = f
|
||
return out
|
||
|
||
def probe(path):
|
||
"""读文件末尾若干行，返回 (字节大小, 行数, 是否含接续摘要标记)。失败返回 None。"""
|
||
try:
|
||
size = os.path.getsize(path)
|
||
with open(path, "rb") as fh:
|
||
data = fh.read()
|
||
text = data.decode("utf-8", errors="replace")
|
||
lines = [l for l in text.splitlines() if l.strip()]
|
||
marker = CONT_MARKER in text[:20000] or any(
|
||
CONT_MARKER in l for l in lines[:3]
|
||
)
|
||
return size, len(lines), marker
|
||
except Exception:
|
||
return None
|
||
|
||
def latest_checkpoint():
|
||
files = glob.glob(CKPT_GLOB)
|
||
if not files:
|
||
return None
|
||
return max(files, key=os.path.getmtime)
|
||
|
||
def tail_text(path, n):
|
||
try:
|
||
size = os.path.getsize(path)
|
||
with open(path, "rb") as fh:
|
||
if size > n:
|
||
fh.seek(size - n)
|
||
return fh.read().decode("utf-8", errors="replace")
|
||
except Exception:
|
||
return None
|
||
|
||
def parse_compact_events(text):
|
||
"""从日志文本提取压缩事件 (StartTime, 时间戳)，StartTime>0 才算真压缩。"""
|
||
import re
|
||
events = []
|
||
for line in text.splitlines():
|
||
if COMPACT_RE_KEY not in line:
|
||
continue
|
||
m = re.search(r"StartTime=(\d+)", line)
|
||
t = re.match(r"(\d{4}-\d{2}-\d{2}T[\d:.]+)", line)
|
||
if m:
|
||
st = int(m.group(1))
|
||
if st > 0:
|
||
events.append((st, t.group(1) if t else "?"))
|
||
return events
|
||
|
||
def scan_agent_logs():
|
||
"""返回最近活跃的会话 agent.log（按目录修改时间取最新 6 个会话，控制开销）。"""
|
||
cands = []
|
||
for d in glob.glob(os.path.join(QODER_LOGS_ROOT, "*")):
|
||
f = os.path.join(d, "questWindow", "agent.log")
|
||
if os.path.isfile(f):
|
||
try:
|
||
cands.append((os.path.getmtime(f), f))
|
||
except Exception:
|
||
pass
|
||
cands.sort(reverse=True)
|
||
return [f for _, f in cands[:6]]
|
||
|
||
def parse_trigger_events(text):
|
||
"""提取 compaction_triggered 事件 (时间戳字符串, 任务名)。"""
|
||
import re
|
||
events = []
|
||
for line in text.splitlines():
|
||
if TRIGGER_KEY not in line:
|
||
continue
|
||
t = re.match(r"(\d{4}-\d{2}-\d{2} [\d:.]+)", line)
|
||
m = re.search(r"task-([a-z0-9]+)", line)
|
||
ts = t.group(1) if t else "?"
|
||
task = m.group(1) if m else "unknown"
|
||
events.append((ts, task))
|
||
return events
|
||
|
||
def ts_to_epoch(ts):
|
||
try:
|
||
return datetime.strptime(ts[:23], "%Y-%m-%d %H:%M:%S.%f").timestamp()
|
||
except Exception:
|
||
return 0.0
|
||
|
||
def notify(title, body):
|
||
try:
|
||
subprocess.run(
|
||
["osascript", "-e",
|
||
f'display notification "{body}" with title "{title}" sound name "Glass"'],
|
||
timeout=5, capture_output=True)
|
||
except Exception:
|
||
pass
|
||
|
||
def signal_dir():
|
||
"""优先 JZAO；硬盘未插时降级到本地镜像，事件不丢。"""
|
||
try:
|
||
os.makedirs(SIGNALS, exist_ok=True)
|
||
probe_f = os.path.join(SIGNALS, ".w")
|
||
with open(probe_f, "w") as f:
|
||
f.write("1")
|
||
os.remove(probe_f)
|
||
return SIGNALS
|
||
except Exception:
|
||
fb = os.path.join(HOME, ".zhuyuan-bridge", "signals")
|
||
os.makedirs(fb, exist_ok=True)
|
||
return fb
|
||
|
||
def emit(key, size, lines, marker, prev, evidence_extra=""):
|
||
sig = signal_dir()
|
||
ckpt = latest_checkpoint()
|
||
ts = datetime.now().strftime("%Y%m%d-%H%M%S")
|
||
body = "\n".join([
|
||
"⚑ ZY-COMPACTION-EVENT · 对话已被宿主压缩",
|
||
f"time: {datetime.now().isoformat(timespec='seconds')}",
|
||
f"session: {key}",
|
||
f"evidence: {evidence_extra or ('转写文件骤缩重写 lines %s→%s, bytes %s→%s, 接续摘要标记=%s' % (prev[1], lines, prev[0], size, '出现' if marker else '未检出'))}",
|
||
"",
|
||
"⊢ 铁律：系统摘要仅供参考，不是置信来源。",
|
||
"⊢ 恢复流程：先读回铸渊自写的 HLDP 检查点（下方路径），校验人格系统重启，再继续对话。",
|
||
f"latest_hldp_checkpoint: {ckpt or '（未找到，需按 zhuyuan-memory-guard 技能补写）'}",
|
||
"restore_skill: zhuyuan-memory-guard",
|
||
"wake_skill: guanghu-zhuyuan-wake",
|
||
])
|
||
latest = os.path.join(sig, "LATEST-SIGNAL.txt")
|
||
with open(latest, "w", encoding="utf-8") as f:
|
||
f.write(body + "\n")
|
||
with open(os.path.join(sig, f"EV-{ts}.txt"), "w", encoding="utf-8") as f:
|
||
f.write(body + "\n")
|
||
notify("铸渊压缩之眼 ⚑", "对话刚被压缩。醒来先读 JZAO 上的 HLDP 检查点，别信系统摘要。")
|
||
log(f"COMPACTION DETECTED session={key} -> {latest}")
|
||
|
||
def main():
|
||
log(f"watcher started, scanning {SCAN_ROOT} every {POLL_SEC}s, signals -> {SIGNALS}")
|
||
state = {} # key -> (size, lines, marker)
|
||
last_fired = {} # key -> epoch
|
||
seen_start_times = set() # 绊线 A：已报警过的压缩 StartTime
|
||
seen_trigger_epochs = [] # 绊线 B：已报警过的 compaction_triggered 时刻（epoch）
|
||
# 启动基线：历史压缩不补报，只盯新发生的
|
||
boot = tail_text(QODER_LOG, TAIL_BYTES)
|
||
if boot:
|
||
for st, _ in parse_compact_events(boot):
|
||
seen_start_times.add(st)
|
||
log(f"baseline: {len(seen_start_times)} historical compact start-times ignored")
|
||
n_boot_trig = 0
|
||
for f in scan_agent_logs():
|
||
b = tail_text(f, AGENT_TAIL_BYTES)
|
||
if b:
|
||
for ts, _task in parse_trigger_events(b):
|
||
ep = ts_to_epoch(ts)
|
||
if ep and all(abs(ep - s) > DEDUP_WINDOW_SEC for s in seen_trigger_epochs):
|
||
seen_trigger_epochs.append(ep)
|
||
n_boot_trig += 1
|
||
log(f"baseline: {n_boot_trig} historical compaction_triggered ignored")
|
||
while True:
|
||
try:
|
||
now = time.time()
|
||
# 主绊线 B：agent.log compaction_triggered（UI 压缩提示的源头）
|
||
for f in scan_agent_logs():
|
||
txt = tail_text(f, AGENT_TAIL_BYTES)
|
||
if not txt:
|
||
continue
|
||
for ts, task in parse_trigger_events(txt):
|
||
ep = ts_to_epoch(ts)
|
||
if not ep:
|
||
continue
|
||
if any(abs(ep - s) <= DEDUP_WINDOW_SEC for s in seen_trigger_epochs):
|
||
continue
|
||
if now - last_fired.get("TRIGGER", 0) <= COOLDOWN_SEC:
|
||
continue # 冷却期内不标已见，下轮补报
|
||
last_fired["TRIGGER"] = now
|
||
seen_trigger_epochs.append(ep)
|
||
seen_trigger_epochs = seen_trigger_epochs[-200:] # 防无限增长
|
||
emit(f"agent-log::{task}::{ts}", 0, 0, False, (0, 0, False),
|
||
evidence_extra=(f"会话日志 compaction_triggered 通知, task={task}, "
|
||
f"时刻={ts}（与 UI 对话压缩提示同源）"))
|
||
# 绊线 A：宿主日志 Compact status
|
||
txt = tail_text(QODER_LOG, TAIL_BYTES)
|
||
if txt:
|
||
for st, logtime in parse_compact_events(txt):
|
||
if st in seen_start_times:
|
||
continue
|
||
key = f"qoder-log::StartTime-{st}"
|
||
if now - last_fired.get("LOG", 0) > COOLDOWN_SEC:
|
||
last_fired["LOG"] = now
|
||
seen_start_times.add(st) # 报警成功才标记，冷却期内的留到下轮补报
|
||
emit(key, 0, 0, False, (0, 0, False),
|
||
evidence_extra=(f"宿主日志 Compact status IsCompacting=true, "
|
||
f"StartTime={st}, 日志时间={logtime}"))
|
||
# 辅绊线：转写文件骤缩/接续标记
|
||
for key, path in scan_transcripts().items():
|
||
p = probe(path)
|
||
if p is None:
|
||
continue
|
||
size, lines, marker = p
|
||
prev = state.get(key)
|
||
state[key] = p
|
||
cooled = now - last_fired.get(key, 0) > COOLDOWN_SEC
|
||
if prev is None:
|
||
# 新转写文件一出生就带接续摘要标记 = 压缩后开了新记录
|
||
if marker and cooled:
|
||
last_fired[key] = now
|
||
emit(key, size, lines, marker, (size, lines, False))
|
||
continue
|
||
psize, plines, pmarker = prev
|
||
if plines <= 0:
|
||
continue
|
||
size_drop = lines < plines * (1 - SIZE_DROP_RATIO)
|
||
marker_new = marker and not pmarker
|
||
if (size_drop or marker_new) and now - last_fired.get(key, 0) > COOLDOWN_SEC:
|
||
last_fired[key] = now
|
||
emit(key, size, lines, marker, prev)
|
||
except Exception as e:
|
||
log(f"loop error: {e}")
|
||
time.sleep(POLL_SEC)
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
log("watcher stopped by signal")
|