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

1"""Canonical Markdown renderers for ship artifacts. 

2 

3These helpers keep public GitHub artifacts deterministic and consumer-neutral. 

4Adapters should post the rendered bodies verbatim instead of hand-writing PR 

5descriptions, review verdicts, jury verdicts, or extension result summaries. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import Any 

11 

12from . import evidence 

13 

14SCHEMA_VERSION = "keel.artifacts.v1" 

15EXTENSION_RESULT_MARKER = "<!-- keel.extension-result.v1 -->" 

16ISSUE_UPDATE_MARKER = "<!-- keel.issue-update.v1 -->" 

17STEP_HANDOFF_MARKER = "<!-- keel.step-handoff.v1 -->" 

18RUN_CONTROL_HALT_MARKER = "<!-- keel.run-control-halt.v1 -->" 

19REVIEW_CYCLE_SUMMARY_MARKER = "keel.review-cycle-summary.v1" 

20COVERAGE_DELTA_MARKER = "keel.coverage-delta.v1" 

21DEPS_AUDIT_MARKER = "keel.deps-audit.v1" 

22FLAKE_AUDIT_MARKER = "keel.flake-audit.v1" 

23SCAN_FINDING_MARKER = "keel.scan-finding.v1" 

24TRIAGE_AUDIT_MARKER = "keel.triage-audit.v1" 

25 

26#: Severity buckets that drive the consolidated histogram + merge recommendation, 

27#: in must-fix → advisory order. ``critical`` folds into ``blocker`` (must-fix). 

28SEVERITY_ORDER = ("blocker", "major", "minor", "nit") 

29_SEVERITY_ALIASES = {"critical": "blocker"} 

30 

31#: Dependency-advisory severity buckets, most → least severe. 

32DEPS_SEVERITY_ORDER = ("critical", "high", "moderate", "low") 

33 

34 

35def contract_as_dict() -> dict[str, Any]: 

36 """Return the canonical artifact renderer contract for ship-like flows.""" 

37 return { 

38 "schema_version": SCHEMA_VERSION, 

39 "consumer_neutral": True, 

40 "deterministic": True, 

41 "renderers": { 

42 "pr_body": "keel.artifacts.render_pr_body", 

43 "issue_update": "keel.artifacts.render_issue_update", 

44 "review_verdict": "keel.artifacts.render_review_verdict", 

45 "jury_verdict": "keel.artifacts.render_jury_verdict", 

46 "review_cycle_summary": "keel.artifacts.render_review_cycle_summary", 

47 "extension_result": "keel.artifacts.render_extension_result", 

48 "step_handoff": "keel.artifacts.render_step_handoff", 

49 "run_control_halt": "keel.artifacts.render_run_control_halt", 

50 }, 

51 "markers": { 

52 "review_verdict": evidence.REVIEW_VERDICT_MARKER, 

53 "jury_verdict": evidence.JURY_VERDICT_MARKER, 

54 "review_cycle_summary": REVIEW_CYCLE_SUMMARY_MARKER, 

55 "issue_update": ISSUE_UPDATE_MARKER, 

56 "extension_result": EXTENSION_RESULT_MARKER, 

57 "step_handoff": STEP_HANDOFF_MARKER, 

58 "run_control_halt": RUN_CONTROL_HALT_MARKER, 

59 }, 

60 "adapter_rule": "post rendered markdown verbatim when available", 

61 } 

62 

63 

64def render_pr_body( 

65 *, 

66 issue_number: int | None = None, 

67 issue_intake: dict[str, Any] | None = None, 

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

69 testing: list[str] | tuple[str, ...] = (), 

70 docs_impact: str | None = None, 

71) -> str: 

72 """Render the canonical PR body used by ship implementers.""" 

73 intake = issue_intake if isinstance(issue_intake, dict) else {} 

74 lines = [ 

75 "## Summary", 

76 f"- { _value(intake.get('deliverable'), 'Implement the requested change.') }", 

77 "", 

78 "## Context / Root Cause", 

79 _value(intake.get("objective"), "See the linked issue for context."), 

80 "", 

81 "## Changes Made", 

82 ] 

83 # `None` is "git could not be read", which must not render as "nothing changed". 

84 if changed_files is None: 

85 lines.append("- The changed-file list could not be read from git.") 

86 else: 

87 files = [file for file in changed_files if isinstance(file, str)] 

88 if files: 

89 lines.extend(f"- Updated `{file}`." for file in files) 

90 else: 

91 lines.append("- No changed files recorded yet.") 

92 lines.extend(["", "## Testing"]) 

93 tests = [item for item in testing if isinstance(item, str) and item.strip()] 

94 lines.extend(f"- {item.strip()}" for item in tests) if tests else lines.append( 

95 "- Not run yet; update this section before marking the PR ready." 

96 ) 

97 lines.extend([ 

98 "", 

99 "## Docs Impact", 

100 _value(docs_impact, "Docs Impact: none — no operator-facing behavior changed."), 

101 "", 

102 _closing_reference(issue_number), 

103 ]) 

104 return "\n".join(lines).rstrip() + "\n" 

105 

106 

107def render_issue_update( 

108 *, 

109 issue_number: int | None = None, 

110 pull_request: int | None = None, 

111 status: str = "in-progress", 

112 summary: str | None = None, 

113 next_step: str | None = None, 

114) -> str: 

115 """Render a stable issue progress/update comment.""" 

116 lines = [ 

117 ISSUE_UPDATE_MARKER, 

118 "", 

119 "## Ship update", 

120 "", 

121 f"- **Issue:** {_issue(issue_number)}", 

122 f"- **Pull request:** {_pr(pull_request)}", 

123 f"- **Status:** {_value(status, 'in-progress')}", 

124 f"- **Summary:** {_value(summary, 'No summary recorded.')}", 

125 f"- **Next step:** {_value(next_step, 'Continue the ship workflow.')}", 

126 ] 

127 return "\n".join(lines) + "\n" 

128 

129 

130def render_review_verdict( 

131 *, 

132 reviewer: str, 

133 head_sha: str | None, 

134 verdict: str = "LGTM", 

135 scope: str | None = None, 

136 findings: list[dict[str, Any]] | tuple[dict[str, Any], ...] = (), 

137 testing: str | None = None, 

138 vendor: str | None = None, 

139 model: str | None = None, 

140) -> str: 

141 """Render a head-bound reviewer verdict comment accepted by evidence verification. 

