handoff_snapshot.py 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190
  1. #!/usr/bin/env python3
  2. # -*- coding: utf-8 -*-
  3. """PreCompact hook (GLOBAL): 上下文压缩前写 ≤2KB 交接快照到项目记忆根。
  4. 落点:{PROJECT_ROOT}/.ai-memory/_handoff_snapshot.md
  5. (PROJECT_ROOT 按 project-memory-management「路径根目录探测协议」解析,绝
  6. 不写死 .codebuddy 等其它 IDE 约定路径;记忆跟随项目,跨 IDE 通用。)
  7. hook 拿不到对话内容,故快照基于文件系统派生(压缩后最实用的重定位信息):
  8. - 近 4 小时内被改动的项目文件(排除备份/依赖/缓存/记忆目录),按时间倒序取前 25
  9. - 当前 Plan/*.md 的状态行(## 状态)
  10. - 最新一份每日 memory 文件名
  11. """
  12. import sys
  13. import os
  14. import json
  15. import time
  16. CUTOFF_HOURS = 4
  17. MAX_FILES = 25
  18. EXCLUDE = {"backup", "vendor", "node_modules", "uploads",
  19. ".git", ".codebuddy", ".ai-memory", "ecachefiles",
  20. "__pycache__", "dist", "build", "cache", "tmp", "temp", "logs", "runtime"}
  21. # 排除备份/临时文件扩展名(避免 backup_on_write 的 .bak/.bak.1/.bak.2 污染快照)
  22. EXCLUDE_EXT = {".bak", ".tmp", ".log", ".cache", ".pyc", ".swp", ".swo"}
  23. def _find_git_root(start):
  24. """向上找最近的 .git 目录,返回其所在目录(仓库根),找不到返回 None。"""
  25. d = os.path.abspath(start)
  26. while True:
  27. if os.path.isdir(os.path.join(d, ".git")):
  28. return d
  29. nd = os.path.dirname(d)
  30. if nd == d:
  31. return None
  32. d = nd
  33. def resolve_project_root(root):
  34. """按技能「路径根目录探测协议」解析 PROJECT_ROOT:
  35. 1. 环境变量 PROJECT_ROOT(跨 IDE 通用,优先):绝对路径直接用,相对路径按 root 拼接。
  36. 2. Git 仓库根目录:向上找含 .git 的目录。
  37. 3. IDE 工作区根目录:仅当 cwd 自身已含 .ai-memory(确为项目记忆根)时才回退。
  38. 4. 以上均未命中:hook 无法询问用户,返回 None(由调用方跳过,严禁自行创建目录)。
  39. """
  40. env = os.environ.get("PROJECT_ROOT")
  41. if env:
  42. return env if os.path.isabs(env) else os.path.join(root, env)
  43. git_root = _find_git_root(root)
  44. if git_root:
  45. return git_root
  46. if os.path.isdir(os.path.join(root, ".ai-memory")):
  47. return root
  48. return None
  49. def main():
  50. try:
  51. sys.stdin.read()
  52. except Exception:
  53. pass
  54. cwd = os.getcwd()
  55. proj_root = resolve_project_root(cwd)
  56. if not proj_root:
  57. # 项目根不可解析(如运行在 skill 目录、或记忆在数据库的运行时不落文件系统):跳过落盘
  58. return 0
  59. mem_dir = os.path.join(proj_root, ".ai-memory")
  60. root = proj_root # 所有扫描与相对路径均以解析出的项目根为基准
  61. cutoff = time.time() - CUTOFF_HOURS * 3600
  62. recent = []
  63. for dirpath, dirnames, filenames in os.walk(root):
  64. dirnames[:] = [d for d in dirnames
  65. if d.lower() not in EXCLUDE and not d.startswith(".")]
  66. for fn in filenames:
  67. # 过滤备份/临时文件扩展名
  68. ext = os.path.splitext(fn)[1].lower()
  69. if ext in EXCLUDE_EXT:
  70. continue
  71. fp = os.path.join(dirpath, fn)
  72. try:
  73. mtime = os.path.getmtime(fp)
  74. except OSError:
  75. continue
  76. if mtime >= cutoff:
  77. recent.append((mtime, fp))
  78. recent.sort(reverse=True)
  79. recent = recent[:MAX_FILES]
  80. def rel(p):
  81. return os.path.relpath(p, root).replace("\\", "/")
  82. lines = ["# 压缩交接快照 (PreCompact 自动生成)", ""]
  83. lines.append("生成时间: " + time.strftime("%Y-%m-%d %H:%M:%S"))
  84. lines.append("工作区: " + root)
  85. lines.append("")
  86. # 输出顺序:Plan 状态 → 最新记忆 → 改动文件列表(最重要的放前面,避免截断丢失)
  87. # Plan 状态(兼容 Plan/plan 目录名大小写)
  88. lines.append("## 进行中计划 (Plan/*.md 状态)")
  89. plan_dir = None
  90. for name in ("Plan", "plan"):
  91. candidate = os.path.join(root, name)
  92. if os.path.isdir(candidate):
  93. plan_dir = candidate
  94. break
  95. if plan_dir:
  96. for fn in sorted(os.listdir(plan_dir)):
  97. if not fn.endswith("_plan.md"):
  98. continue
  99. p = os.path.join(plan_dir, fn)
  100. status = ""
  101. try:
  102. with open(p, encoding="utf-8", errors="ignore") as f:
  103. for line in f:
  104. if line.strip().startswith("## 状态"):
  105. status = line.strip()
  106. break
  107. except OSError:
  108. pass
  109. lines.append("- %s %s" % (fn, status))
  110. else:
  111. lines.append("- (无 Plan 目录)")
  112. lines.append("")
  113. # 最新 memory(按 mtime 排序,避免非日期格式文件名排序不准)
  114. lines.append("## 最新记忆")
  115. if os.path.isdir(mem_dir):
  116. mds = []
  117. for f in os.listdir(mem_dir):
  118. if not f.endswith(".md"):
  119. continue
  120. fp = os.path.join(mem_dir, f)
  121. try:
  122. mds.append((os.path.getmtime(fp), f))
  123. except OSError:
  124. continue
  125. mds.sort(reverse=True) # 按 mtime 降序,最新的在前
  126. rel_mem = os.path.relpath(mem_dir, root).replace("\\", "/")
  127. if mds:
  128. lines.append("- 最新: %s/%s" % (rel_mem, mds[0][1]))
  129. else:
  130. lines.append("- (无)")
  131. else:
  132. lines.append("- (无 memory 目录)")
  133. lines.append("")
  134. # 改动文件列表(放最后,截断时只丢失文件列表,不丢失 Plan/记忆)
  135. lines.append("## 近 %d 小时改动文件 (前 %d)" % (CUTOFF_HOURS, MAX_FILES))
  136. if recent:
  137. for mtime, fp in recent:
  138. lines.append("- %s (%s)" % (rel(fp),
  139. time.strftime("%m-%d %H:%M", time.localtime(mtime))))
  140. else:
  141. lines.append("- (无)")
  142. out = "\n".join(lines) + "\n"
  143. b = out.encode("utf-8")
  144. if len(b) > 2048:
  145. # 按字节截断并在字符边界安全解码,避免切断多字节中文
  146. b = b[:2048]
  147. out = b.decode("utf-8", errors="ignore")
  148. try:
  149. os.makedirs(mem_dir, exist_ok=True)
  150. snap = os.path.join(mem_dir, "_handoff_snapshot.md")
  151. tmp = snap + ".tmp"
  152. # 二进制写入:避免 Windows 文本模式把 \n 转 \r\n 撑大文件致超 2048
  153. with open(tmp, "wb") as f:
  154. f.write(out.encode("utf-8"))
  155. try:
  156. os.replace(tmp, snap) # 原子替换,避免目标被读锁占用时写失败
  157. except OSError:
  158. # 极少数情况目标被锁:退回直接覆盖写
  159. with open(snap, "wb") as f:
  160. f.write(out.encode("utf-8"))
  161. try:
  162. os.remove(tmp)
  163. except OSError:
  164. pass
  165. except OSError:
  166. return 0
  167. return 0
  168. if __name__ == "__main__":
  169. sys.exit(main())