Coverage for core/langgraph/_builtins.py: 72%

1""" 

2_builtins.py — Domain builtins for agent-platform pipelines (v1). 

3 

4Toteuttaa ne builtin-komponentit, joita agent-platform-pipelinet käyttävät 

5ja jotka porttaavat ei-LLM-logiikan generic-muotoon: 

6 

7- path-verifier tarkistaa dokumenttien viittaamat polut 

8- finding-prefilter suodattaa false-positive findingit ennen LLM-triagea 

9- ticket-publisher kirjoittaa hyväksytyn ticketin staging → published 

10- run-finalizer koostaa stats + flushaa Langfusen 

11 

12Kaikki builtinit noudattavat sopimusta: 

13 fn(step: dict, state: dict) -> dict # palauttaa uuden state-snapshotin 

14ja kirjoittavat tuloksensa kenttään `state[step["id"]]`. 

15""" 

16from __future__ import annotations 

17 

18import json 

19import logging 

20import os 

21import re 

22import shutil 

23import subprocess 

24from concurrent.futures import ThreadPoolExecutor, as_completed 

25from datetime import date 

26from pathlib import Path 

27from typing import Any 

28from urllib.parse import urlparse, urlunparse 

29 

30log = logging.getLogger(__name__) 

31 

32# Same dual-import pattern as runner.py (flat sys.path in tests vs qualified 

33# PYTHONPATH=/app in main.py) — slugify is the ADR-0018 slug used for the 

34# TICKET-profile per-run feature branch name. 

35try: 

36 from exec_runner import slugify # noqa: F401 

37except ImportError: # pragma: no cover — main.py-ajossa (qualified importit) 

38 from langgraph.exec_runner import slugify # noqa: F401 

39 

40# Files-output-helper (validate_and_write_files) — sama dual-import-kuvio 

41# kuin runner.py:ssa: flat sys.path (testit) vs. langgraph.* (main.py). 

42try: 

43 from _files_output import ( # noqa: F401 

44 FilesOutputError, 

45 _resolve_path, 

46 resolve_workspace_root, 

47 validate_and_write_files, 

48 ) 

49except ImportError: # pragma: no cover — main.py-ajossa (qualified importit) 

50 from langgraph._files_output import ( # noqa: F401 

51 FilesOutputError, 

52 _resolve_path, 

53 resolve_workspace_root, 

54 validate_and_write_files, 

55 ) 

56 

57 

58_DEFAULT_SKIP_DIRS = { 

59 ".git", "node_modules", "__pycache__", ".venv", "venv", 

60 ".mypy_cache", ".pytest_cache", ".ruff_cache", 

61} 

62 

63 

64# --------------------------------------------------------------------------- 

65# path-verifier 

66# --------------------------------------------------------------------------- 

67 

68 

69def _path_resolution_candidates(rel: str, source_doc: str = "") -> list[str]: 

70 rel = rel.strip().lstrip("/") 

71 if not rel or rel.startswith("http"): 

72 return [] 

73 candidates: list[str] = [rel] 

74 if source_doc: 

75 doc_dir = Path(source_doc).parent.as_posix() 

76 if doc_dir and doc_dir != ".": 

77 joined = f"{doc_dir}/{rel}".replace("//", "/") 

78 if joined not in candidates: 

79 candidates.append(joined) 

80 if not rel.startswith(("docs/", ".agent-platform/", "skills/", "assets/", "scripts/")): 

81 if f"docs/{rel}" not in candidates: 

82 candidates.append(f"docs/{rel}") 

83 if rel.startswith("adr/"): 

84 prefixed = f"docs/{rel}" 

85 if prefixed not in candidates: 

86 candidates.append(prefixed) 

87 return candidates 

88 

89 

90def _paths_in_text(text: str, max_paths: int = 25) -> list[str]: 

91 seen: set[str] = set() 

92 out: list[str] = [] 

93 

94 def add(raw: str) -> None: 

95 p = raw.strip().split("#")[0].strip() 

96 if not p or p.startswith("http") or p in seen: 

97 return 

98 # Suodata pois ei-polku-osumat: rivinvaihdot, välilyönnit, liian pitkät 

99 # (esim. mermaid-koodilohkojen sisältö, joka match-aa backtick-regexiin). 

100 if any(ch in p for ch in ("\n", "\r", " ", "\t")) or len(p) > 255: 

101 return 

102 if "/" in p or p.endswith((".md", ".py", ".yaml", ".yml", ".sh", ".json", ".toml")): 

103 seen.add(p) 

104 out.append(p) 

105 

106 for p in re.findall(r"`([^`]+)`", text): 

107 add(p) 

108 if len(out) >= max_paths: 

109 return out[:max_paths] 

110 for p in re.findall(r"\]\(([^)]+)\)", text): 

111 add(p) 

112 if len(out) >= max_paths: 

113 break 

114 return out[:max_paths] 

115 

116 

117def _verify_paths_for_doc(root: Path, source_doc: str, doc_text: str) -> str: 

118 paths = _paths_in_text(doc_text) 

119 if not paths: 

120 return "" 

121 lines = [ 

122 "\n\n## Path verification (project mount — authoritative)\n", 

123 f"Document under audit: `{source_doc}` — resolve relative links from this directory.\n", 

124 "Do NOT report a path as missing when any resolved candidate below is `exists`.\n", 

125 ] 

126 for p in paths: 

127 resolved = _path_resolution_candidates(p, source_doc) 

128 hit = next((c for c in resolved if (root / c).is_file()), None) 

129 if hit: 

130 lines.append(f"- `{p}`: exists (at `{hit}`)") 

131 else: 

132 shown = ", ".join(f"`{c}`" for c in resolved[:4]) 

133 lines.append(f"- `{p}`: MISSING (checked: {shown})") 

134 return "\n".join(lines) 

135 

136 

137_HOST_AI_CONTEXT_MAP_PLACEHOLDER = "[host-ai-context-map-path]" 

138_CONVENTION_MAP_REL = "docs/ai-context.md" 

139 

140 

141def _resolve_required_path_element(raw: str) -> tuple[str, bool]: 

142 """Resolve one `input.paths` element to a root-relative path. 

143 

144 Returns (root_relative_path, is_placeholder). The host-ai-context-map-path 

145 placeholder resolves to the convention map location 

146 `<project_root>/docs/ai-context.md` (ADR-0012 mandatory anchor). Never 

147 returns the placeholder literal. 

148 """ 

149 stripped = raw.strip() 

150 if stripped == _HOST_AI_CONTEXT_MAP_PLACEHOLDER: 

151 return _CONVENTION_MAP_REL, True 

152 return stripped.lstrip("/"), False 

153 

154 

155def _map_exists_block(resolved_rel: str) -> str: 

156 return ( 

157 "\n\n## Path verification (map anchor — project mount)\n" 

158 f"- `{resolved_rel}`: exists\n" 

159 ) 

160 

161 

162def _missing_map_hard_failure(step_id: str, raw: str, resolved_rel: str) -> "RunHardFailure": 

163 try: 

164 from runner import RunHardFailure 

165 except ImportError: # pragma: no cover — main.py-ajossa (qualified importit) 

166 from langgraph.runner import RunHardFailure 

167 message = ( 

168 f"MISSING_MAP: {step_id}: required map path {raw!r} (resolved " 

169 f"{resolved_rel!r}) is missing under the project root — a missing " 

170 "required map is a hard error (ADR-0012), no silent no-op." 

171 ) 

172 return RunHardFailure( 

173 "MISSING_MAP", 

174 message, 

175 { 

176 "error_code": "MISSING_MAP", 

177 "step_id": step_id, 

178 "map_path": resolved_rel, 

179 }, 

180 ) 

181 

182 

183def path_verifier(step: dict, state: dict) -> dict: 

184 """ 

185 params: 

186 docs_from: dot-path state-avain (esim. "gather.doc_files") — gather-docs 

187 validation when `input.paths` is absent (backward compat). 

188 input: 

189 paths: [str] required paths to verify existence (CS map anchor). A 

190 missing required path is a HARD error (RunHardFailure, ADR-0012) — 

191 never a silent empty by_path / no-op. 

192 output (state[step.id]): 

193 {by_path: {<path>: <verification_block_string>}, 

194 resolved_map_path?: <resolved convention map path>} 

195 """ 

196 step_id = step.get("id", "verify_paths") 

197 params = step.get("params") or {} 

198 root = Path(state.get("project_root") or os.getenv("PROJECT_ROOT", "/project")) 

199 

200 input_spec = step.get("input") 

201 required_paths: list[str] | str | None = None 

202 if isinstance(input_spec, dict) and input_spec.get("paths") is not None: 

203 required_paths = input_spec["paths"] 

204 

205 if required_paths is None: 

206 return _verify_docs_from(step, state, step_id, params, root) 

207 

208 if isinstance(required_paths, str): 

209 required_paths = [required_paths] 

210 if not isinstance(required_paths, list) or not required_paths: 

211 raise _missing_map_hard_failure(step_id, "<empty required paths>", "<none>") 

212 

213 by_path: dict[str, str] = {} 

214 resolved_map_path: str | None = None 

215 for raw in required_paths: 

216 resolved_rel, is_placeholder = _resolve_required_path_element(str(raw)) 

217 if is_placeholder: 

218 resolved_map_path = resolved_rel 

219 if (root / resolved_rel).is_file(): 

220 by_path[resolved_rel] = _map_exists_block(resolved_rel) 

221 else: 

222 raise _missing_map_hard_failure(step_id, str(raw), resolved_rel) 

223 

224 result: dict[str, Any] = {"by_path": by_path} 

225 if resolved_map_path is not None: 

226 result["resolved_map_path"] = resolved_map_path 

227 log.info("[path-verifier] verified %d required map path(s)", len(by_path)) 

228 return {**state, step_id: result} 

229 

230 

231def _verify_docs_from(step: dict, state: dict, step_id: str, params: dict, root: Path) -> dict: 

232 """Legacy gather-docs validation via `params.docs_from` (backward compat).""" 

233 docs_from = params.get("docs_from", "gather.doc_files") 

234 

235 # Naive dot-path lookup against state 

236 cur: Any = state 

237 for part in docs_from.split("."): 

238 if isinstance(cur, dict): 

239 cur = cur.get(part) 

240 else: 

241 cur = None 

242 break 

243 docs_list = cur if isinstance(cur, list) else [] 

244 

245 by_path: dict[str, str] = {} 

246 for d in docs_list: 

247 if not isinstance(d, dict): 

248 continue 

249 path = d.get("path") or "" 

250 content = d.get("content") or "" 

251 if path and content: 

252 by_path[path] = _verify_paths_for_doc(root, path, content) 

253 log.info("[path-verifier] verified %d docs", len(by_path)) 

254 return {**state, step_id: {"by_path": by_path}} 

255 

256 

257# --------------------------------------------------------------------------- 

258# finding-prefilter 

259# --------------------------------------------------------------------------- 

260 

261 

262_FALSE_MISSING_PHRASES = ( 

263 "does not exist", "do not exist", "non-existent", "non existent", 

264 "not exist in the project", "missing from the project", "cannot be found", 

265 "can't be found", "file is missing", "not found in the project", 

266) 

267_NON_ACTIONABLE_PHRASES = ( 

268 "no violation", "no missing path", "not a claim of absence", 

269 "no observable mismatch", "does not violate", "without verification", 

270 "not an issue", "no actionable", "no issue found", 

271 "no concrete doc defect", "no documentation defect", 

272 "not a documentation defect", "forward reference, not a claim", 

273 "this is a forward reference", "no missing path claims", 

274 "not verified to exist", "are not verified to exist", 

275 "is not verified to exist", 

276) 

277_FUTURE_CLAIM_PHRASES = ( 

278 "is in the future", 

279 "which is in the future", 

280 "future date", 

281 "in the future", 

282) 

283_DATE_PATTERN = re.compile(r"\b(\d{4}-\d{2}-\d{2})\b") 

284_SEVERITY_RANK = {"high": 0, "medium": 1, "low": 2} 

285 

286 

287def _is_non_actionable(detail: str) -> bool: 

288 lower = (detail or "").lower() 

289 if not lower.strip(): 

290 return True 

291 for phrase in _NON_ACTIONABLE_PHRASES: 

292 if phrase in lower: 

293 return True 

294 if "does not cite" in lower and "not existence" in lower: 

295 return True 

296 return False 

297 

298 

299def _is_false_future_date(detail: str) -> bool: 

300 """Return True when the finding claims a past-or-present date is in the future. 

301 

302 Auditors occasionally hallucinate temporal direction — labelling a document 

303 date that has already passed as "in the future". The check is conservative: 

304 we only reject when the detail contains an explicit future-claim phrase AND 

305 every ISO-8601 date found in the text is on or before today. 

306 """ 

307 lower = (detail or "").lower() 

308 if not any(phrase in lower for phrase in _FUTURE_CLAIM_PHRASES): 

309 return False 

310 today = date.today() 

311 dates_found = _DATE_PATTERN.findall(detail or "") 

312 if not dates_found: 

313 return False 

314 # Reject only when all mentioned dates are past-or-present (no genuinely future date) 

315 for raw in dates_found: 

316 try: 

317 if date.fromisoformat(raw) > today: 

318 return False # at least one date really is in the future → keep 

319 except ValueError: 

320 continue 

321 return True 

322 

323 

324def _claimed_missing_paths(detail: str) -> list[str]: 

325 paths = [p.strip() for p in re.findall(r"Referenced path `([^`]+)`", detail or "", re.I)] 

326 if not paths: 

327 for p in re.findall(r"`([^`]+)`", detail or ""): 

328 p = p.strip() 

329 if p and not p.startswith("http") and ("/" in p or "." in p): 

330 paths.append(p) 

331 seen: set[str] = set() 

332 out: list[str] = [] 

333 for p in paths: 

334 if p not in seen: 

335 seen.add(p) 

336 out.append(p) 

337 return out 

338 

339 

340def _project_path_exists(root: Path, rel: str, source_doc: str = "") -> bool: 

341 for c in _path_resolution_candidates(rel, source_doc): 

342 if (root / c).is_file(): 

343 return True 

344 return False 

345 

346 

347def _finding_source_doc(finding: dict) -> str: 

348 for r in finding.get("refs") or []: 

349 if isinstance(r, str) and ("/" in r or r.endswith(".md")): 

350 return r 

351 return "" 

352 

353 

354def _is_false_positive(root: Path, finding: dict) -> tuple[bool, str]: 

355 detail = finding.get("detail") or "" 

356 if _is_non_actionable(detail): 

357 return True, "non_actionable" 

358 if _is_false_future_date(detail): 

359 return True, "false_future_date" 

360 lower = detail.lower() 

361 layer = finding.get("layer", "") 

362 

363 if "docs/tickets/" in lower and ("ticket `" in lower or "ticket '" in lower): 

364 return True, "meta_ticket_noise" 

365 

366 if any(p in lower for p in _FALSE_MISSING_PHRASES): 

367 source_doc = _finding_source_doc(finding) 

368 claimed = _claimed_missing_paths(detail) 

369 if claimed and all(_project_path_exists(root, p, source_doc) for p in claimed): 

370 return True, "false_missing_file" 

371 refs = [r for r in (finding.get("refs") or []) if isinstance(r, str)] 

372 if refs and all(_project_path_exists(root, r, source_doc) for r in refs): 

373 return True, "false_missing_file" 

374 

375 if layer == "C" and "doc sample" in lower and ("empty" in lower or "does not include" in lower): 

376 return True, "false_empty_doc_sample" 

377 

378 if finding.get("severity") == "low": 

379 for noise in ("typo", "capitalization", "readability", "finnish", "english", 

380 "language mixing", "mixed language"): 

381 if noise in lower: 

382 return True, "low_style_noise" 

383 

384 return False, "" 

385 

386 

387def _severity_allowed(severity: str, min_severity: str) -> bool: 

388 s = _SEVERITY_RANK.get(str(severity).lower(), 3) 

389 m = _SEVERITY_RANK.get(min_severity, 1) 

390 return s <= m 

391 

392 

393def finding_prefilter(step: dict, state: dict) -> dict: 

394 """ 

395 params: 

396 min_severity: 'high'|'medium'|'low' 

397 finding: the finding dict (passed via params via {{ item }} in manifest) 

398 OR step.input resolves to it (preferred shape). 

399 output (state[step.id]): 

400 {keep: bool, reason: str, finding: dict} 

401 """ 

402 step_id = step.get("id", "prefilter") 

403 params = step.get("params") or {} 

404 min_severity = str(params.get("min_severity", "medium")).lower() 

405 

406 # The finding can arrive via input (resolved) or params.finding 

407 finding = step.get("input") 

408 if not isinstance(finding, dict): 

409 finding = params.get("finding") 

410 if not isinstance(finding, dict): 

411 return {**state, step_id: {"keep": False, "reason": "no_finding", "finding": {}}} 

412 

413 root = Path(state.get("project_root") or os.getenv("PROJECT_ROOT", "/project")) 

414 is_fp, reason = _is_false_positive(root, finding) 

415 if is_fp: 

416 return {**state, step_id: {"keep": False, "reason": reason, "finding": finding}} 

417 if not _severity_allowed(finding.get("severity", "low"), min_severity): 

418 return {**state, step_id: {"keep": False, "reason": f"below_min_severity_{min_severity}", "finding": finding}} 

419 return {**state, step_id: {"keep": True, "reason": "", "finding": finding}} 

420 

421 

422# --------------------------------------------------------------------------- 

423# ticket-publisher 

424# --------------------------------------------------------------------------- 

425 

426 

427_TICKET_ORIGIN = "pipeline" 

428 

429 

430def _ensure_ticket_date(text: str, today_str: str) -> str: 

431 """Inject **Date:** after the title line if not already present.""" 

432 if re.search(r"^\*\*Date:\*\*", text, re.M | re.I): 

433 return text 

434 m = re.search(r"^(#\s+Ticket:.*)$", text, re.M | re.I) 

435 if m: 

436 insert_at = m.end() 

437 return text[:insert_at] + f"\n\n**Date:** {today_str}" + text[insert_at:] 

438 return text 

439 

440 

441def _ensure_ticket_origin(text: str) -> str: 

442 if re.search(r"^\*\*Origin:\*\*", text, re.M | re.I): 

443 return re.sub( 

444 r"^\*\*Origin:\*\*\s*.*$", 

445 f"**Origin:** {_TICKET_ORIGIN}", 

446 text, 

447 count=1, 

448 flags=re.M | re.I, 

449 ) 

450 m = re.search(r"^(#\s+Ticket:.*)$", text, re.M | re.I) 

451 if m: 

452 insert_at = m.end() 

453 return text[:insert_at] + f"\n\n**Origin:** {_TICKET_ORIGIN}" + text[insert_at:] 

454 return f"**Origin:** {_TICKET_ORIGIN}\n\n" + text 

455 

456 

457def _slug_from_text(text: str) -> str: 

458 m = re.search(r"^#\s+Ticket:\s*(.+)$", text, re.M | re.I) 

459 title = m.group(1).strip() if m else "pipeline-issue" 

460 return re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")[:40] or "issue" 

461 

462 

463def ticket_publisher(step: dict, state: dict) -> dict: 

464 """ 

465 params: 

466 staging_dir: str (oletus .agent-platform/tmp/pipeline-tickets) 

467 publish_dir: str (oletus docs/tickets) 

468 status_from: str (DRAFT) 

469 status_to: str (GATED) 

470 source_footer: str (Source: pipeline) 

471 input (resolved): ticket-markdown teksti TAI {output: "..."} dict 

472 output (state[step.id]): {path: str|None, slug: str|None, error?: str} 

473 """ 

474 step_id = step.get("id", "publish") 

475 params = step.get("params") or {} 

476 root = Path(state.get("project_root") or os.getenv("PROJECT_ROOT", "/project")) 

477 

478 ticket_text = step.get("input") 

479 if isinstance(ticket_text, dict): 

480 ticket_text = ticket_text.get("output") or ticket_text.get("text") or "" 

481 if not isinstance(ticket_text, str) or not ticket_text.strip(): 

482 return {**state, step_id: {"path": None, "slug": None, "error": "empty_ticket"}} 

483 

484 status_from = str(params.get("status_from", "DRAFT")) 

485 status_to = str(params.get("status_to", "GATED")) 

486 publish_dir = root / str(params.get("publish_dir", "docs/tickets")) 

487 source_footer = str(params.get("source_footer", "Source: pipeline")) 

488 

489 body = ticket_text.strip() 

490 today = date.today().isoformat() 

491 # Strip optional draft prefix line "> **DRAFT** — ..." 

492 body = re.sub(r"^>\s*\*\*DRAFT\*\*.*\n\n", "", body, count=1, flags=re.M) 

493 # Status DRAFT → GATED 

494 if re.search(rf"^\*\*Status:\*\*\s*{status_from}\s*$", body, re.M | re.I): 

495 body = re.sub( 

496 rf"^\*\*Status:\*\*\s*{status_from}\s*$", 

497 f"**Status:** {status_to}", 

498 body, 

499 count=1, 

500 flags=re.M | re.I, 

501 ) 

502 elif not re.search(r"^\*\*Status:\*\*", body, re.M | re.I): 

503 # Inject Status block after first heading 

504 body = re.sub( 

505 r"^(#\s+Ticket:.*)$", 

506 rf"\1\n\n**Status:** {status_to}", 

507 body, 

508 count=1, 

509 flags=re.M, 

510 ) 

511 body = _ensure_ticket_date(body, today) 

512 body = _ensure_ticket_origin(body) 

513 

514 slug = _slug_from_text(body) 

515 publish_dir.mkdir(parents=True, exist_ok=True) 

516 target = publish_dir / f"{today}-{slug}.md" 

517 n = 1 

518 while target.exists(): 

519 target = publish_dir / f"{today}-{slug}-{n}.md" 

520 n += 1 

521 

522 footer = f"\n\n---\n{source_footer}\n" 

523 try: 

524 target.write_text(body.rstrip() + footer, encoding="utf-8") 

525 except OSError as exc: 

526 return {**state, step_id: {"path": None, "slug": slug, "error": str(exc)}} 

527 

528 rel_path = str(target.relative_to(root)) if target.is_relative_to(root) else str(target) 

529 log.info("[ticket-publisher] published %s", rel_path) 

530 return {**state, step_id: {"path": rel_path, "slug": slug}} 

531 

532 

533# --------------------------------------------------------------------------- 

534# run-finalizer 

535# --------------------------------------------------------------------------- 

536 

537 

538def run_finalizer(step: dict, state: dict) -> dict: 

539 """ 

540 Koostaa loppustatuksen + runlog-rivin + flushaa Langfuse. 

541 params: 

542 runlog: str (oletus .agent-platform/tmp/runlog) 

543 pipeline_id: str 

544 output (state[step.id]): {run_status, stats, runlog_path?} 

545 """ 

546 step_id = step.get("id", "finalize") 

547 params = step.get("params") or {} 

548 pipeline_id = str(params.get("pipeline_id", "pipeline")) 

549 root = Path(state.get("project_root") or os.getenv("PROJECT_ROOT", "/project")) 

550 

551 stats: dict[str, Any] = dict(state.get("stats") or {}) 

552 

553 # Aggregate counts from common state fields if present 

554 # foreach tallentaa publish-stepin outputin (= {"path": rel}) sellaisenaan 

555 # results-listaan, ei {"publish": ...} -kääreessä. Tuetaan molempia muotoja. 

556 process_results = state.get("process_findings") or [] 

557 if isinstance(process_results, list): 

558 def _published_path(r: Any) -> str | None: 

559 if not isinstance(r, dict): 

560 return None 

561 # Suora muoto: {"path": "..."} (foreachin publish_out) 

562 if r.get("path"): 

563 return r.get("path") 

564 # Käärittu muoto: {"publish": {"path": "..."}} 

565 return ((r.get("publish") or {}).get("path")) 

566 

567 published = [r for r in process_results if _published_path(r)] 

568 stats["tickets_published"] = len(published) 

569 stats["tickets_created"] = len(published) 

570 

571 merged = state.get("merge_findings") 

572 if isinstance(merged, dict): 

573 findings = merged.get("findings") or [] 

574 stats["findings_merged"] = len(findings) 

575 

576 gather = state.get("gather") 

577 if isinstance(gather, dict) and gather.get("stats"): 

578 for k, v in gather["stats"].items(): 

579 stats.setdefault(k, v) 

580 

581 run_status = state.get("run_status") or "ok" 

582 if state.get("errors"): 

583 run_status = "partial" if run_status != "failed" else run_status 

584 

585 # Append runlog line (best-effort) 

586 runlog_path: str | None = None 

587 runlog_dir = root / str(params.get("runlog", ".agent-platform/tmp/runlog")) 

588 try: 

589 runlog_dir.mkdir(parents=True, exist_ok=True) 

590 line = { 

591 "pipeline": pipeline_id, 

592 "status": run_status, 

593 "stats": stats, 

594 "errors": (state.get("errors") or [])[:10], 

595 } 

596 target = runlog_dir / f"{date.today().isoformat()}-{pipeline_id}.jsonl" 

597 with target.open("a", encoding="utf-8") as fh: 

598 fh.write(json.dumps(line, ensure_ascii=False) + "\n") 

599 runlog_path = str(target.relative_to(root)) if target.is_relative_to(root) else str(target) 

600 except OSError as exc: 

601 log.warning("[run-finalizer] runlog write failed: %s", exc) 

602 

603 out_state = { 

604 **state, 

605 step_id: {"run_status": run_status, "stats": stats, "runlog_path": runlog_path}, 

606 "run_status": run_status, 

607 "stats": stats, 

608 } 

609 return out_state 

610 

611# --------------------------------------------------------------------------- 

612# git-checkout 

613# --------------------------------------------------------------------------- 

614 

615 

616_PR_DIFF_MAX_CHARS = 20_000 

617_PR_DIFF_TRUNCATION_SUFFIX = "\n… (truncated)" 

618 

619 

620def _git_argv(*args: str) -> list[str]: 

621 return ["git", "-c", "safe.directory=*", *args] 

622 

623 

624def _require_env(name: str) -> str: 

625 value = os.environ.get(name) 

626 if not value: 

627 raise ValueError(f"git-checkout: required env var {name!r} is not set") 

628 return value 

629 

630 

631def _resolve_default_branch(project_dir: str) -> str: 

632 """Resolve the repo's remote default branch (repo fact, not a fallback). 

633 

634 TICKET profile with an empty BASE_BRANCH: the base branch is the remote 

635 default reported by `git ls-remote --symref origin HEAD` — never a 

636 hardcoded "main" assumption (ADR-0012, contract BASE_BRANCH row). 

637 """ 

638 try: 

639 proc = subprocess.run( 

640 _git_argv("ls-remote", "--symref", "origin", "HEAD"), 

641 cwd=project_dir, 

642 capture_output=True, 

643 text=True, 

644 check=True, 

645 ) 

646 except subprocess.CalledProcessError as exc: 

647 raise ValueError( 

648 f"git-checkout: 'git ls-remote --symref origin HEAD' failed " 

649 f"(exit {exc.returncode}): {(exc.stderr or exc.stdout or '').strip()}" 

650 ) from exc 

651 for line in proc.stdout.splitlines(): 

652 if line.startswith("ref: "): 

653 # Format: "ref: refs/heads/main\tHEAD" 

654 parts = line.split("\t") 

655 ref = parts[0][len("ref: "):].strip() 

656 if ref.startswith("refs/heads/"): 

657 return ref[len("refs/heads/"):] 

658 raise ValueError( 

659 "git-checkout: could not determine the remote default branch from " 

660 "'git ls-remote --symref origin HEAD' (no ref: refs/heads/... HEAD line)" 

661 ) 

662 

663 

664def _ticket_branch(issue_key: str, title: str) -> str: 

665 """Deterministic per-run feature branch for the TICKET profile. 

666 

667 Format (locked): ticket/<issue_key>-<slug(title)>; slugify is the 

668 ADR-0018 slug. Retries and continuation runs reuse the same branch: 

669 git-checkout resumes an existing remote branch instead of creating a 

670 new one per run, and the push is plain — never a forced push. 

671 """ 

672 return f"ticket/{issue_key}-{slugify(title)}" 

673 

674 

675def _remote_branch_exists(remote_url: str, branch: str) -> bool: 

676 """True when the remote already has the branch (TICKET resume path). 

677 

678 `git ls-remote --heads <url> refs/heads/<branch>` runs BEFORE the 

679 clone — a non-empty stdout line means the branch exists and the 

680 checkout resumes it instead of creating a new one (continuation runs 

681 continue the SAME branch and PR). Failure → hard error (fail-fast, 

682 ADR-0012). 

683 """ 

684 try: 

685 proc = subprocess.run( 

686 _git_argv("ls-remote", "--heads", remote_url, f"refs/heads/{branch}"), 

687 capture_output=True, 

688 text=True, 

689 check=True, 

690 ) 

691 except subprocess.CalledProcessError as exc: 

692 raise ValueError( 

693 f"git-checkout: 'git ls-remote --heads {remote_url} refs/heads/{branch}' failed " 

694 f"(exit {exc.returncode}): {(exc.stderr or exc.stdout or '').strip()}" 

695 ) from exc 

696 return bool(proc.stdout.strip()) 

697 

698 

699def _join_pr_diff(patches: list[str]) -> str: 

700 joined = "\n".join(p for p in patches if p) 

701 if len(joined) > _PR_DIFF_MAX_CHARS: 

702 joined = joined[:_PR_DIFF_MAX_CHARS].rstrip() + _PR_DIFF_TRUNCATION_SUFFIX 

703 return joined 

704 

705 

706def _git_checkout(step: dict, state: dict) -> dict: 

707 url = _require_env("GIT_REPO_URL") 

708 token = _require_env("GIT_TOKEN") 

709 run_profile = _require_env("RUN_PROFILE") 

710 

711 if run_profile == "PR": 

712 branch = _require_env("PR_HEAD_BRANCH") 

713 elif run_profile == "BRANCH": 

714 branch = _require_env("BASE_BRANCH") 

715 elif run_profile != "TICKET": 

716 raise ValueError( 

717 f"git-checkout: unsupported RUN_PROFILE {run_profile!r} " 

718 "(expected 'PR', 'BRANCH', or 'TICKET')" 

719 ) 

720 

721 parsed = urlparse(url) 

722 host = parsed.hostname or "" 

723 new_netloc = f"x-access-token:{token}@{host}" 

724 url = urlunparse(parsed._replace(netloc=new_netloc)) 

725 

726 project_dir = "/project" 

727 if os.path.exists(project_dir): 

728 with os.scandir(project_dir) as entries: 

729 for entry in entries: 

730 if entry.is_symlink() or not entry.is_dir(): 

731 os.unlink(entry.path) 

732 else: 

733 shutil.rmtree(entry.path) 

734 

735 if run_profile == "TICKET": 

736 base_branch = os.environ.get("BASE_BRANCH") or "" 

737 issue_key = _require_env("ISSUE_KEY") 

738 title = _require_env("PR_TITLE") 

739 feature_branch = _ticket_branch(issue_key, title) 

740 

741 if _remote_branch_exists(url, feature_branch): 

742 subprocess.run( 

743 _git_argv("clone", "--depth", "1", "--branch", feature_branch, url, project_dir),