142 

143 When ``vendor`` (and optionally ``model``) is supplied, structured 

144 ``vendor:`` / ``model:`` provenance lines are emitted so evidence 

145 verification can enforce vendor distinctness across required verdicts. The 

146 fields use the same vendor/model conventions as ``keel.provenance`` and are 

147 omitted entirely when not supplied, so the default rendering is unchanged. 

148 """ 

149 lines = [ 

150 evidence.REVIEW_VERDICT_MARKER, 

151 f"reviewer: {_slug(reviewer)}", 

152 f"head: {_value(head_sha, '<head-sha>')}", 

153 ] 

154 if isinstance(vendor, str) and vendor.strip(): 

155 lines.append(f"vendor: {_slug(vendor)}") 

156 if isinstance(model, str) and model.strip(): 

157 lines.append(f"model: {_slug(model)}") 

158 lines.extend([ 

159 "", 

160 f"Verdict: {_value(verdict, 'LGTM')}", 

161 "", 

162 f"Scope reviewed: {_value(scope, 'Full changed-file diff and relevant contracts.')}", 

163 "", 

164 "Findings:", 

165 ]) 

166 lines.extend(_finding_lines(findings)) 

167 lines.extend(["", f"Testing noted: {_value(testing, 'See PR Testing section.')}"]) 

168 return "\n".join(lines) + "\n" 

169 

170 

171def render_jury_verdict( 

172 *, 

173 head_sha: str | None, 

174 participants: list[str] | tuple[str, ...] = (), 

175 verdict: str = "LGTM", 

176 findings_summary: list[str] | tuple[str, ...] = (), 

177 remaining_risks: str | None = None, 

178 participating_vendors: int | None = None, 

179) -> str: 

180 """Render a head-bound jury verdict comment accepted by evidence verification. 

181 

182 The verdict declares ``vendors: <N>`` — the distinct vendors that actually 

183 took part. That line is the only channel by which the panel size reaches a 

184 CI evidence check: the run ledger and the jury artifact both live under the 

185 gitignored ``.keel/state/``, so a hosted runner cannot read them, while PR 

186 comments are always visible. When ``participating_vendors`` is omitted it is 

187 inferred from ``participants``, so a caller that already lists them does not 

188 have to count twice. 

