aaronrene / overseer-kit public
compact.py python
166 lines 4.8 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago
1 """Archive old handover change-log bullets (§LT.6)."""
2
3 from __future__ import annotations
4
5 import re
6 from dataclasses import dataclass
7 from datetime import date
8 from pathlib import Path
9
10 from adapters.config import OverseerConfig, resolve_lane_docs
11
12 CHANGE_LOG_START = "<!-- overseer:anchor:change-log -->"
13 CHANGE_LOG_END = "<!-- /overseer:anchor:change-log -->"
14 DATED_ENTRY_START = re.compile(r"^- \*\*\d{4}-\d{2}-\d{2}\*\*")
15 POINTER_PREFIX = "- Older entries:"
16
17
18 @dataclass(frozen=True)
19 class CompactReport:
20 ok: bool
21 compacted: int
22 keep: int
23 archive: str
24 wrote: bool
25 reason: str | None = None
26
27
28 @dataclass(frozen=True)
29 class _ChangeLogEntry:
30 lines: tuple[str, ...]
31
32
33 def compact_handover_change_log(
34 config: OverseerConfig,
35 repo_root: Path,
36 *,
37 keep: int,
38 write: bool,
39 lane: str | None = None,
40 ) -> CompactReport:
41 """Compact dated change-log bullets; default dry-run when write is False."""
42 lane_docs = resolve_lane_docs(config, lane)
43 docs_root = config.repo.root_relative_docs
44 handover_rel = f"{docs_root}/{lane_docs.handover}".replace("//", "/")
45 handover_path = repo_root / handover_rel
46 archive_rel = f"{docs_root}/archive/handover/CHANGE-LOG.md".replace("//", "/")
47 archive_path = repo_root / archive_rel
48
49 if not handover_path.is_file():
50 return CompactReport(
51 ok=False,
52 compacted=0,
53 keep=keep,
54 archive=archive_rel,
55 wrote=False,
56 reason="change_log_anchor_missing",
57 )
58
59 text = handover_path.read_text(encoding="utf-8")
60 region = _extract_region(text)
61 if region is None:
62 return CompactReport(
63 ok=False,
64 compacted=0,
65 keep=keep,
66 archive=archive_rel,
67 wrote=False,
68 reason="change_log_anchor_missing",
69 )
70
71 before, region_body, after = region
72 entries = _collect_dated_entries(region_body)
73 if len(entries) <= keep:
74 return CompactReport(
75 ok=True,
76 compacted=0,
77 keep=keep,
78 archive=archive_rel,
79 wrote=False,
80 )
81
82 kept = entries[:keep]
83 archived = entries[keep:]
84 compacted_count = len(archived)
85
86 kept_text = _render_entries(kept)
87 pointer = f"{POINTER_PREFIX} {archive_rel}"
88 new_region = kept_text
89 if kept_text:
90 new_region += "\n\n"
91 new_region += pointer + "\n"
92
93 new_handover = before + new_region + after
94 archive_block = _render_entries(archived)
95 heading = f"## Compacted {date.today().isoformat()}\n\n"
96 archive_append = heading + archive_block
97 if archive_block and not archive_block.endswith("\n"):
98 archive_append += "\n"
99
100 if not write:
101 return CompactReport(
102 ok=True,
103 compacted=compacted_count,
104 keep=keep,
105 archive=archive_rel,
106 wrote=False,
107 )
108
109 archive_path.parent.mkdir(parents=True, exist_ok=True)
110 if archive_path.is_file():
111 existing = archive_path.read_text(encoding="utf-8")
112 if existing and not existing.endswith("\n"):
113 existing += "\n"
114 archive_path.write_text(existing + archive_append, encoding="utf-8")
115 else:
116 title = f"# Archived handover change log — {config.repo.name}\n\n"
117 archive_path.write_text(title + archive_append, encoding="utf-8")
118
119 handover_path.write_text(new_handover, encoding="utf-8")
120 return CompactReport(
121 ok=True,
122 compacted=compacted_count,
123 keep=keep,
124 archive=archive_rel,
125 wrote=True,
126 )
127
128
129 def _extract_region(text: str) -> tuple[str, str, str] | None:
130 start = text.find(CHANGE_LOG_START)
131 end = text.find(CHANGE_LOG_END)
132 if start == -1 or end == -1 or end <= start:
133 return None
134 region_start = start + len(CHANGE_LOG_START)
135 before = text[:region_start]
136 if before and not before.endswith("\n"):
137 before += "\n"
138 region_body = text[region_start:end]
139 after = text[end:]
140 return before, region_body, after
141
142
143 def _collect_dated_entries(region_body: str) -> list[_ChangeLogEntry]:
144 entries: list[_ChangeLogEntry] = []
145 current: list[str] = []
146 for line in region_body.splitlines():
147 if line.strip().startswith(POINTER_PREFIX):
148 continue
149 if DATED_ENTRY_START.match(line):
150 if current:
151 entries.append(_ChangeLogEntry(lines=tuple(current)))
152 current = [line]
153 elif current:
154 current.append(line)
155 if current:
156 entries.append(_ChangeLogEntry(lines=tuple(current)))
157 return entries
158
159
160 def _render_entries(entries: list[_ChangeLogEntry]) -> str:
161 chunks: list[str] = []
162 for entry in entries:
163 chunk = "\n".join(entry.lines).rstrip()
164 if chunk:
165 chunks.append(chunk)
166 return "\n\n".join(chunks)
File History 1 commit
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago