Coverage for little_loops / cli / issues / decisions.py: 0%

1"""ll-issues decisions: Manage rules, decisions, and exceptions log.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import sys 

7from datetime import UTC, datetime 

8from typing import TYPE_CHECKING 

9 

10if TYPE_CHECKING: 

11 from little_loops.config import BRConfig 

12 

13 

14def add_decisions_parser(subs: argparse._SubParsersAction) -> argparse.ArgumentParser: 

15 """Register the decisions subparser with all sub-sub-commands on *subs*.""" 

16 from little_loops.cli_args import add_config_arg 

17 

18 p = subs.add_parser( 

19 "decisions", 

20 help="Manage rules, decisions, and exceptions log", 

21 ) 

22 p.set_defaults(command="decisions") 

23 p.set_defaults(_decisions_parser=p) 

24 

25 add_config_arg(p) 

26 subsubs = p.add_subparsers(dest="subcommand") 

27 

28 # -- list -- 

29 list_p = subsubs.add_parser("list", help="List decisions log entries") 

30 list_p.add_argument( 

31 "--type", 

32 choices=["rule", "decision", "exception", "coupling"], 

33 default=None, 

34 help="Filter by entry type", 

35 ) 

36 list_p.add_argument("--category", default=None, help="Filter by category") 

37 list_p.add_argument("--label", default=None, help="Filter by label") 

38 list_p.add_argument( 

39 "--archetype", 

40 default=None, 

41 help="Filter coupling entries by archetype label (e.g. add-cli-command)", 

42 ) 

43 list_p.add_argument( 

44 "--no-outcome", 

45 action="store_true", 

46 help="Show only DecisionEntry records with no recorded outcome", 

47 ) 

48 list_p.add_argument( 

49 "--before", 

50 default=None, 

51 metavar="ISO-8601", 

52 help="Show entries with timestamp before this date", 

53 ) 

54 list_p.add_argument( 

55 "--scope", 

56 default=None, 

57 help="Filter DecisionEntry records by scope (e.g. 'issue', 'project')", 

58 ) 

59 list_p.add_argument( 

60 "--format", 

61 "-f", 

62 choices=["text", "json"], 

63 default="text", 

64 help="Output format (default: text)", 

65 ) 

66 list_p.add_argument( 

67 "--active-only", 

68 action="store_true", 

69 help="Exclude entries superseded by a newer entry", 

70 ) 

71 add_config_arg(list_p) 

72 

73 # -- add -- 

74 add_p = subsubs.add_parser("add", help="Add a new entry to the decisions log") 

75 add_p.add_argument( 

76 "--type", 

77 required=True, 

78 choices=["rule", "decision", "exception", "coupling"], 

79 help="Entry type", 

80 ) 

81 add_p.add_argument("--category", required=True, help="Category (e.g. 'architecture')") 

82 add_p.add_argument( 

83 "--rule", 

84 default=None, 

85 help="Rule or decision text (required for type=rule or type=decision)", 

86 ) 

87 add_p.add_argument( 

88 "--rationale", 

89 required=True, 

90 help="Why this rule/decision/exception applies", 

91 ) 

92 add_p.add_argument("--issue", default=None, help="Related issue ID (e.g. FEAT-1894)") 

93 add_p.add_argument("--label", default=None, help="Comma-separated labels") 

94 add_p.add_argument( 

95 "--enforcement", 

96 choices=["required", "advisory"], 

97 default="advisory", 

98 help="Enforcement level for type=rule (default: advisory)", 

99 ) 

100 add_p.add_argument( 

101 "--rule-ref", 

102 default=None, 

103 dest="rule_ref", 

104 metavar="RULE-ID", 

105 help="Reference to the rule being excepted (required for type=exception)", 

106 ) 

107 add_p.add_argument( 

108 "--alternatives-rejected", 

109 default=None, 

110 dest="alternatives_rejected", 

111 help="Alternatives considered and rejected (for type=decision or type=exception)", 

112 ) 

113 add_p.add_argument( 

114 "--supersedes", 

115 default=None, 

116 metavar="ENTRY-ID", 

117 help="ID of the entry this one supersedes (for type=rule)", 

118 ) 

119 add_p.add_argument( 

120 "--scope", 

121 default="issue", 

122 help="Decision scope: 'issue' (default) or 'project' (for type=decision)", 

123 ) 

124 add_p.add_argument( 

125 "--id", 

126 default=None, 

127 dest="entry_id", 

128 metavar="ENTRY-ID", 

129 help="Explicit entry ID (auto-generated if omitted)", 

130 ) 

131 # Coupling-specific flags (only relevant when --type=coupling) 

132 add_p.add_argument( 

133 "--if-changed", 

134 default=None, 

135 dest="if_changed", 

136 metavar="GLOB", 

137 help="Glob pattern matching files being modified (required for type=coupling)", 

138 ) 

139 add_p.add_argument( 

140 "--then-check", 

141 default=None, 

142 dest="then_check", 

143 metavar="PATHS", 

144 help="Comma-separated file/pattern list to audit for wiring gaps (required for type=coupling)", 

145 ) 

146 add_p.add_argument( 

147 "--tier", 

148 choices=["hard", "soft", "fyi"], 

149 default="soft", 

150 help="Coupling tier: hard (must change together), soft (should update), fyi (informational)", 

151 ) 

152 add_p.add_argument( 

153 "--archetype", 

154 default=None, 

155 dest="archetype", 

156 help="Named bundle label grouping related coupling rules (e.g. add-cli-command)", 

157 ) 

158 add_config_arg(add_p) 

159 

160 # -- outcome -- 

161 outcome_p = subsubs.add_parser("outcome", help="Record the outcome of a decision entry") 

162 outcome_p.add_argument("id", help="Entry ID to record outcome for") 

163 outcome_p.add_argument( 

164 "--result", 

165 required=True, 

166 choices=["worked", "did_not_work", "mixed", "reversed"], 

167 help="Outcome result", 

168 ) 

169 outcome_p.add_argument("--notes", default=None, help="Free-text notes about the outcome") 

170 outcome_p.add_argument( 

171 "--measured-at", 

172 default=None, 

173 dest="measured_at", 

174 metavar="ISO-8601", 

175 help="When the outcome was measured (default: now)", 

176 ) 

177 outcome_p.add_argument( 

178 "--force", 

179 action="store_true", 

180 help="Overwrite an existing outcome", 

181 ) 

182 add_config_arg(outcome_p) 

183 

184 # -- generate (stub) -- 

185 gen_p = subsubs.add_parser( 

186 "generate", 

187 help="Generate decisions log entries from completed issues", 

188 ) 

189 gen_p.add_argument( 

190 "--from", 

191 dest="source", 

192 default="completed", 

193 choices=["completed"], 

194 help="Source to generate from (default: completed)", 

195 ) 

196 add_config_arg(gen_p) 

197 

198 # -- sync (stub) -- 

199 sync_p = subsubs.add_parser("sync", help="Sync active rules to .ll/ll.local.md") 

200 add_config_arg(sync_p) 

201 

202 # -- suggest-rules -- 

203 suggest_p = subsubs.add_parser( 

204 "suggest-rules", 

205 help="Suggest decision entries that are candidates for promotion to rules", 

206 ) 

207 add_config_arg(suggest_p) 

208 

209 # -- extract-from-completed -- 

210 extract_p = subsubs.add_parser( 

211 "extract-from-completed", 

212 help="Extract decisions and rules from completed issues via LLM", 

213 ) 

214 extract_p.add_argument( 

215 "--since", 

216 metavar="YYYY-MM-DD", 

217 default=None, 

218 help="Only process issues completed on or after this date", 

219 ) 

220 extract_p.add_argument( 

221 "--issue", 

222 metavar="ID", 

223 default=None, 

224 help="Only extract from this specific issue ID (e.g. ENH-2151)", 

225 ) 

226 extract_p.add_argument( 

227 "--dry-run", 

228 action="store_true", 

229 default=False, 

230 help="Print candidates without writing to decisions.yaml", 

231 ) 

232 extract_p.add_argument( 

233 "--min-confidence", 

234 type=float, 

235 default=0.7, 

236 dest="min_confidence", 

237 metavar="FLOAT", 

238 help="Minimum LLM confidence to accept a candidate (default: 0.7)", 

239 ) 

240 add_config_arg(extract_p) 

241 

242 # -- promote -- 

243 promote_p = subsubs.add_parser( 

244 "promote", 

245 help="Convert a decision entry to a rule", 

246 ) 

247 promote_p.add_argument("entry_id", help="ID of the decision entry to promote") 

248 promote_p.add_argument( 

249 "--enforcement", 

250 choices=["required", "advisory"], 

251 default="required", 

252 help="Enforcement level for the resulting rule (default: required)", 

253 ) 

254 add_config_arg(promote_p) 

255 

256 return p 

257 

258 

259def cmd_decisions(config: BRConfig, args: argparse.Namespace) -> int: 

260 """Dispatch decisions sub-sub-commands; returns 0 on success, 1 on error.""" 

261 from pathlib import Path 

262 

263 from little_loops.decisions import ( 

264 CouplingEntry, 

265 DecisionEntry, 

266 ExceptionEntry, 

267 RuleEntry, 

268 add_entry, 

269 list_entries, 

270 resolve_active, 

271 set_outcome, 

272 ) 

273 

274 if not getattr(args, "subcommand", None): 

275 args._decisions_parser.print_help() 

276 return 1 

277 

278 # Resolve the decisions log path from config 

279 path = Path(config.project_root) / config.decisions.log_path 

280 

281 sub = args.subcommand 

282 

283 if sub == "list": 

284 return _cmd_list(args, path, list_entries, resolve_active) 

285 

286 if sub == "add": 

287 return _cmd_add( 

288 args, 

289 path, 

290 RuleEntry, 

291 DecisionEntry, 

292 ExceptionEntry, 

293 CouplingEntry, 

294 add_entry, 

295 list_entries, 

296 ) 

297 

298 if sub == "outcome": 

299 return _cmd_outcome(args, path, set_outcome) 

300 

301 if sub == "generate": 

302 from little_loops.decisions import generate_from_completed 

303 

304 count = generate_from_completed(config) 

305 print( 

306 f"Generated {count} decision entr{'y' if count == 1 else 'ies'} from completed issues." 

307 ) 

308 return 0 

309 

310 if sub == "sync": 

311 return _cmd_sync(path) 

312 

313 if sub == "suggest-rules": 

314 return _cmd_suggest_rules(path) 

315 

316 if sub == "extract-from-completed": 

317 return _cmd_extract_from_completed(config, args, path) 

318 

319 if sub == "promote": 

320 from little_loops.decisions import DecisionEntry, RuleEntry, load_decisions, save_decisions 

321 

322 return _cmd_promote(args, path, load_decisions, save_decisions, RuleEntry, DecisionEntry) 

323 

324 print(f"Unknown subcommand: {sub!r}", file=sys.stderr) 

325 return 1 

326 

327 

328def _cmd_list(args, path, list_entries, resolve_active) -> int: 

329 import json 

330 

331 entries = list_entries( 

332 path=path, 

333 type=getattr(args, "type", None), 

334 category=getattr(args, "category", None), 

335 label=getattr(args, "label", None), 

336 ) 

337 

338 # Post-filters 

339 if getattr(args, "no_outcome", False): 

340 from little_loops.decisions import DecisionEntry 

341 

342 entries = [e for e in entries if isinstance(e, DecisionEntry) and e.outcome is None] 

343 

344 if getattr(args, "before", None): 

345 entries = [e for e in entries if e.timestamp < args.before] 

346 

347 if getattr(args, "scope", None): 

348 from little_loops.decisions import DecisionEntry 

349 

350 entries = [e for e in entries if isinstance(e, DecisionEntry) and e.scope == args.scope] 

351 

352 if getattr(args, "active_only", False): 

353 entries = resolve_active(entries) 

354 

355 if getattr(args, "archetype", None): 

356 from little_loops.decisions import CouplingEntry as _CE 

357 

358 entries = [e for e in entries if isinstance(e, _CE) and e.archetype == args.archetype] 

359 

360 fmt = getattr(args, "format", "text") or "text" 

361 if fmt == "json": 

362 print(json.dumps([e.to_dict() for e in entries], indent=2)) 

363 return 0 

364 

365 if not entries: 

366 print("(no entries)") 

367 return 0 

368 

369 for entry in entries: 

370 _print_entry(entry) 

371 return 0 

372 

373 

374def _print_entry(entry) -> None: 

375 from little_loops.decisions import CouplingEntry, DecisionEntry, ExceptionEntry, RuleEntry 

376 

377 label_str = ", ".join(entry.labels) if entry.labels else "" 

378 print( 

379 f"{entry.id} [{entry.type}] {entry.category}" + (f" ({label_str})" if label_str else "") 

380 ) 

381 if isinstance(entry, (RuleEntry, DecisionEntry)): 

382 print(f" rule: {entry.rule}") 

383 elif isinstance(entry, ExceptionEntry): 

384 print(f" rule_ref: {entry.rule_ref}") 

385 elif isinstance(entry, CouplingEntry): 

386 print(f" if_changed: {entry.if_changed}") 

387 print(f" then_check: {', '.join(entry.then_check)}") 

388 print(f" tier: {entry.tier}") 

389 if entry.archetype: 

390 print(f" archetype: {entry.archetype}") 

391 if entry.rationale: 

392 print(f" rationale: {entry.rationale}") 

393 if isinstance(entry, DecisionEntry) and entry.outcome: 

394 print(f" outcome: {entry.outcome.result} @ {entry.outcome.measured_at}") 

395 

396 

397def _cmd_add( 

398 args, path, RuleEntry, DecisionEntry, ExceptionEntry, CouplingEntry, add_entry, list_entries 

399) -> int: 

400 entry_type = args.type 

401 

402 # Validate type-specific required fields 

403 if entry_type == "rule" and not getattr(args, "rule", None): 

404 print("Error: --rule is required for type 'rule'", file=sys.stderr) 

405 return 1 

406 if entry_type == "decision" and not getattr(args, "rule", None): 

407 print("Error: --rule is required for type 'decision' (the decision text)", file=sys.stderr) 

408 return 1 

409 if entry_type == "exception" and not getattr(args, "rule_ref", None): 

410 print("Error: --rule-ref is required for type 'exception'", file=sys.stderr) 

411 return 1 

412 if entry_type == "coupling": 

413 if not getattr(args, "if_changed", None): 

414 print("Error: --if-changed is required for type 'coupling'", file=sys.stderr) 

415 return 1 

416 if not getattr(args, "then_check", None): 

417 print("Error: --then-check is required for type 'coupling'", file=sys.stderr) 

418 return 1 

419 

420 # Generate an ID if not provided 

421 entry_id = getattr(args, "entry_id", None) 

422 if not entry_id: 

423 existing = list_entries(path=path, type=entry_type) 

424 prefix = { 

425 "rule": args.category.upper() if args.category else "RULE", 

426 "decision": args.category.upper() if args.category else "DEC", 

427 "exception": args.category.upper() + "-EX" if args.category else "EX", 

428 "coupling": "COUPLING", 

429 }[entry_type] 

430 entry_id = f"{prefix}-{len(existing) + 1:03d}" 

431 

432 timestamp = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") 

433 labels = [lbl.strip() for lbl in args.label.split(",")] if getattr(args, "label", None) else [] 

434 

435 if entry_type == "rule": 

436 entry = RuleEntry( 

437 id=entry_id, 

438 timestamp=timestamp, 

439 category=args.category, 

440 labels=labels, 

441 rationale=args.rationale, 

442 rule=args.rule, 

443 enforcement=getattr(args, "enforcement", "advisory"), 

444 supersedes=getattr(args, "supersedes", None), 

445 issue=getattr(args, "issue", None), 

446 ) 

447 elif entry_type == "decision": 

448 entry = DecisionEntry( 

449 id=entry_id, 

450 timestamp=timestamp, 

451 category=args.category, 

452 labels=labels, 

453 rationale=args.rationale, 

454 rule=args.rule, 

455 alternatives_rejected=getattr(args, "alternatives_rejected", None), 

456 issue=getattr(args, "issue", None), 

457 scope=getattr(args, "scope", "issue"), 

458 ) 

459 elif entry_type == "coupling": 

460 then_check = [t.strip() for t in args.then_check.split(",")] 

461 entry = CouplingEntry( 

462 id=entry_id, 

463 timestamp=timestamp, 

464 category=args.category, 

465 labels=labels, 

466 rationale=args.rationale, 

467 if_changed=args.if_changed, 

468 then_check=then_check, 

469 tier=getattr(args, "tier", "soft"), 

470 archetype=getattr(args, "archetype", None), 

471 enforcement=getattr(args, "enforcement", "advisory"), 

472 issue=getattr(args, "issue", None), 

473 ) 

474 else: # exception 

475 entry = ExceptionEntry( 

476 id=entry_id, 

477 timestamp=timestamp, 

478 category=args.category, 

479 labels=labels, 

480 rationale=args.rationale, 

481 rule_ref=args.rule_ref, 

482 issue=getattr(args, "issue", "") or "", 

483 alternatives_rejected=getattr(args, "alternatives_rejected", None), 

484 ) 

485 

486 add_entry(entry, path=path) 

487 print(f"Added {entry_type} entry: {entry_id}") 

488 return 0 

489 

490 

491def _cmd_outcome(args, path, set_outcome) -> int: 

492 measured_at = getattr(args, "measured_at", None) or datetime.now(UTC).strftime( 

493 "%Y-%m-%dT%H:%M:%SZ" 

494 ) 

495 try: 

496 set_outcome( 

497 args.id, 

498 args.result, 

499 measured_at, 

500 notes=getattr(args, "notes", None), 

501 path=path, 

502 force=getattr(args, "force", False), 

503 ) 

504 except KeyError: 

505 print(f"Error: entry {args.id!r} not found", file=sys.stderr) 

506 return 1 

507 except TypeError as exc: 

508 print(f"Error: {exc}", file=sys.stderr) 

509 return 1 

510 except ValueError as exc: 

511 print(f"Error: {exc} (use --force to overwrite)", file=sys.stderr) 

512 return 1 

513 print(f"Recorded outcome for {args.id}: {args.result}") 

514 return 0 

515 

516 

517def _cmd_sync(path) -> int: 

518 try: 

519 from little_loops.decisions_sync import sync_to_local_md 

520 

521 sync_to_local_md(path=path) 

522 except ImportError: 

523 print("sync not yet available (requires FEAT-1895)", file=sys.stderr) 

524 return 1 

525 return 0 

526 

527 

528def _cmd_suggest_rules(path) -> int: 

529 """Analyze decision entries and suggest candidates for promotion to rules.""" 

530 import re 

531 from collections import defaultdict 

532 

533 from little_loops.decisions import DecisionEntry, list_entries 

534 

535 all_entries = list_entries(path=path) 

536 decisions = [e for e in all_entries if isinstance(e, DecisionEntry)] 

537 

538 if len(decisions) < 3: 

539 count = len(decisions) 

540 noun = "entry" if count == 1 else "entries" 

541 print(f"(only {count} decision {noun} found; need at least 3 to suggest rule candidates)") 

542 return 1 

543 

544 # Heuristic: rule texts that are one-off choices rather than general constraints 

545 _one_off_re = re.compile(r"^(Option [A-C]\b|NO-GO\b|Captured:\s*)", re.IGNORECASE) 

546 

547 def _is_general_constraint(e: DecisionEntry) -> bool: 

548 return not _one_off_re.match(e.rule or "") 

549 

550 def _extract_tokens(text: str) -> set[str]: 

551 """Extract snake_case and kebab-case identifiers (min 5 chars) and file paths.""" 

552 tokens: set[str] = set() 

553 for m in re.finditer(r"\b([a-z][a-z0-9]*(?:[_-][a-z0-9]+)+)\b", text): 

554 t = m.group(1).replace("-", "_") 

555 if len(t) >= 5: 

556 tokens.add(t) 

557 for m in re.finditer(r"([a-z][a-z0-9_/.-]+\.(?:py|yaml|md|json))", text): 

558 tokens.add(m.group(1)) 

559 return tokens 

560 

561 by_category: dict[str, list[DecisionEntry]] = defaultdict(list) 

562 for d in decisions: 

563 by_category[d.category].append(d) 

564 

565 clusters: list[tuple[list[DecisionEntry], str, list[str], bool]] = [] 

566 

567 for category in sorted(by_category): 

568 cat_entries = by_category[category] 

569 high_signal = len(cat_entries) >= 3 

570 general = [e for e in cat_entries if _is_general_constraint(e)] 

571 

572 if not general: 

573 continue 

574 

575 if high_signal: 

576 if len(general) >= 2: 

577 # Find tokens shared by at least 2 entries in this high-signal category 

578 tok_freq: dict[str, int] = {} 

579 for e in general: 

580 seen_in_entry: set[str] = set() 

581 for t in _extract_tokens((e.rationale or "") + " " + (e.rule or "")): 

582 if t not in seen_in_entry: 

583 tok_freq[t] = tok_freq.get(t, 0) + 1 

584 seen_in_entry.add(t) 

585 shared = sorted(t for t, c in tok_freq.items() if c >= 2)[:3] 

586 clusters.append((general, category, shared, True)) 

587 else: 

588 # Single general constraint in a high-signal category — still worth surfacing 

589 clusters.append((general, category, [], True)) 

590 else: 

591 # Low-signal category: only pair entries that share common tokens 

592 tok: dict[str, set[str]] = { 

593 e.id: _extract_tokens((e.rationale or "") + " " + (e.rule or "")) for e in general 

594 } 

595 used: set[str] = set() 

596 for i, e1 in enumerate(general): 

597 if e1.id in used: 

598 continue 

599 group = [e1] 

600 for e2 in general[i + 1 :]: 

601 if e2.id in used: 

602 continue 

603 if tok[e1.id] & tok[e2.id]: 

604 group.append(e2) 

605 used.add(e2.id) 

606 if len(group) >= 2: 

607 used.add(e1.id) 

608 shared_tokens = tok[e1.id].copy() 

609 for e in group[1:]: 

610 shared_tokens &= tok[e.id] 

611 clusters.append((group, category, sorted(shared_tokens)[:3], False)) 

612 

613 if not clusters: 

614 print("(no rule candidates found in current decisions log)") 

615 return 1 

616 

617 for group, category, shared, high_signal in clusters: 

618 ids = ", ".join(e.id for e in group) 

619 ref_str = " and reference " + ", ".join(shared) if shared else "" 

620 signal_str = " [high-signal]" if high_signal else "" 

621 best = max(group, key=lambda e: len(e.rule or "")) 

622 

623 print(f"[SUGGEST]{signal_str} {ids} share category={category}{ref_str}") 

624 print(f' — consider promoting to a rule: "{best.rule}"') 

625 for e in group: 

626 print(f" • {e.id}: {e.rule}") 

627 print() 

628 

629 return 0 

630 

631 

632def _is_near_duplicate(rule_text: str, existing_rules: list[str]) -> bool: 

633 """Return True if rule_text shares ≥60% significant tokens with any existing rule.""" 

634 import re 

635 

636 def _tokens(text: str) -> set[str]: 

637 return {w.lower() for w in re.findall(r"\b[a-zA-Z]{4,}\b", text)} 

638 

639 new_tokens = _tokens(rule_text) 

640 if not new_tokens: 

641 return False 

642 

643 for existing in existing_rules: 

644 overlap = new_tokens & _tokens(existing) 

645 if overlap and len(overlap) / len(new_tokens) >= 0.6: 

646 return True 

647 

648 return False 

649 

650 

651_EXTRACTION_SCHEMA: dict = { 

652 "type": "object", 

653 "properties": { 

654 "candidates": { 

655 "type": "array", 

656 "items": { 

657 "type": "object", 

658 "properties": { 

659 "rule": {"type": "string"}, 

660 "rationale": {"type": "string"}, 

661 "category": { 

662 "type": "string", 

663 "enum": [ 

664 "architecture", 

665 "testing", 

666 "style", 

667 "security", 

668 "performance", 

669 "other", 

670 ], 

671 }, 

672 "confidence": {"type": "number"}, 

673 "scope": {"type": "string", "enum": ["global", "issue"]}, 

674 }, 

675 "required": ["rule", "rationale", "category", "confidence", "scope"], 

676 }, 

677 } 

678 }, 

679 "required": ["candidates"], 

680} 

681 

682 

683def _cmd_extract_from_completed(config, args, path) -> int: 

684 """Extract rules from completed issues via LLM and append to decisions.yaml.""" 

685 import json 

686 import subprocess 

687 from datetime import UTC, date, datetime 

688 from pathlib import Path 

689 

690 from little_loops.decisions import RuleEntry, add_entry, list_entries 

691 from little_loops.host_runner import resolve_host 

692 from little_loops.issue_history.parsing import scan_completed_issues 

693 

694 project_root = Path(config.project_root) 

695 issues_dir = project_root / config.issues.base_dir 

696 

697 since_date_str = getattr(args, "since", None) 

698 issue_filter = getattr(args, "issue", None) 

699 dry_run = getattr(args, "dry_run", False) 

700 min_confidence = float(getattr(args, "min_confidence", 0.7)) 

701 

702 # Filesystem scan always used (body content required for extraction) 

703 completed = scan_completed_issues(issues_dir) 

704 

705 if issue_filter: 

706 completed = [c for c in completed if c.issue_id == issue_filter] 

707 

708 if since_date_str: 

709 try: 

710 cutoff = date.fromisoformat(since_date_str) 

711 except ValueError: 

712 print( 

713 f"Error: invalid --since date {since_date_str!r}; expected YYYY-MM-DD", 

714 file=sys.stderr, 

715 ) 

716 return 1 

717 completed = [c for c in completed if c.completed_date and c.completed_date >= cutoff] 

718 

719 if not completed: 

720 print("No completed issues found matching filters.") 

721 return 0 

722 

723 # Load existing entries for deduplication 

724 existing = list_entries(path) 

725 existing_issue_ids = {e.issue for e in existing if getattr(e, "issue", None)} 

726 existing_rule_texts = [ 

727 e.rule.lower() # type: ignore[union-attr] 

728 for e in existing 

729 if getattr(e, "rule", None) and isinstance(e.rule, str) # type: ignore[union-attr] 

730 ] 

731 

732 timestamp = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") 

733 

734 total_candidates = 0 

735 total_written = 0 

736 total_skipped = 0 

737 

738 for issue in completed: 

739 # Level-1 dedup: skip if any entry already references this issue 

740 if issue.issue_id in existing_issue_ids: 

741 total_skipped += 1 

742 continue 

743 

744 if not issue.path or not issue.path.exists(): 

745 continue 

746 

747 body = issue.path.read_text(encoding="utf-8") 

748 

749 prompt = ( 

750 f"Issue: {issue.issue_id}\n" 

751 f"Type: {issue.issue_type}\n" 

752 f"Body:\n{body[:4000]}\n\n" 

753 "Extract any decisions, constraints, or rules that should guide future coding agents.\n" 

754 "Only extract rules general enough to apply to future work (not one-off tactical decisions).\n" 

755 "For each candidate return:\n" 

756 " - rule: imperative sentence, ≤ 120 chars\n" 

757 " - rationale: why this rule exists, drawn from the issue context\n" 

758 " - category: architecture | testing | style | security | performance | other\n" 

759 " - confidence: 0.0–1.0 (how likely this generalizes beyond this one issue)\n" 

760 " - scope: global | issue\n" 

761 "Return an empty candidates array if no generalizable rules can be extracted." 

762 ) 

763 

764 invocation = resolve_host().build_blocking_json(prompt=prompt, model="sonnet") 

765 llm_args = list(invocation.args) + [ 

766 "--json-schema", 

767 json.dumps(_EXTRACTION_SCHEMA), 

768 "--no-session-persistence", 

769 ] 

770 

771 try: 

772 proc = subprocess.run( 

773 [invocation.binary, *llm_args], 

774 capture_output=True, 

775 text=True, 

776 timeout=120, 

777 ) 

778 except subprocess.TimeoutExpired: 

779 print(f"Warning: LLM call timed out for {issue.issue_id}", file=sys.stderr) 

780 continue 

781 except FileNotFoundError: 

782 print(f"Error: host CLI not found ({invocation.binary})", file=sys.stderr) 

783 return 1 

784 

785 if proc.returncode != 0: 

786 print( 

787 f"Warning: LLM call failed for {issue.issue_id}: {proc.stderr[:200]}",