189 """ 

190 people = [person.strip() for person in participants if isinstance(person, str) 

191 and person.strip()] 

192 vendors = participating_vendors if participating_vendors is not None else len(people) 

193 lines = [ 

194 evidence.JURY_VERDICT_MARKER, 

195 f"head: {_value(head_sha, '<head-sha>')}", 

196 f"vendors: {vendors}", 

197 "", 

198 f"AI Jury verdict: {_value(verdict, 'LGTM')}.", 

199 "", 

200 f"Participants: {', '.join(people) if people else 'not recorded'}.", 

201 "", 

202 "Findings summary:", 

203 ] 

204 summaries = [item.strip() for item in findings_summary if isinstance(item, str) 

205 and item.strip()] 

206 lines.extend(f"- {item}" for item in summaries) if summaries else lines.append("- none") 

207 lines.extend(["", f"Remaining risks: {_value(remaining_risks, 'none identified')}."]) 

208 return "\n".join(lines) + "\n" 

209 

210 

211def render_review_cycle_summary( 

212 *, 

213 reviewers: list[dict[str, Any]] | tuple[dict[str, Any], ...] = (), 

214 head_sha: str | None = None, 

215 run_id: str | None = None, 

216) -> str: 

217 """Render the deterministic multi-reviewer review-cycle summary comment. 

218 

219 Emits one section per reviewer (codename · focus · verdict · a 

220 ``Severity | File:Line | Description | Suggested Fix`` finding table) followed 

221 by a consolidated summary whose severity histogram — not the verdict strings — 

222 is the source of truth for the merge recommendation. The output is byte-stable 

223 for a given input so the orchestrator posts it verbatim instead of improvising 

224 a layout. When ``run_id`` is supplied an invisible ``keel.run-id`` marker is 

225 appended so an idempotent re-post edits the existing comment in place. 

226 """ 

227 clean = [reviewer for reviewer in reviewers if isinstance(reviewer, dict)] 

228 lines = [ 

229 REVIEW_CYCLE_SUMMARY_MARKER, 

230 f"head: {_value(head_sha, '<head-sha>')}", 

231 "", 

232 ] 

233 for index, reviewer in enumerate(clean): 

234 if index: 

235 lines.extend(["", "---", ""]) 

236 lines.extend(_cycle_reviewer_lines(reviewer)) 

237 if clean: 

238 lines.extend(["", "---", ""]) 

239 lines.extend(_cycle_summary_lines(clean)) 

240 if isinstance(run_id, str) and run_id.strip(): 

241 lines.extend(["", f"<!-- keel.run-id: {run_id.strip()} -->"]) 

242 return "\n".join(lines) + "\n" 

243 

244 

245def render_coverage_delta( 

246 *, 

247 codename: str, 

248 base_sha: str | None = None, 

249 head_sha: str | None = None, 

250 areas: list[dict[str, Any]] | tuple[dict[str, Any], ...] = (), 

251) -> str: 

252 """Render the deterministic per-PR coverage delta comment. 

253 

254 The literal first line is the caller-supplied ``codename`` (e.g. 

255 ``COVERAGE-<PR>-<UTC>``) — the load-bearing anchor the adapter finds by prefix 

256 to update the comment in place, so nothing precedes it. The adapter supplies 

257 the timestamped codename, so the renderer stays pure and byte-stable for a 

258 given input. 

259 """ 

260 lines = [ 

261 codename, 

262 "", 

263 f"Coverage delta: base@{_value(base_sha, '<base>')} → head@{_value(head_sha, '<head>')}", 

264 ] 

265 for area in areas: 

266 if isinstance(area, dict): 

267 lines.extend(["", *_coverage_area_lines(area)]) 

268 lines.extend(["", f"<!-- {COVERAGE_DELTA_MARKER} -->"]) 

269 return "\n".join(lines) + "\n" 

270 

271 

272def render_deps_audit( 

273 *, 

274 codename: str, 

275 ecosystems: list[dict[str, Any]] | tuple[dict[str, Any], ...] = (), 

276 licences: list[dict[str, Any]] | tuple[dict[str, Any], ...] = (), 

277 skipped: list[str] | tuple[str, ...] = (), 

278 security_only: bool = False, 

279) -> str: 

280 """Render the deterministic dependency-audit comment for the tracking issue. 

281 

282 The literal first line is the caller-supplied ``codename`` (e.g. 

283 ``DEPS-AUDIT-<DATE>-<UTC>``); a fresh comment is appended per run, found later 

