9702f1b186
- 新增 ChangeClassifier（爆炸半径分类：IMPLEMENTATION/INTERFACE/GLOBAL_CONSTRAINT） - 新增 ImpactPropagator（BFS影响传播：IMPACTED/BOUNDARY/SAFE差异化标记） - task_graph.py：invalidate_by_adr()差异化失效 + CascadeReport扩展字段（向后兼容） - eng_mode.py：三阶段差异化流程（分类→传播→失效→git操作→验证任务→重规划） - eng_mode.py：_git_squash_merge_and_tag() + Phase 7 集成 - do_mode.py：_ensure_all_committed() Worker git操作强制 - adr_watcher.py：内容快照 + get_content_for_classification() - events.py：ADR_CLASSIFIED/IMPACT_PROPAGATED/BOUNDARY_VERIFICATION_GENERATED - partial_replanner.py：replan_with_constraints() + generate_verification_tasks() - project_bootstrap.py：ensure_git_initialized() Arc规划前检测 - test_p1_21_phase2.py：21个新测试，95个全量测试0失败 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
102 lines
4.1 KiB
Python
102 lines
4.1 KiB
Python
"""
|
||
ADR 变更爆炸半径分类 — P1-21 / 3.2.17 Phase 1 组件。
|
||
将 ADR 内容变更按影响范围分为 IMPLEMENTATION / INTERFACE / GLOBAL_CONSTRAINT 三级。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import enum
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
|
||
|
||
class BlastRadius(enum.Enum):
|
||
IMPLEMENTATION = "implementation"
|
||
INTERFACE = "interface"
|
||
GLOBAL_CONSTRAINT = "global_constraint"
|
||
|
||
|
||
@dataclass
|
||
class ChangeClassification:
|
||
adr_id: str
|
||
blast_radius: BlastRadius
|
||
reason: str = ""
|
||
changed_sections: list[str] = field(default_factory=list)
|
||
|
||
|
||
class ChangeClassifier:
|
||
"""基于关键词启发式的 ADR 变更爆炸半径分类。
|
||
|
||
分类规则：
|
||
- IMPLEMENTATION: 仅实现细节变更，不影响公开接口。关键词：implementation detail, can use, for example, internal, private
|
||
- INTERFACE: 公开 API/合约变更。关键词：interface, api, contract, signature, must implement, public header, exported, caller
|
||
- GLOBAL_CONSTRAINT: 跨模块约束变更。关键词：must, shall, all modules, coding standard, every module, entire system, global policy
|
||
|
||
回退策略：内容不足无法确定时回退到 INTERFACE（保守：宁可多失效不少失效）。
|
||
"""
|
||
|
||
_INTERFACE_KEYWORDS = [
|
||
"interface", "api", "contract", "signature", "must implement",
|
||
"public header", "exported", "caller",
|
||
]
|
||
_GLOBAL_KEYWORDS = [
|
||
"must", "shall", "all modules", "coding standard",
|
||
"every module", "entire system", "global policy",
|
||
]
|
||
_IMPLEMENTATION_KEYWORDS = [
|
||
"implementation detail", "can use", "for example",
|
||
"internally", "private", "optional",
|
||
]
|
||
|
||
def classify(self, adr_id: str, old_content: str | None,
|
||
new_content: str | None) -> ChangeClassification:
|
||
if new_content is None:
|
||
return ChangeClassification(
|
||
adr_id=adr_id, blast_radius=BlastRadius.GLOBAL_CONSTRAINT,
|
||
reason="ADR deleted — all dependent tasks must be invalidated",
|
||
)
|
||
if old_content is None:
|
||
return ChangeClassification(
|
||
adr_id=adr_id, blast_radius=BlastRadius.IMPLEMENTATION,
|
||
reason="new ADR — no existing tasks to invalidate",
|
||
)
|
||
|
||
text = new_content.lower()
|
||
|
||
global_hits = sum(1 for kw in self._GLOBAL_KEYWORDS if kw in text)
|
||
interface_hits = sum(1 for kw in self._INTERFACE_KEYWORDS if kw in text)
|
||
impl_hits = sum(1 for kw in self._IMPLEMENTATION_KEYWORDS if kw in text)
|
||
|
||
if global_hits >= 2:
|
||
return ChangeClassification(
|
||
adr_id=adr_id, blast_radius=BlastRadius.GLOBAL_CONSTRAINT,
|
||
reason=f"global constraint keywords matched ({global_hits} hits)",
|
||
)
|
||
if interface_hits >= 2:
|
||
return ChangeClassification(
|
||
adr_id=adr_id, blast_radius=BlastRadius.INTERFACE,
|
||
reason=f"interface keywords matched ({interface_hits} hits)",
|
||
)
|
||
if interface_hits >= 1:
|
||
return ChangeClassification(
|
||
adr_id=adr_id, blast_radius=BlastRadius.INTERFACE,
|
||
reason="interface keyword matched — conservative classification",
|
||
)
|
||
if impl_hits >= interface_hits and impl_hits > 0:
|
||
return ChangeClassification(
|
||
adr_id=adr_id, blast_radius=BlastRadius.IMPLEMENTATION,
|
||
reason=f"implementation keywords matched ({impl_hits} hits)",
|
||
)
|
||
|
||
# 无法确定 → 保守回退
|
||
return ChangeClassification(
|
||
adr_id=adr_id, blast_radius=BlastRadius.INTERFACE,
|
||
reason="inconclusive — defaulting to INTERFACE (conservative)",
|
||
)
|
||
|
||
def classify_from_files(self, adr_id: str, old_path: Path | None,
|
||
new_path: Path | None) -> ChangeClassification:
|
||
old_content = old_path.read_text(encoding="utf-8") if old_path and old_path.exists() else None
|
||
new_content = new_path.read_text(encoding="utf-8") if new_path and new_path.exists() else None
|
||
return self.classify(adr_id, old_content, new_content)
|