651 lines
23 KiB
Python
651 lines
23 KiB
Python
#!/usr/bin/env python3
|
||
"""agent_on_stop — agentStop 合并 hook 脚本（v3：含 LLM 摘要生成）。
|
||
|
||
合并原 audit_reminder + change_compliance_prescan + build_audit_context + session_extract：
|
||
1. 全量会话记录提取 → docs/audit/session_logs/（无论是否有代码变更）
|
||
2. 为刚提取的 session 调用百炼 API 生成 description → 写入双索引
|
||
3. 扫描工作区 → 与 promptSubmit 基线对比 → 精确检测本次对话变更
|
||
4. 若无任何文件变更 → 跳过审查，静默退出
|
||
5. 合规预扫描 → .kiro/state/.compliance_state.json
|
||
6. 构建审计上下文 → .kiro/state/.audit_context.json
|
||
7. 审计提醒（15 分钟限频）→ stderr
|
||
|
||
变更检测基于文件 mtime+size 基线对比，不依赖 git commit 历史。
|
||
所有功能块用 try/except 隔离，单个失败不影响其他。
|
||
"""
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
from datetime import datetime, timezone, timedelta
|
||
|
||
# 同目录导入文件基线模块 + cwd 校验
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from file_baseline import scan_workspace, load_baseline, diff_baselines, total_changes
|
||
from _ensure_root import ensure_repo_root
|
||
|
||
TZ_TAIPEI = timezone(timedelta(hours=8))
|
||
MIN_INTERVAL = timedelta(minutes=15)
|
||
|
||
# 路径常量
|
||
STATE_PATH = os.path.join(".kiro", "state", ".audit_state.json")
|
||
COMPLIANCE_PATH = os.path.join(".kiro", "state", ".compliance_state.json")
|
||
CONTEXT_PATH = os.path.join(".kiro", "state", ".audit_context.json")
|
||
PROMPT_ID_PATH = os.path.join(".kiro", "state", ".last_prompt_id.json")
|
||
# 噪声路径（用于过滤变更列表中的非业务文件）
|
||
NOISE_PATTERNS = [
|
||
re.compile(r"^docs/audit/"),
|
||
re.compile(r"^\.kiro/"),
|
||
re.compile(r"^\.hypothesis/"),
|
||
re.compile(r"^tmp/"),
|
||
re.compile(r"\.png$"),
|
||
re.compile(r"\.jpg$"),
|
||
]
|
||
|
||
# 高风险路径
|
||
HIGH_RISK_PATTERNS = [
|
||
re.compile(r"^apps/etl/connectors/feiqiu/(api|cli|config|database|loaders|models|orchestration|scd|tasks|utils|quality)/"),
|
||
re.compile(r"^apps/backend/app/"),
|
||
re.compile(r"^apps/admin-web/src/"),
|
||
re.compile(r"^apps/miniprogram/"),
|
||
re.compile(r"^packages/shared/"),
|
||
re.compile(r"^db/"),
|
||
]
|
||
|
||
# 文档映射（合规检查用）
|
||
DOC_MAP = {
|
||
"apps/backend/app/routers/": ["apps/backend/docs/API-REFERENCE.md", "docs/contracts/openapi/backend-api.json"],
|
||
"apps/backend/app/services/": ["apps/backend/docs/API-REFERENCE.md", "apps/backend/README.md"],
|
||
"apps/backend/app/auth/": ["apps/backend/docs/API-REFERENCE.md", "apps/backend/README.md", "docs/contracts/openapi/backend-api.json"],
|
||
"apps/backend/app/schemas/": ["docs/contracts/openapi/backend-api.json"],
|
||
"apps/backend/app/main.py": ["docs/contracts/openapi/backend-api.json"],
|
||
"apps/etl/connectors/feiqiu/tasks/": ["apps/etl/connectors/feiqiu/docs/etl_tasks/"],
|
||
"apps/etl/connectors/feiqiu/loaders/": ["apps/etl/connectors/feiqiu/docs/etl_tasks/"],
|
||
"apps/etl/connectors/feiqiu/scd/": ["apps/etl/connectors/feiqiu/docs/business-rules/scd2_rules.md"],
|
||
"apps/etl/connectors/feiqiu/orchestration/": ["apps/etl/connectors/feiqiu/docs/architecture/"],
|
||
"apps/admin-web/src/": ["apps/admin-web/README.md"],
|
||
"apps/miniprogram/": ["apps/miniprogram/README.md"],
|
||
"packages/shared/": ["packages/shared/README.md"],
|
||
}
|
||
|
||
# 接口变更检测模式（routers / auth / schemas / main.py）
|
||
API_CHANGE_PATTERNS = [
|
||
re.compile(r"^apps/backend/app/routers/"),
|
||
re.compile(r"^apps/backend/app/auth/"),
|
||
re.compile(r"^apps/backend/app/schemas/"),
|
||
re.compile(r"^apps/backend/app/main\.py$"),
|
||
]
|
||
|
||
MIGRATION_PATTERNS = [
|
||
re.compile(r"^db/etl_feiqiu/migrations/.*\.sql$"),
|
||
re.compile(r"^db/zqyy_app/migrations/.*\.sql$"),
|
||
re.compile(r"^db/fdw/.*\.sql$"),
|
||
]
|
||
|
||
BD_MANUAL_PATTERN = re.compile(r"^docs/database/BD_Manual_.*\.md$")
|
||
DDL_BASELINE_DIR = "docs/database/ddl/"
|
||
AUDIT_CHANGES_DIR = "docs/audit/changes/"
|
||
|
||
|
||
def now_taipei():
|
||
return datetime.now(TZ_TAIPEI)
|
||
|
||
|
||
def sha1hex(s: str) -> str:
|
||
return hashlib.sha1(s.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def is_noise(f: str) -> bool:
|
||
return any(p.search(f) for p in NOISE_PATTERNS)
|
||
|
||
|
||
def safe_read_json(path):
|
||
if not os.path.isfile(path):
|
||
return {}
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def write_json(path, data):
|
||
os.makedirs(os.path.dirname(path) or os.path.join(".kiro", "state"), exist_ok=True)
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||
|
||
|
||
def git_diff_stat():
|
||
try:
|
||
r = subprocess.run(
|
||
["git", "diff", "--stat", "HEAD"],
|
||
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15
|
||
)
|
||
return r.stdout.strip() if r.returncode == 0 else ""
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def git_diff_files(files, max_total=30000, max_per_file=15000):
|
||
"""获取文件的实际 diff 内容。对已跟踪文件用 git diff HEAD，对新文件直接读取内容。"""
|
||
if not files:
|
||
return ""
|
||
all_diff = []
|
||
total_len = 0
|
||
for f in files:
|
||
if total_len >= max_total:
|
||
all_diff.append(f"\n[TRUNCATED: diff exceeds {max_total // 1000}KB]")
|
||
break
|
||
try:
|
||
# 先尝试 git diff HEAD
|
||
r = subprocess.run(
|
||
["git", "diff", "HEAD", "--", f],
|
||
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10
|
||
)
|
||
chunk = ""
|
||
if r.returncode == 0 and r.stdout.strip():
|
||
chunk = r.stdout.strip()
|
||
elif os.path.isfile(f):
|
||
# untracked 新文件：直接读取内容作为 diff
|
||
try:
|
||
with open(f, "r", encoding="utf-8", errors="replace") as fh:
|
||
file_content = fh.read(max_per_file + 100)
|
||
chunk = f"--- /dev/null\n+++ b/{f}\n@@ -0,0 +1 @@\n" + file_content
|
||
except Exception:
|
||
continue
|
||
|
||
if chunk:
|
||
if len(chunk) > max_per_file:
|
||
chunk = chunk[:max_per_file] + f"\n[TRUNCATED: {f} diff too long]"
|
||
all_diff.append(chunk)
|
||
total_len += len(chunk)
|
||
except Exception:
|
||
continue
|
||
return "\n".join(all_diff)
|
||
|
||
|
||
def get_latest_prompt_log():
|
||
log_dir = os.path.join("docs", "audit", "prompt_logs")
|
||
if not os.path.isdir(log_dir):
|
||
return ""
|
||
try:
|
||
files = sorted(
|
||
[f for f in os.listdir(log_dir) if f.startswith("prompt_log_")],
|
||
reverse=True
|
||
)
|
||
if not files:
|
||
return ""
|
||
with open(os.path.join(log_dir, files[0]), "r", encoding="utf-8") as f:
|
||
content = f.read()
|
||
return content[:3000] + "\n[TRUNCATED]" if len(content) > 3000 else content
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
# ── 步骤 1：基于文件基线检测变更 ──
|
||
def detect_changes_via_baseline():
|
||
"""扫描当前工作区，与 promptSubmit 基线对比，返回精确的变更列表。
|
||
|
||
返回 (all_changed_files, external_files, diff_result, no_change)
|
||
- all_changed_files: 本次对话期间所有变更文件（added + modified）
|
||
- external_files: 暂时等于 all_changed_files（后续可通过 Kiro 写入日志细化）
|
||
- diff_result: 完整的 diff 结果 {added, modified, deleted}
|
||
- no_change: 是否无任何变更
|
||
"""
|
||
before = load_baseline()
|
||
after = scan_workspace(".")
|
||
|
||
if not before:
|
||
# 没有基线（首次运行或基线丢失），无法对比，回退到全部文件
|
||
return [], [], {"added": [], "modified": [], "deleted": []}, True
|
||
|
||
diff = diff_baselines(before, after)
|
||
count = total_changes(diff)
|
||
|
||
if count == 0:
|
||
return [], [], diff, True
|
||
|
||
# 所有变更文件 = added + modified（deleted 的文件已不存在，不参与风险判定）
|
||
all_changed = sorted(set(diff["added"] + diff["modified"]))
|
||
|
||
# 过滤噪声
|
||
real_files = [f for f in all_changed if not is_noise(f)]
|
||
|
||
if not real_files:
|
||
return [], [], diff, True
|
||
|
||
# 外部变更：目前所有基线检测到的变更都记录，
|
||
# 因为 Kiro 的写入也会改变 mtime，所以这里的"外部"含义是
|
||
# "本次对话期间发生的所有变更"，包括 Kiro 和非 Kiro 的。
|
||
# 精确区分需要 Kiro 运行时提供写入文件列表，目前不可用。
|
||
external_files = [] # 不再误报外部变更
|
||
|
||
return real_files, external_files, diff, False
|
||
|
||
|
||
|
||
|
||
|
||
|
||
# ── 步骤 3：合规预扫描 ──
|
||
def do_compliance_prescan(all_files):
|
||
result = {
|
||
"new_migration_sql": [],
|
||
"new_or_modified_sql": [],
|
||
"code_without_docs": [],
|
||
"new_files": [],
|
||
"has_bd_manual": False,
|
||
"has_audit_record": False,
|
||
"has_ddl_baseline": False,
|
||
"api_changed": False,
|
||
"openapi_spec_stale": False,
|
||
}
|
||
|
||
code_files = []
|
||
doc_files = set()
|
||
|
||
for f in all_files:
|
||
if is_noise(f):
|
||
continue
|
||
for mp in MIGRATION_PATTERNS:
|
||
if mp.search(f):
|
||
result["new_migration_sql"].append(f)
|
||
break
|
||
if f.endswith(".sql"):
|
||
result["new_or_modified_sql"].append(f)
|
||
if BD_MANUAL_PATTERN.search(f):
|
||
result["has_bd_manual"] = True
|
||
if f.startswith(AUDIT_CHANGES_DIR):
|
||
result["has_audit_record"] = True
|
||
if f.startswith(DDL_BASELINE_DIR):
|
||
result["has_ddl_baseline"] = True
|
||
if f.endswith(".md") or "/docs/" in f:
|
||
doc_files.add(f)
|
||
if f.endswith((".py", ".ts", ".tsx", ".js", ".jsx")):
|
||
code_files.append(f)
|
||
# 检测接口相关文件变更
|
||
for ap in API_CHANGE_PATTERNS:
|
||
if ap.search(f):
|
||
result["api_changed"] = True
|
||
break
|
||
|
||
# 接口变更但 openapi spec 未同步更新 → 标记过期
|
||
if result["api_changed"] and "docs/contracts/openapi/backend-api.json" not in all_files:
|
||
result["openapi_spec_stale"] = True
|
||
|
||
for cf in code_files:
|
||
expected_docs = []
|
||
for prefix, docs in DOC_MAP.items():
|
||
if cf.startswith(prefix):
|
||
expected_docs.extend(docs)
|
||
if expected_docs:
|
||
has_doc = False
|
||
for ed in expected_docs:
|
||
if ed in doc_files:
|
||
has_doc = True
|
||
break
|
||
if ed.endswith("/") and any(d.startswith(ed) for d in doc_files):
|
||
has_doc = True
|
||
break
|
||
if not has_doc:
|
||
result["code_without_docs"].append({
|
||
"file": cf,
|
||
"expected_docs": expected_docs,
|
||
})
|
||
|
||
needs_check = bool(
|
||
result["new_migration_sql"]
|
||
or result["code_without_docs"]
|
||
or result["openapi_spec_stale"]
|
||
)
|
||
|
||
now = now_taipei()
|
||
write_json(COMPLIANCE_PATH, {
|
||
"needs_check": needs_check,
|
||
"scanned_at": now.isoformat(),
|
||
**result,
|
||
})
|
||
return result
|
||
|
||
|
||
# ── 步骤 4：构建审计上下文 ──
|
||
def do_build_audit_context(all_files, diff_result, compliance):
|
||
now = now_taipei()
|
||
audit_state = safe_read_json(STATE_PATH)
|
||
prompt_info = safe_read_json(PROMPT_ID_PATH)
|
||
|
||
# 使用 audit_state 中的 changed_files（来自 git status 的风险文件）
|
||
# 与本次对话的 baseline diff 合并
|
||
git_changed = audit_state.get("changed_files", [])
|
||
session_changed = all_files # 本次对话期间变更的文件
|
||
|
||
# 合并两个来源，去重
|
||
all_changed = sorted(set(git_changed + session_changed))
|
||
|
||
high_risk_files = [
|
||
f for f in all_changed
|
||
if any(p.search(f) for p in HIGH_RISK_PATTERNS)
|
||
]
|
||
|
||
diff_stat = git_diff_stat()
|
||
high_risk_diff = git_diff_files(high_risk_files)
|
||
prompt_log = get_latest_prompt_log()
|
||
|
||
context = {
|
||
"built_at": now.isoformat(),
|
||
"prompt_id": prompt_info.get("prompt_id", "unknown"),
|
||
"prompt_at": prompt_info.get("at", ""),
|
||
"audit_required": audit_state.get("audit_required", False),
|
||
"db_docs_required": audit_state.get("db_docs_required", False),
|
||
"reasons": audit_state.get("reasons", []),
|
||
"changed_files": all_changed[:100],
|
||
"high_risk_files": high_risk_files,
|
||
"session_diff": {
|
||
"added": diff_result.get("added", [])[:50],
|
||
"modified": diff_result.get("modified", [])[:50],
|
||
"deleted": diff_result.get("deleted", [])[:50],
|
||
},
|
||
"compliance": {
|
||
"code_without_docs": compliance.get("code_without_docs", []),
|
||
"new_migration_sql": compliance.get("new_migration_sql", []),
|
||
"has_bd_manual": compliance.get("has_bd_manual", False),
|
||
"has_audit_record": compliance.get("has_audit_record", False),
|
||
"has_ddl_baseline": compliance.get("has_ddl_baseline", False),
|
||
"api_changed": compliance.get("api_changed", False),
|
||
"openapi_spec_stale": compliance.get("openapi_spec_stale", False),
|
||
},
|
||
"diff_stat": diff_stat,
|
||
"high_risk_diff": high_risk_diff,
|
||
"latest_prompt_log": prompt_log,
|
||
}
|
||
|
||
write_json(CONTEXT_PATH, context)
|
||
|
||
|
||
# ── 步骤 5：审计提醒（15 分钟限频） ──
|
||
def do_audit_reminder(real_files):
|
||
state = safe_read_json(STATE_PATH)
|
||
if not state.get("audit_required"):
|
||
return
|
||
|
||
# 无变更时不提醒
|
||
if not real_files:
|
||
return
|
||
|
||
now = now_taipei()
|
||
last_str = state.get("last_reminded_at")
|
||
if last_str:
|
||
try:
|
||
last = datetime.fromisoformat(last_str)
|
||
if (now - last) < MIN_INTERVAL:
|
||
return
|
||
except Exception:
|
||
pass
|
||
|
||
state["last_reminded_at"] = now.isoformat()
|
||
write_json(STATE_PATH, state)
|
||
|
||
reasons = state.get("reasons", [])
|
||
reason_text = ", ".join(reasons) if reasons else "high-risk paths changed"
|
||
|
||
# 仅信息性提醒，exit(0) 避免 agent 将其视为错误并自行执行审计
|
||
# 审计留痕统一由用户手动触发 /audit 完成
|
||
sys.stderr.write(
|
||
f"[AUDIT REMINDER] Pending audit ({reason_text}), "
|
||
f"{len(real_files)} files changed this session. "
|
||
f"Run /audit to sync. (15min rate limit)\n"
|
||
)
|
||
sys.exit(0)
|
||
|
||
|
||
# ── 步骤 6：全量会话记录提取 ──
|
||
def do_full_session_extract():
|
||
"""从 Kiro globalStorage 提取当前 execution 的全量对话记录。
|
||
调用 scripts/ops/extract_kiro_session.py 的核心逻辑。
|
||
仅提取最新一条未索引的 execution，避免重复。
|
||
"""
|
||
# 动态导入提取器（避免启动时 import 开销）
|
||
scripts_ops = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "scripts", "ops")
|
||
scripts_ops = os.path.normpath(scripts_ops)
|
||
if scripts_ops not in sys.path:
|
||
sys.path.insert(0, scripts_ops)
|
||
|
||
try:
|
||
from extract_kiro_session import extract_latest
|
||
except ImportError:
|
||
return # 提取器不存在则静默跳过
|
||
|
||
# globalStorage 路径：从环境变量或默认位置
|
||
global_storage = os.environ.get(
|
||
"KIRO_GLOBAL_STORAGE",
|
||
os.path.join(os.environ.get("APPDATA", ""), "Kiro", "User", "globalStorage")
|
||
)
|
||
workspace_path = os.getcwd()
|
||
|
||
extract_latest(global_storage, workspace_path)
|
||
|
||
|
||
def _extract_summary_content(md_content: str) -> str:
|
||
"""从 session log markdown 中提取适合生成摘要的内容。
|
||
|
||
策略：如果"用户输入"包含 CONTEXT TRANSFER（跨轮续接），
|
||
则替换为简短标注，避免历史背景干扰本轮摘要生成。
|
||
"""
|
||
import re
|
||
# 检测用户输入中是否包含 context transfer
|
||
ct_pattern = re.compile(r"## 2\. 用户输入\s*\n```\s*\n.*?CONTEXT TRANSFER", re.DOTALL)
|
||
if ct_pattern.search(md_content):
|
||
# 替换"用户输入"section 为简短标注
|
||
# 匹配从 "## 2. 用户输入" 到下一个 "## 3." 之间的内容
|
||
md_content = re.sub(
|
||
r"(## 2\. 用户输入)\s*\n```[\s\S]*?```\s*\n(?=## 3\.)",
|
||
r"\1\n\n[本轮为 Context Transfer 续接，用户输入为历史多轮摘要，已省略。请基于执行摘要和对话记录中的实际工具调用判断本轮工作。]\n\n",
|
||
md_content,
|
||
)
|
||
return md_content
|
||
|
||
|
||
# ── 步骤 7：为最新 session 生成 LLM 摘要 ──
|
||
_SUMMARY_SYSTEM_PROMPT = """你是一个专业的技术对话分析师。你的任务是为 AI 编程助手的一轮执行（execution）生成简洁的中文摘要。
|
||
|
||
背景：一个对话（chatSession）包含多轮执行（execution）。每轮执行 = 用户发一条消息 → AI 完成响应。你收到的是单轮执行的完整记录。
|
||
|
||
摘要规则：
|
||
1. 只描述本轮执行实际完成的工作，不要描述历史背景
|
||
2. 列出完成的功能点/任务（一轮可能完成多个）
|
||
3. 包含关键技术细节：文件路径、模块名、数据库表、API 端点等
|
||
4. bug 修复要说明原因和方案
|
||
5. 不写过程性描述（"用户说..."），只写结果
|
||
6. 内容太短或无实质内容的，写"无实质内容"
|
||
7. 不限字数，信息完整优先，避免截断失真
|
||
|
||
重要：
|
||
- "执行摘要"（📋）是最可靠的信息源，优先基于它判断本轮做了什么
|
||
- 如果"用户输入"包含 CONTEXT TRANSFER，那是之前多轮的历史摘要，不是本轮工作
|
||
- 对话记录中的实际工具调用和文件变更才是本轮的真实操作
|
||
|
||
请直接输出摘要，不要添加任何前缀或解释。"""
|
||
|
||
|
||
def do_generate_description():
|
||
"""为缺少 description 的主对话 entry 调用百炼 API 生成摘要，写入双索引。"""
|
||
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
|
||
api_key = os.environ.get("BAILIAN_API_KEY", "")
|
||
if not api_key:
|
||
return
|
||
|
||
model = os.environ.get("BAILIAN_MODEL", "qwen-plus")
|
||
base_url = os.environ.get("BAILIAN_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1")
|
||
|
||
scripts_ops = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "scripts", "ops")
|
||
scripts_ops = os.path.normpath(scripts_ops)
|
||
if scripts_ops not in sys.path:
|
||
sys.path.insert(0, scripts_ops)
|
||
|
||
try:
|
||
from extract_kiro_session import load_index, save_index, load_full_index, save_full_index
|
||
except ImportError:
|
||
return
|
||
|
||
index = load_index()
|
||
entries = index.get("entries", {})
|
||
if not entries:
|
||
return
|
||
|
||
# 收集所有缺少 description 的主对话 entry
|
||
targets = []
|
||
for eid, ent in entries.items():
|
||
if ent.get("is_sub"):
|
||
continue
|
||
if not ent.get("description"):
|
||
targets.append((eid, ent))
|
||
|
||
if not targets:
|
||
return
|
||
|
||
# agent_on_stop 场景下限制处理数量，避免超时
|
||
# 批量处理积压用独立脚本 batch_generate_summaries.py
|
||
MAX_PER_RUN = 10
|
||
if len(targets) > MAX_PER_RUN:
|
||
# 优先处理最新的（按 startTime 降序）
|
||
targets.sort(key=lambda t: t[1].get("startTime", ""), reverse=True)
|
||
targets = targets[:MAX_PER_RUN]
|
||
|
||
try:
|
||
from openai import OpenAI
|
||
client = OpenAI(api_key=api_key, base_url=base_url)
|
||
except Exception:
|
||
return
|
||
|
||
full_index = load_full_index()
|
||
full_entries = full_index.get("entries", {})
|
||
generated = 0
|
||
|
||
for target_eid, target_entry in targets:
|
||
out_dir = target_entry.get("output_dir", "")
|
||
if not out_dir or not os.path.isdir(out_dir):
|
||
continue
|
||
|
||
# 找到该 entry 对应的 main_*.md 文件
|
||
main_files = sorted(
|
||
f for f in os.listdir(out_dir)
|
||
if f.startswith("main_") and f.endswith(".md")
|
||
and target_eid[:8] in f # 按 executionId 短码匹配
|
||
)
|
||
if not main_files:
|
||
# 回退：取目录下所有 main 文件
|
||
main_files = sorted(
|
||
f for f in os.listdir(out_dir)
|
||
if f.startswith("main_") and f.endswith(".md")
|
||
)
|
||
if not main_files:
|
||
continue
|
||
|
||
content_parts = []
|
||
for mf in main_files:
|
||
try:
|
||
with open(os.path.join(out_dir, mf), "r", encoding="utf-8") as fh:
|
||
content_parts.append(fh.read())
|
||
except Exception:
|
||
continue
|
||
if not content_parts:
|
||
continue
|
||
|
||
content = "\n\n---\n\n".join(content_parts)
|
||
content = _extract_summary_content(content)
|
||
if len(content) > 60000:
|
||
content = content[:60000] + "\n\n[TRUNCATED]"
|
||
|
||
try:
|
||
resp = client.chat.completions.create(
|
||
model=model,
|
||
messages=[
|
||
{"role": "system", "content": _SUMMARY_SYSTEM_PROMPT},
|
||
{"role": "user", "content": f"请为以下单轮执行记录生成摘要：\n\n{content}"},
|
||
],
|
||
max_tokens=4096,
|
||
)
|
||
description = resp.choices[0].message.content.strip()
|
||
except Exception:
|
||
continue # 单条失败不影响其他
|
||
|
||
if not description:
|
||
continue
|
||
|
||
# 写入双索引（内存中）
|
||
entries[target_eid]["description"] = description
|
||
if target_eid in full_entries:
|
||
full_entries[target_eid]["description"] = description
|
||
generated += 1
|
||
|
||
# 批量保存
|
||
if generated > 0:
|
||
save_index(index)
|
||
save_full_index(full_index)
|
||
|
||
|
||
def main():
|
||
ensure_repo_root()
|
||
now = now_taipei()
|
||
force_rebuild = "--force-rebuild" in sys.argv
|
||
|
||
# 全量会话记录提取（无论是否有文件变更，每次对话都要记录）
|
||
try:
|
||
do_full_session_extract()
|
||
except Exception:
|
||
pass
|
||
|
||
# 步骤 1：基于文件基线检测变更
|
||
real_files, external_files, diff_result, no_change = detect_changes_via_baseline()
|
||
|
||
# 无任何文件变更 → 跳过所有审查（除非 --force-rebuild）
|
||
if no_change and not force_rebuild:
|
||
return
|
||
|
||
# --force-rebuild 且无变更时，仍需基于 git status 重建 context
|
||
if no_change and force_rebuild:
|
||
try:
|
||
compliance = do_compliance_prescan(real_files or [])
|
||
except Exception:
|
||
compliance = {}
|
||
try:
|
||
do_build_audit_context(real_files or [], diff_result, compliance)
|
||
except Exception:
|
||
pass
|
||
return
|
||
|
||
# 步骤 2：合规预扫描（基于本次对话变更的文件）
|
||
compliance = {}
|
||
try:
|
||
compliance = do_compliance_prescan(real_files)
|
||
except Exception:
|
||
pass
|
||
|
||
# 步骤 4：构建审计上下文
|
||
try:
|
||
do_build_audit_context(real_files, diff_result, compliance)
|
||
except Exception:
|
||
pass
|
||
|
||
# 步骤 7：审计提醒（信息性，exit(0)，不触发 agent 自行审计）
|
||
try:
|
||
do_audit_reminder(real_files)
|
||
except SystemExit:
|
||
pass # exit(0) 信息性退出，不需要 re-raise
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except SystemExit as e:
|
||
sys.exit(e.code)
|
||
except Exception:
|
||
pass
|