284 by that prefix. Under ``security_only`` the licence-drift section is omitted. 

285 """ 

286 counts = _deps_counts(ecosystems) 

287 lines = [ 

288 codename, 

289 "", 

290 " | ".join(f"{severity}: {counts[severity]}" for severity in DEPS_SEVERITY_ORDER), 

291 ] 

292 for ecosystem in ecosystems: 

293 if isinstance(ecosystem, dict): 

294 lines.extend(["", *_deps_ecosystem_lines(ecosystem)]) 

295 if not security_only: 

296 lines.extend(["", *_deps_licence_lines(licences)]) 

297 skipped_items = _string_list(skipped) 

298 if skipped_items: 

299 lines.extend(["", "## Skipped", ""]) 

300 lines.extend(f"- {item}" for item in skipped_items) 

301 lines.extend(["", f"<!-- {DEPS_AUDIT_MARKER} -->"]) 

302 return "\n".join(lines) + "\n" 

303 

304 

305def render_flake_audit( 

306 *, 

307 codename: str, 

308 summary: dict[str, Any] | None = None, 

309 new_flakes: list[dict[str, Any]] | tuple[dict[str, Any], ...] = (), 

310 tracked: list[dict[str, Any]] | tuple[dict[str, Any], ...] = (), 

311 limitations: list[str] | tuple[str, ...] = (), 

312) -> str: 

313 """Render the deterministic flake-audit report comment. 

314 

315 The literal first line is the caller-supplied ``codename`` (e.g. 

316 ``FLAKE-AUDIT-<DATE>-<UTC>``). Newly-classified flakes render as a table (or a 

317 single italic line when none cleared the threshold); already-tracked flakes 

318 and honest limitations render only when present. 

319 """ 

320 stats = summary if isinstance(summary, dict) else {} 

321 lines = [ 

322 codename, 

323 "", 

324 (f"runs examined: {_count(stats.get('runs'))} · " 

325 f"distinct failing tests: {_count(stats.get('distinct'))} · " 

326 f"classified flakes: {_count(stats.get('classified'))} · " 

327 f"newly-opened issues: {_count(stats.get('opened'))}"), 

328 "", 

329 "## Newly classified flakes", 

330 "", 

331 ] 

332 flakes = _dict_list(new_flakes) 

333 if flakes: 

334 lines.append("| Test | Fail rate | Failures | Sample runs | Signature |") 

335 lines.append("| --- | --- | --- | --- | --- |") 

336 lines.extend( 

337 _table_row([ 

338 _cell(_value(flake.get("test"), "—")), 

339 _cell(_value(flake.get("fail_rate"), "—")), 

340 _cell(str(_count(flake.get("failures")))), 

341 _cell(", ".join(_string_list(flake.get("samples"))) or "—"), 

342 _cell(_value(flake.get("signature"), "—")), 

343 ]) 

344 for flake in flakes 

345 ) 

346 else: 

347 lines.append("_no new flakes above threshold_") 

348 tracked_rows = _dict_list(tracked) 

349 if tracked_rows: 

350 lines.extend(["", "## Already tracked (deduped)", ""]) 

351 lines.extend( 

352 f"- {_value(row.get('test'), '—')} — see {_issue_ref(row.get('issue'))}" 

353 for row in tracked_rows 

354 ) 

355 limitation_items = _string_list(limitations) 

356 if limitation_items: 

357 lines.extend(["", "## Limitations", ""]) 

358 lines.extend(f"- {item}" for item in limitation_items) 

359 lines.extend(["", f"<!-- {FLAKE_AUDIT_MARKER} -->"]) 

360 return "\n".join(lines) + "\n" 

361 

362 

363def _table_row(cells: list[str]) -> str: 

364 return "| " + " | ".join(cells) + " |" 

365 

366 

367def _dict_list(raw: Any) -> list[dict[str, Any]]: 

368 if not isinstance(raw, (list, tuple)): 

369 return [] 

370 return [item for item in raw if isinstance(item, dict)] 

371 

372 

373def _count(value: Any) -> int: 

374 return value if isinstance(value, int) and not isinstance(value, bool) else 0 

375 

376 

377def _issue_ref(value: Any) -> str: 

378 if isinstance(value, int) and not isinstance(value, bool): 

379 return f"#{value}" 

380 return _value(value, "?") 

381 

382 

383def _scalar(value: Any, fallback: str) -> str: 

384 if isinstance(value, bool): 

385 return fallback 

386 if isinstance(value, int): 

387 return str(value) 

388 return _value(value, fallback) 

389 

390 

391def _is_number(value: Any) -> bool: 

392 return isinstance(value, (int, float)) and not isinstance(value, bool) 

393 

394 

395def _fmt_pct(value: Any) -> str: 

396 return f"{value:.1f}%" if _is_number(value) else "—" 

397 

398 

399def _fmt_delta(base: Any, head: Any) -> tuple[str, bool]: 

400 if _is_number(base) and _is_number(head): 

401 delta = head - base 

402 return f"{delta:+.1f}%", abs(delta) >= 0.5 

403 return "—", False 

404 

405 

406def _coverage_row(unit: str, base: Any, head: Any, files: Any, *, is_overall: bool = False) -> str: 

407 delta_text, bold = _fmt_delta(base, head) 

408 files_text = _cell(str(files)) if str(files).strip() else "" 

409 cells = [_cell(unit), _fmt_pct(base), _fmt_pct(head), delta_text, files_text] 

410 if bold: 

411 cells = [f"**{cell}**" if cell else "" for cell in cells] 

412 elif is_overall: 

413 cells[0] = f"**{cells[0]}**" 

414 return _table_row(cells) 

415 

416 

417def _coverage_area_lines(area: dict[str, Any]) -> list[str]: 

418 name = _value(area.get("name"), "area") 

419 if area.get("skipped"): 

420 return [f"_{name} coverage skipped: {_value(area.get('skip_reason'), 'not run')}_"] 

421 lines = [ 

422 f"## {name}", 

423 "", 

424 "| Unit | Base % | Head % | Δ | Files |", 

425 "| --- | --- | --- | --- | --- |", 

426 ] 

427 lines.extend( 

428 _coverage_row(_value(row.get("unit"), "—"), row.get("base"), row.get("head"), 

429 row.get("files", "")) 

430 for row in _dict_list(area.get("rows")) 

431 ) 

432 overall = area.get("overall") 

433 if isinstance(overall, dict): 

434 lines.append( 

435 _coverage_row("overall", overall.get("base"), overall.get("head"), "", 

436 is_overall=True) 

437 ) 

438 return lines 

439 

440 

441def _deps_sev_rank(severity: str) -> int: 

442 lowered = severity.lower() 

443 return DEPS_SEVERITY_ORDER.index(lowered) if lowered in DEPS_SEVERITY_ORDER \ 

444 else len(DEPS_SEVERITY_ORDER) 

445 

446 

447def _deps_counts(ecosystems: Any) -> dict[str, int]: 

448 counts = dict.fromkeys(DEPS_SEVERITY_ORDER, 0) 

449 for ecosystem in _dict_list(ecosystems): 

450 for finding in _dict_list(ecosystem.get("findings")): 

451 severity = _value(finding.get("severity"), "low").lower() 

452 if severity in counts: 

453 counts[severity] += 1 

454 return counts 

455 

456 

457def _deps_ecosystem_lines(ecosystem: dict[str, Any]) -> list[str]: 

458 name = _value(ecosystem.get("name"), "ecosystem") 

459 findings = _dict_list(ecosystem.get("findings")) 

460 if not findings: 

461 threshold = _value(ecosystem.get("threshold"), "low") 

462 return [f"_No {name} findings at or above {threshold} severity._"] 

463 findings = sorted(findings, key=lambda f: _deps_sev_rank(_value(f.get("severity"), "low"))) 

464 lines = [ 

465 f"## {name}", 

466 "", 

467 "| Package | Version | Severity | Advisory | Fix available |", 

468 "| --- | --- | --- | --- | --- |", 

469 ] 

470 lines.extend( 

471 _table_row([ 

472 _cell(_value(finding.get("package"), "—")), 

473 _cell(_value(finding.get("version"), "—")), 

474 _cell(_value(finding.get("severity"), "low")), 

475 _cell(_value(finding.get("advisory"), "—")), 

476 _cell(_value(finding.get("fix_available"), "—")), 

477 ]) 

478 for finding in findings 

479 ) 

480 return lines 

481 

482 

483def _deps_licence_lines(licences: Any) -> list[str]: 

484 rows = _dict_list(licences) 

485 if not rows: 

486 return ["_licences: no drift_"] 

487 lines = [ 

488 "## Licences", 

489 "", 

490 "| Status | Package | Baseline | Current |", 

491 "| --- | --- | --- | --- |", 

492 ] 

493 lines.extend( 

494 _table_row([ 

495 _cell(_value(row.get("status"), "—")), 

496 _cell(_value(row.get("package"), "—")), 

497 _cell(_value(row.get("baseline"), "—")), 

498 _cell(_value(row.get("current"), "—")), 

499 ]) 

500 for row in rows 

501 ) 

502 return lines 

503 

504 

505def render_scan_finding_issue( 

506 *, 

507 problem: str | None = None, 

508 location: str | None = None, 

509 severity: str | None = None, 

510 justification: str | None = None, 

511 evidence: str | None = None, 

512 suggested_fix: str | None = None, 

513 source: str | None = None, 

514 regression_of: int | None = None, 

515) -> str: 

516 """Render the deterministic issue body for a scan finding. 

