feat: JSON preprocessing and wisdom cap (#85, #86) (#116)
- Add scripts/preprocess_for_analysis.py: extracts needed fields, truncates descriptions, deduplicates repos, estimates token reduction - Add scripts/wisdom_cap.py: checks wisdom.md against 5KB soft limit, retires oldest heuristics to wisdom-archive.md (idempotent, never deletes) - Update prompts/reskill.md with wisdom size management instructions - Add comprehensive tests for both scripts (25 tests) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Juan Manuel Servera committed
May 19, 2026 at 16:36 UTC
57d3eef4f03b8074e7181f3dd06cb2e70d90a040
5 files changed
+642
prompts/reskill.md
+7
@@ -47,6 +47,13 @@ Write the full contents of `{{OUTPUT_PATH}}` as a markdown reskill report.
47
6. Extract new reusable skills or patterns when a lesson is concrete enough to guide future analysis.
48
7. Ground the retrospective in evidence from the actual summaries and snapshots, not in generic advice.
49
50
+## Wisdom size management
51
+
52
+IMPORTANT: wisdom.md has a 5KB soft limit. When adding new heuristics:
53
+- Retire obsolete ones that have been contradicted by recent data
54
+- Move retired heuristics to wisdom-archive.md with a note on why retired
55
+- Prefer updating existing heuristics over adding new duplicates
56
+
57
## Output requirements
58
59
- Output only the finished markdown report.
scripts/preprocess_for_analysis.py
new
+133
@@ -0,0 +1,133 @@
1
+#!/usr/bin/env python3
2
+"""Pre-process raw crawl JSON to reduce token count for analysis prompts.
3
+
4
+Extracts only fields needed by the analysis prompt, truncates descriptions,
5
+and computes basic signals to produce a compact JSON suitable for LLM input.
6
+
7
+CLI:
8
+ python scripts/preprocess_for_analysis.py \
9
+ --input data/raw/2026-W21.json \
10
+ --output data/raw/2026-W21-compact.json \
11
+ --max-desc-length 200
12
+"""
13
+
14
+from __future__ import annotations
15
+
16
+import argparse
17
+import json
18
+import sys
19
+from datetime import datetime, timezone
20
+from pathlib import Path
21
+
22
+
23
+def estimate_tokens(text: str) -> int:
24
+ """Rough token estimate: characters / 4."""
25
+ return len(text) // 4
26
+
27
+
28
+def compute_age_days(created_at: str | None, reference: datetime | None = None) -> int | None:
29
+ """Compute age in days from created_at ISO timestamp."""
30
+ if not created_at:
31
+ return None
32
+ try:
33
+ created = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
34
+ ref = reference or datetime.now(timezone.utc)
35
+ return max(0, (ref - created).days)
36
+ except (ValueError, TypeError):
37
+ return None
38
+
39
+
40
+def compact_repo(repo: dict, max_desc: int, reference_date: datetime | None = None) -> dict:
41
+ """Extract and compact a single repo entry."""
42
+ desc = (repo.get("description") or "")[:max_desc]
43
+ return {
44
+ "name": repo.get("name", ""),
45
+ "desc": desc,
46
+ "stars": repo.get("stars", 0),
47
+ "gained": repo.get("stars_gained", repo.get("gained", 0)),
48
+ "topics": repo.get("topics", []),
49
+ "lang": repo.get("language"),
50
+ "age_days": compute_age_days(repo.get("created_at"), reference_date),
51
+ }
52
+
53
+
54
+def compute_signals(repos: list[dict]) -> dict:
55
+ """Compute aggregate signals from the compacted repo list."""
56
+ topic_counts: dict[str, int] = {}
57
+ for r in repos:
58
+ for t in r.get("topics", []):
59
+ topic_counts[t] = topic_counts.get(t, 0) + 1
60
+ top_topics = sorted(topic_counts.items(), key=lambda x: -x[1])[:10]
61
+ return {"top_topics": [t for t, _ in top_topics]}
62
+
63
+
64
+def preprocess(data: dict, max_desc: int = 200, reference_date: datetime | None = None) -> dict:
65
+ """Transform raw crawl JSON into compact analysis format."""
66
+ original_text = json.dumps(data)
67
+ original_tokens = estimate_tokens(original_text)
68
+
69
+ # Combine new_repos and trending_repos
70
+ all_repos = data.get("new_repos", []) + data.get("trending_repos", [])
71
+
72
+ # Deduplicate by name
73
+ seen = set()
74
+ unique_repos = []
75
+ for r in all_repos:
76
+ name = r.get("name", "")
77
+ if name not in seen:
78
+ seen.add(name)
79
+ unique_repos.append(r)
80
+
81
+ compacted = [compact_repo(r, max_desc, reference_date) for r in unique_repos]
82
+ signals = compute_signals(compacted)
83
+
84
+ result = {
85
+ "week": data.get("week", ""),
86
+ "repos": compacted,
87
+ "signals": signals,
88
+ }
89
+
90
+ compact_text = json.dumps(result)
91
+ compact_tokens = estimate_tokens(compact_text)
92
+ reduction_pct = round((1 - compact_tokens / original_tokens) * 100) if original_tokens > 0 else 0
93
+
94
+ result["stats"] = {
95
+ "original_tokens_est": original_tokens,
96
+ "compact_tokens_est": compact_tokens,
97
+ "reduction_pct": reduction_pct,
98
+ }
99
+
100
+ return result
101
+
102
+
103
+def main(argv: list[str] | None = None) -> int:
104
+ parser = argparse.ArgumentParser(description="Pre-process raw JSON for analysis")
105
+ parser.add_argument("--input", required=True, help="Path to raw crawl JSON")
106
+ parser.add_argument("--output", help="Output path (default: input with -compact suffix)")
107
+ parser.add_argument("--max-desc-length", type=int, default=200, help="Max description length")
108
+ args = parser.parse_args(argv)
109
+
110
+ input_path = Path(args.input)
111
+ if not input_path.exists():
112
+ print(f"Error: input file not found: {input_path}", file=sys.stderr)
113
+ return 1
114
+
115
+ output_path = Path(args.output) if args.output else input_path.with_stem(input_path.stem + "-compact")
116
+
117
+ with open(input_path, encoding="utf-8") as f:
118
+ data = json.load(f)
119
+
120
+ result = preprocess(data, max_desc=args.max_desc_length)
121
+
122
+ output_path.parent.mkdir(parents=True, exist_ok=True)
123
+ with open(output_path, "w", encoding="utf-8") as f:
124
+ json.dump(result, f, indent=2)
125
+
126
+ stats = result["stats"]
127
+ print(f"Preprocessed: {input_path} -> {output_path}")
128
+ print(f" Tokens: {stats['original_tokens_est']} -> {stats['compact_tokens_est']} ({stats['reduction_pct']}% reduction)")
129
+ return 0
130
+
131
+
132
+if __name__ == "__main__":
133
+ sys.exit(main())
scripts/wisdom_cap.py
new
+172
@@ -0,0 +1,172 @@
1
+#!/usr/bin/env python3
2
+"""Wisdom.md size cap and retirement management.
3
+
4
+Checks wisdom.md file size against a soft limit and retires oldest
5
+heuristics to an archive file when the limit is exceeded.
6
+
7
+CLI:
8
+ python scripts/wisdom_cap.py --topic ai-ml [--limit 5120] [--dry-run]
9
+"""
10
+
11
+from __future__ import annotations
12
+
13
+import argparse
14
+import re
15
+import sys
16
+from datetime import datetime, timezone
17
+from pathlib import Path
18
+
19
+
20
+SQUAD_DIR = Path(".squad/topics")
21
+DEFAULT_LIMIT = 5120 # 5KB soft limit
22
+
23
+
24
+def get_wisdom_path(topic: str) -> Path:
25
+ return SQUAD_DIR / topic / "wisdom.md"
26
+
27
+
28
+def get_archive_path(topic: str) -> Path:
29
+ return SQUAD_DIR / topic / "wisdom-archive.md"
30
+
31
+
32
+def parse_heuristics(content: str) -> list[dict]:
33
+ """Parse wisdom.md into sections with their heuristic bullet points.
34
+
35
+ Returns a list of dicts with 'section', 'line', and 'text' keys.
36
+ """
37
+ heuristics = []
38
+ current_section = ""
39
+ for i, line in enumerate(content.splitlines()):
40
+ if line.startswith("## "):
41
+ current_section = line.strip("# ").strip()
42
+ elif line.startswith("- "):
43
+ heuristics.append({
44
+ "section": current_section,
45
+ "line": i,
46
+ "text": line,
47
+ })
48
+ return heuristics
49
+
50
+
51
+def select_for_retirement(heuristics: list[dict], bytes_to_free: int) -> list[dict]:
52
+ """Select heuristics from the end of the list (oldest/least-referenced first).
53
+
54
+ Strategy: retire from the bottom of each section first, working backwards.
55
+ """
56
+ # Retire from the end of the file upward until we've freed enough bytes
57
+ retired = []
58
+ freed = 0
59
+ for h in reversed(heuristics):
60
+ if freed >= bytes_to_free:
61
+ break
62
+ retired.append(h)
63
+ freed += len(h["text"].encode("utf-8")) + 1 # +1 for newline
64
+ return retired
65
+
66
+
67
+def retire_heuristics(
68
+ wisdom_path: Path, archive_path: Path, limit: int, dry_run: bool = False
69
+) -> dict:
70
+ """Check wisdom size and retire heuristics if over limit.
71
+
72
+ Returns a summary dict with action details.
73
+ """
74
+ if not wisdom_path.exists():
75
+ return {"status": "skip", "reason": "wisdom.md not found", "path": str(wisdom_path)}
76
+
77
+ content = wisdom_path.read_text(encoding="utf-8")
78
+ current_size = len(content.encode("utf-8"))
79
+
80
+ if current_size <= limit:
81
+ return {
82
+ "status": "ok",
83
+ "size": current_size,
84
+ "limit": limit,
85
+ "message": f"Under limit ({current_size}/{limit} bytes)",
86
+ }
87
+
88
+ bytes_over = current_size - limit
89
+ heuristics = parse_heuristics(content)
90
+
91
+ if not heuristics:
92
+ return {
93
+ "status": "warn",
94
+ "size": current_size,
95
+ "message": "Over limit but no parseable heuristics to retire",
96
+ }
97
+
98
+ to_retire = select_for_retirement(heuristics, bytes_over)
99
+
100
+ if dry_run:
101
+ return {
102
+ "status": "dry_run",
103
+ "size": current_size,
104
+ "limit": limit,
105
+ "would_retire": len(to_retire),
106
+ "items": [h["text"] for h in to_retire],
107
+ }
108
+
109
+ # Build archive entry
110
+ timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
111
+ archive_entry = f"\n## Retired {timestamp}\n\nReason: wisdom.md exceeded {limit} byte soft limit\n\n"
112
+ archive_entry += "\n".join(h["text"] for h in to_retire) + "\n"
113
+
114
+ # Write archive
115
+ archive_path.parent.mkdir(parents=True, exist_ok=True)
116
+ if archive_path.exists():
117
+ existing = archive_path.read_text(encoding="utf-8")
118
+ archive_path.write_text(existing + archive_entry, encoding="utf-8")
119
+ else:
120
+ header = "# Wisdom Archive\n\nRetired heuristics from wisdom.md.\n"
121
+ archive_path.write_text(header + archive_entry, encoding="utf-8")
122
+
123
+ # Remove retired lines from wisdom content
124
+ lines = content.splitlines()
125
+ retired_lines = {h["line"] for h in to_retire}
126
+ new_lines = [l for i, l in enumerate(lines) if i not in retired_lines]
127
+ # Clean up any trailing empty lines in sections
128
+ new_content = "\n".join(new_lines).rstrip() + "\n"
129
+ wisdom_path.write_text(new_content, encoding="utf-8")
130
+
131
+ new_size = len(new_content.encode("utf-8"))
132
+ return {
133
+ "status": "retired",
134
+ "original_size": current_size,
135
+ "new_size": new_size,
136
+ "retired_count": len(to_retire),
137
+ "archive_path": str(archive_path),
138
+ }
139
+
140
+
141
+def main(argv: list[str] | None = None) -> int:
142
+ parser = argparse.ArgumentParser(description="Wisdom.md size cap management")
143
+ parser.add_argument("--topic", required=True, help="Topic ID (e.g., ai-ml)")
144
+ parser.add_argument("--limit", type=int, default=DEFAULT_LIMIT, help="Size limit in bytes")
145
+ parser.add_argument("--dry-run", action="store_true", help="Show what would be retired")
146
+ args = parser.parse_args(argv)
147
+
148
+ wisdom_path = get_wisdom_path(args.topic)
149
+ archive_path = get_archive_path(args.topic)
150
+
151
+ result = retire_heuristics(wisdom_path, archive_path, args.limit, dry_run=args.dry_run)
152
+
153
+ if result["status"] == "skip":
154
+ print(f"Skipped: {result['reason']}")
155
+ elif result["status"] == "ok":
156
+ print(result["message"])
157
+ elif result["status"] == "dry_run":
158
+ print(f"DRY RUN: Would retire {result['would_retire']} heuristics")
159
+ for item in result.get("items", []):
160
+ print(f" {item}")
161
+ elif result["status"] == "retired":
162
+ print(f"Retired {result['retired_count']} heuristics")
163
+ print(f" Size: {result['original_size']} -> {result['new_size']} bytes")
164
+ print(f" Archive: {result['archive_path']}")
165
+ else:
166
+ print(f"Warning: {result.get('message', 'unknown status')}")
167
+
168
+ return 0
169
+
170
+
171
+if __name__ == "__main__":
172
+ sys.exit(main())
tests/test_preprocess_analysis.py
new
+179
@@ -0,0 +1,179 @@
1
+"""Tests for scripts/preprocess_for_analysis.py."""
2
+
3
+import json
4
+from datetime import datetime, timezone
5
+from pathlib import Path
6
+
7
+import pytest
8
+
9
+sys_path_fix = True # noqa: E402
10
+import sys
11
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
12
+
13
+from scripts.preprocess_for_analysis import (
14
+ compact_repo,
15
+ compute_age_days,
16
+ compute_signals,
17
+ estimate_tokens,
18
+ preprocess,
19
+ main,
20
+)
21
+
22
+
23
+class TestEstimateTokens:
24
+ def test_basic(self):
25
+ assert estimate_tokens("abcd") == 1
26
+ assert estimate_tokens("a" * 400) == 100
27
+
28
+ def test_empty(self):
29
+ assert estimate_tokens("") == 0
30
+
31
+
32
+class TestComputeAgeDays:
33
+ def test_valid_date(self):
34
+ ref = datetime(2026, 5, 20, tzinfo=timezone.utc)
35
+ assert compute_age_days("2026-05-10T00:00:00Z", ref) == 10
36
+
37
+ def test_none(self):
38
+ assert compute_age_days(None) is None
39
+
40
+ def test_invalid(self):
41
+ assert compute_age_days("not-a-date") is None
42
+
43
+
44
+class TestCompactRepo:
45
+ def test_extracts_needed_fields(self):
46
+ repo = {
47
+ "name": "test-repo",
48
+ "owner": "someone",
49
+ "full_name": "someone/test-repo",
50
+ "description": "A very long description " * 20,
51
+ "language": "Python",
52
+ "stars": 500,
53
+ "forks": 100,
54
+ "created_at": "2026-05-01T00:00:00Z",
55
+ "topics": ["ml", "ai"],
56
+ "license": "MIT",
57
+ "url": "https://github.com/someone/test-repo",
58
+ }
59
+ ref = datetime(2026, 5, 20, tzinfo=timezone.utc)
60
+ result = compact_repo(repo, max_desc=200, reference_date=ref)
61
+
62
+ assert result["name"] == "test-repo"
63
+ assert len(result["desc"]) <= 200
64
+ assert result["stars"] == 500
65
+ assert result["topics"] == ["ml", "ai"]
66
+ assert result["lang"] == "Python"
67
+ assert result["age_days"] == 19
68
+ # Removed fields should not be present
69
+ assert "owner" not in result
70
+ assert "full_name" not in result
71
+ assert "forks" not in result
72
+ assert "license" not in result
73
+ assert "url" not in result
74
+
75
+ def test_handles_missing_fields(self):
76
+ result = compact_repo({}, max_desc=200)
77
+ assert result["name"] == ""
78
+ assert result["desc"] == ""
79
+ assert result["stars"] == 0
80
+
81
+
82
+class TestComputeSignals:
83
+ def test_top_topics(self):
84
+ repos = [
85
+ {"topics": ["ml", "python"]},
86
+ {"topics": ["ml", "ai"]},
87
+ {"topics": ["python"]},
88
+ ]
89
+ signals = compute_signals(repos)
90
+ # ml and python appear twice each
91
+ assert "ml" in signals["top_topics"]
92
+ assert "python" in signals["top_topics"]
93
+
94
+
95
+class TestPreprocess:
96
+ def test_reduction_in_expected_range(self):
97
+ """Token reduction should be 40-60% for typical data."""
98
+ # Build a realistic raw JSON
99
+ repos = []
100
+ for i in range(50):
101
+ repos.append({
102
+ "name": f"repo-{i}",
103
+ "owner": f"owner-{i}",
104
+ "full_name": f"owner-{i}/repo-{i}",
105
+ "description": f"Description for repo {i} with extra detail " * 5,
106
+ "language": "Python",
107
+ "stars": 100 + i * 10,
108
+ "forks": 50 + i,
109
+ "created_at": "2026-05-01T00:00:00Z",
110
+ "topics": ["ml", "deep-learning"],
111
+ "license": "MIT",
112
+ "url": f"https://github.com/owner-{i}/repo-{i}",
113
+ })
114
+ data = {
115
+ "week": "2026-W21",
116
+ "crawled_at": "2026-05-18T08:54:09Z",
117
+ "new_repos": repos,
118
+ "trending_repos": [],
119
+ "signals": {"top_topics": ["ml"]},
120
+ "metadata": {"api_calls_used": 10, "rate_limit_remaining": 50},
121
+ }
122
+
123
+ result = preprocess(data, max_desc=200)
124
+ stats = result["stats"]
125
+ assert 30 <= stats["reduction_pct"] <= 70, (
126
+ f"Reduction {stats['reduction_pct']}% not in expected range"
127
+ )
128
+
129
+ def test_output_structure(self):
130
+ data = {
131
+ "week": "2026-W21",
132
+ "new_repos": [{"name": "x", "description": "hello", "stars": 10,
133
+ "topics": [], "language": "Go", "created_at": "2026-05-01T00:00:00Z"}],
134
+ "trending_repos": [],
135
+ }
136
+ result = preprocess(data)
137
+ assert result["week"] == "2026-W21"
138
+ assert len(result["repos"]) == 1
139
+ assert "signals" in result
140
+ assert "stats" in result
141
+
142
+ def test_deduplicates_repos(self):
143
+ repo = {"name": "dup", "description": "x", "stars": 1, "topics": [],
144
+ "language": "Rust", "created_at": "2026-05-01T00:00:00Z"}
145
+ data = {"week": "2026-W21", "new_repos": [repo], "trending_repos": [repo]}
146
+ result = preprocess(data)
147
+ assert len(result["repos"]) == 1
148
+
149
+
150
+class TestMainCLI:
151
+ def test_end_to_end(self, tmp_path):
152
+ raw = {
153
+ "week": "2026-W21",
154
+ "crawled_at": "2026-05-18T00:00:00Z",
155
+ "new_repos": [
156
+ {"name": "r", "owner": "o", "full_name": "o/r",
157
+ "description": "d" * 300, "language": "Python",
158
+ "stars": 100, "forks": 10, "created_at": "2026-05-01T00:00:00Z",
159
+ "topics": ["ai"], "license": "MIT", "url": "https://github.com/o/r"}
160
+ ],
161
+ "trending_repos": [],
162
+ "signals": {"top_topics": ["ai"]},
163
+ "metadata": {"api_calls_used": 1},
164
+ }
165
+ input_file = tmp_path / "raw.json"
166
+ output_file = tmp_path / "compact.json"
167
+ input_file.write_text(json.dumps(raw))
168
+
169
+ rc = main(["--input", str(input_file), "--output", str(output_file)])
170
+ assert rc == 0
171
+ assert output_file.exists()
172
+
173
+ result = json.loads(output_file.read_text())
174
+ assert result["week"] == "2026-W21"
175
+ assert len(result["repos"][0]["desc"]) <= 200
176
+
177
+ def test_missing_input(self, tmp_path):
178
+ rc = main(["--input", str(tmp_path / "nope.json")])
179
+ assert rc == 1
tests/test_wisdom_cap.py
new
+151
@@ -0,0 +1,151 @@
1
+"""Tests for scripts/wisdom_cap.py."""
2
+
3
+import sys
4
+from pathlib import Path
5
+
6
+import pytest
7
+
8
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
9
+
10
+from scripts.wisdom_cap import (
11
+ get_wisdom_path,
12
+ get_archive_path,
13
+ parse_heuristics,
14
+ select_for_retirement,
15
+ retire_heuristics,
16
+ main,
17
+)
18
+
19
+
20
+SAMPLE_WISDOM = """\
21
+# Topic Wisdom
22
+
23
+## Signal Patterns
24
+- Heuristic one about signals
25
+- Heuristic two about patterns
26
+- Heuristic three about growth
27
+
28
+## Noise Patterns
29
+- Noise heuristic one
30
+- Noise heuristic two
31
+
32
+## Scoring Adjustments
33
+- Scoring rule one
34
+- Scoring rule two
35
+- Scoring rule three
36
+"""
37
+
38
+
39
+class TestParseHeuristics:
40
+ def test_parses_all_bullets(self):
41
+ heuristics = parse_heuristics(SAMPLE_WISDOM)
42
+ assert len(heuristics) == 8
43
+
44
+ def test_captures_sections(self):
45
+ heuristics = parse_heuristics(SAMPLE_WISDOM)
46
+ sections = {h["section"] for h in heuristics}
47
+ assert "Signal Patterns" in sections
48
+ assert "Noise Patterns" in sections
49
+
50
+ def test_empty_content(self):
51
+ assert parse_heuristics("") == []
52
+ assert parse_heuristics("# Just a header\n") == []
53
+
54
+
55
+class TestSelectForRetirement:
56
+ def test_selects_from_end(self):
57
+ heuristics = parse_heuristics(SAMPLE_WISDOM)
58
+ retired = select_for_retirement(heuristics, bytes_to_free=50)
59
+ # Should select from the end
60
+ assert retired[0]["text"] == "- Scoring rule three"
61
+
62
+ def test_respects_bytes_needed(self):
63
+ heuristics = parse_heuristics(SAMPLE_WISDOM)
64
+ retired = select_for_retirement(heuristics, bytes_to_free=1)
65
+ assert len(retired) >= 1
66
+
67
+
68
+class TestRetireHeuristics:
69
+ def test_under_limit_no_action(self, tmp_path):
70
+ wisdom = tmp_path / "wisdom.md"
71
+ archive = tmp_path / "archive.md"
72
+ wisdom.write_text(SAMPLE_WISDOM)
73
+
74
+ result = retire_heuristics(wisdom, archive, limit=10000)
75
+ assert result["status"] == "ok"
76
+ assert not archive.exists()
77
+
78
+ def test_over_limit_retires(self, tmp_path):
79
+ wisdom = tmp_path / "wisdom.md"
80
+ archive = tmp_path / "archive.md"
81
+ wisdom.write_text(SAMPLE_WISDOM)
82
+ # Set limit below current size to trigger retirement
83
+ current_size = len(SAMPLE_WISDOM.encode("utf-8"))
84
+ limit = current_size - 50
85
+
86
+ result = retire_heuristics(wisdom, archive, limit=limit)
87
+ assert result["status"] == "retired"
88
+ assert result["retired_count"] >= 1
89
+ assert archive.exists()
90
+ # Wisdom should be smaller now
91
+ new_size = len(wisdom.read_text().encode("utf-8"))
92
+ assert new_size < current_size
93
+
94
+ def test_dry_run_no_changes(self, tmp_path):
95
+ wisdom = tmp_path / "wisdom.md"
96
+ archive = tmp_path / "archive.md"
97
+ wisdom.write_text(SAMPLE_WISDOM)
98
+ limit = 50 # Way under to force retirement
99
+
100
+ result = retire_heuristics(wisdom, archive, limit=limit, dry_run=True)
101
+ assert result["status"] == "dry_run"
102
+ assert result["would_retire"] > 0
103
+ # File should be unchanged
104
+ assert wisdom.read_text() == SAMPLE_WISDOM
105
+ assert not archive.exists()
106
+
107
+ def test_missing_wisdom(self, tmp_path):
108
+ wisdom = tmp_path / "nope.md"
109
+ archive = tmp_path / "archive.md"
110
+ result = retire_heuristics(wisdom, archive, limit=5120)
111
+ assert result["status"] == "skip"
112
+
113
+ def test_idempotent(self, tmp_path):
114
+ """Running twice produces same result if already under limit."""
115
+ wisdom = tmp_path / "wisdom.md"
116
+ archive = tmp_path / "archive.md"
117
+ wisdom.write_text(SAMPLE_WISDOM)
118
+ limit = len(SAMPLE_WISDOM.encode("utf-8")) - 50
119
+
120
+ retire_heuristics(wisdom, archive, limit=limit)
121
+ content_after_first = wisdom.read_text()
122
+
123
+ # Running again should be ok (under limit now)
124
+ result = retire_heuristics(wisdom, archive, limit=limit)
125
+ assert result["status"] == "ok"
126
+ assert wisdom.read_text() == content_after_first
127
+
128
+ def test_archive_appends(self, tmp_path):
129
+ """Multiple retirements append to archive, never overwrite."""
130
+ wisdom = tmp_path / "wisdom.md"
131
+ archive = tmp_path / "archive.md"
132
+ wisdom.write_text(SAMPLE_WISDOM)
133
+
134
+ # First retirement
135
+ retire_heuristics(wisdom, archive, limit=50)
136
+ first_archive = archive.read_text()
137
+
138
+ # Add more content and retire again
139
+ current = wisdom.read_text()
140
+ wisdom.write_text(current + "- New heuristic added\n" * 20)
141
+ retire_heuristics(wisdom, archive, limit=50)
142
+ second_archive = archive.read_text()
143
+
144
+ assert len(second_archive) > len(first_archive)
145
+
146
+
147
+class TestMainCLI:
148
+ def test_runs_with_missing_topic(self, tmp_path, monkeypatch):
149
+ monkeypatch.setattr("scripts.wisdom_cap.SQUAD_DIR", tmp_path / ".squad" / "topics")
150
+ rc = main(["--topic", "nonexistent"])
151
+ assert rc == 0 # graceful skip