517 

518 Shared by ``regression`` and ``review-all-day`` — the body carries the 

519 problem statement, ``path:line`` location, severity + justification, fenced 

520 evidence, and suggested fix, plus a provenance marker. When ``regression_of`` 

521 is supplied the grep-able ``regression-of: #N`` cross-reference is the body's 

522 literal last line. 

523 """ 

524 lines = [ 

525 "## Problem", 

526 "", 

527 _value(problem, "A scan finding was reported without a problem statement."), 

528 "", 

529 "## Location", 

530 "", 

531 f"`{_value(location, 'unknown')}`", 

532 "", 

533 "## Severity", 

534 "", 

535 f"{_value(severity, 'minor')} — {_value(justification, 'no justification recorded')}", 

536 "", 

537 "## Evidence", 

538 "", 

539 "```", 

540 _value(evidence, "none provided"), 

541 "```", 

542 "", 

543 "## Suggested fix", 

544 "", 

545 _value(suggested_fix, "none proposed"), 

546 "", 

547 f"Found by keel {_value(source, 'scan')}.", 

548 "", 

549 f"<!-- {SCAN_FINDING_MARKER} -->", 

550 ] 

551 if isinstance(regression_of, int) and not isinstance(regression_of, bool): 

552 lines.extend(["", f"regression-of: #{regression_of}"]) 

553 return "\n".join(lines) + "\n" 

554 

555 

556def render_triage_audit( 

557 *, 

558 issue: int | None = None, 

559 role: str | None = None, 

560 priority: str | None = None, 

561 status: str | None = None, 

562 tier: int | str | None = None, 

563 rationale: str | None = None, 

564 run_id: str | None = None, 

565) -> str: 

566 """Render the deterministic, label-only triage audit comment. 

567 

568 One comment per triaged issue: the applied role / priority / status labels and 

569 risk tier on one line, then the classifier's rationale. When ``run_id`` is 

570 supplied an idempotent re-post edits the existing comment in place. 

571 """ 

572 labels = " · ".join([ 

573 f"role: {_value(role, 'unassigned')}", 

574 f"priority: {_value(priority, 'unset')}", 

575 f"status: {_value(status, 'unset')}", 

576 f"tier: {_scalar(tier, 'n/a')}", 

577 ]) 

578 lines = [ 

579 TRIAGE_AUDIT_MARKER, 

580 f"keel triage — {_issue_ref(issue)}: {labels}", 

581 "", 

582 _value(rationale, "Classified from the existing label set."), 

583 ] 

584 if isinstance(run_id, str) and run_id.strip(): 

585 lines.extend(["", f"<!-- keel.run-id: {run_id.strip()} -->"]) 

586 return "\n".join(lines) + "\n" 

587 

588 

589def render_extension_result( 

590 *, 

591 slot: str, 

592 extension_id: str, 

593 status: str, 

594 mode: str, 

595 summary: str | None = None, 

596 artifacts: list[str] | tuple[str, ...] = (), 

597 follow_ups: list[str] | tuple[str, ...] = (), 

598) -> str: 

599 """Render a canonical extension result block/comment.""" 

600 lines = [ 

601 EXTENSION_RESULT_MARKER, 

602 "", 

603 "## Extension result", 

604 "", 

605 f"- **Slot:** `{_value(slot, 'unknown')}`", 

606 f"- **Extension:** `{_value(extension_id, 'unknown')}`", 

607 f"- **Status:** {_value(status, 'not-recorded')}", 

608 f"- **Mode:** {_value(mode, 'advisory')}", 

609 f"- **Summary:** {_value(summary, 'No summary recorded.')}", 

610 "- **Artifacts:**", 

611 ] 

612 artifact_lines = _string_bullets(artifacts) 

613 lines.extend(artifact_lines if artifact_lines else [" - none"]) 

614 lines.append("- **Follow-ups:**") 

615 follow_up_lines = _string_bullets(follow_ups) 

616 lines.extend(follow_up_lines if follow_up_lines else [" - none"]) 

617 return "\n".join(lines) + "\n" 

618 

619 

620def render_step_handoff( 

621 *, 

622 step_id: str, 

623 step_name: str | None = None, 

624 status: str = "complete", 

625 summary: str | None = None, 

626 next_step: str | None = None, 

627 evidence_ids: list[str] | tuple[str, ...] = (), 

628) -> str: 

629 """Render the canonical structured handoff between backbone steps.""" 

630 lines = [ 

631 STEP_HANDOFF_MARKER, 

632 "", 

633 "## Step handoff", 

634 "", 

635 f"- **Step:** `{_value(step_id, 'unknown')}`", 

636 f"- **Name:** {_value(step_name, 'not recorded')}", 

637 f"- **Status:** {_value(status, 'complete')}", 

638 f"- **Summary:** {_value(summary, 'No summary recorded.')}", 

639 f"- **Next step:** {_value(next_step, 'Continue the backbone plan.')}", 

640 "- **Evidence:**", 

641 ] 

642 evidence_lines = _string_bullets(evidence_ids) 

643 lines.extend(evidence_lines if evidence_lines else [" - none"]) 

644 return "\n".join(lines) + "\n" 

645 

646 

647def render_run_control_halt( 

648 *, 

649 control: str, 

650 reason: str, 

651 scope: str | None = None, 

652 observed: int | str | None = None, 

653 limit: int | str | None = None, 

654 action: str | None = None, 

655) -> str: 

656 """Render a stable hard-halt reason emitted by run controls.""" 

657 lines = [ 

658 RUN_CONTROL_HALT_MARKER, 

659 "", 

660 "## Run control halt", 

661 "", 

662 f"- **Control:** `{_value(control, 'unknown')}`", 

663 f"- **Reason:** {_value(reason, 'No reason recorded.')}", 

664 f"- **Scope:** {_value(scope, 'run')}", 

665 f"- **Observed:** {_value(observed, 'not recorded')}", 

666 f"- **Limit:** {_value(limit, 'not recorded')}", 

667 f"- **Action:** {_value(action, 'halt')}", 

668 ] 

669 return "\n".join(lines) + "\n" 

670 

671 

672def _finding_lines(findings: list[dict[str, Any]] | tuple[dict[str, Any], ...]) -> list[str]: 

673 if not findings: 

674 return ["- none"] 

675 lines: list[str] = [] 

676 for finding in findings: 

677 severity = _value(finding.get("severity") if isinstance(finding, dict) else None, "nit") 

678 message = _value(finding.get("message") if isinstance(finding, dict) else None, "") 

679 if message: 

680 lines.append(f"- {severity}: {message}") 

681 return lines or ["- none"] 

682 

683 

684def _string_bullets(values: list[str] | tuple[str, ...]) -> list[str]: 

685 return [f" - {value.strip()}" for value in values if isinstance(value, str) 

686 and value.strip()] 

687 

688 

689def _string_list(values: Any) -> list[str]: 

690 if not isinstance(values, (list, tuple)): 

691 return [] 

692 return [value.strip() for value in values if isinstance(value, str) and value.strip()] 

693 

694 

695def _canonical_severity(severity: str) -> str: 

696 lowered = severity.strip().lower() 

697 return _SEVERITY_ALIASES.get(lowered, lowered) 

698 

699 

700def _severity_rank(severity: str) -> int: 

701 canonical = _canonical_severity(severity) 

702 return SEVERITY_ORDER.index(canonical) if canonical in SEVERITY_ORDER else len(SEVERITY_ORDER) 

703 

704 

705def _cell(value: str) -> str: 

706 """Escape a non-empty finding string for a Markdown table cell. 

707 

708 Callers pass values already normalised through ``_value`` (never blank), so 

709 escaping the table delimiter and folding newlines keeps the row intact. 

710 """ 

711 return value.replace("\n", " ").replace("|", "\\|") 

712 

713 

714def _cycle_findings(raw: Any) -> list[dict[str, str]]: