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

1""" 

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

3 

4Toteuttaa ne builtin-komponentit, joita doc-quality.yaml käyttää ja jotka 

5porttaavat doc-quality-pipeline.py:n ei-LLM-logiikan generic-muotoon: 

6 

7- file-collector kerää & luokittelee tiedostot mountista 

8- path-verifier tarkistaa dokumenttien viittaamat polut 

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

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

11- run-finalizer koostaa stats + flushaa Langfusen 

12 

13Kaikki builtinit noudattavat sopimusta: 

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

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

16""" 

17from __future__ import annotations 

18 

19import fnmatch 

20import json 

21import logging 

22import os 

23import re 

24import shutil 

25import subprocess 

26from concurrent.futures import ThreadPoolExecutor, as_completed 

27from datetime import date 

28from pathlib import Path 

29from typing import Any 

30from urllib.parse import urlparse, urlunparse 

31 

32log = logging.getLogger(__name__) 

33 

34 

35# --------------------------------------------------------------------------- 

36# file-collector 

37# --------------------------------------------------------------------------- 

38 

39 

40_DEFAULT_SKIP_DIRS = { 

41 ".git", "node_modules", "__pycache__", ".venv", "venv", 

42 ".mypy_cache", ".pytest_cache", ".ruff_cache", 

43} 

44_DEFAULT_SUFFIXES = {".py", ".sh", ".yaml", ".yml", ".md", ".json", ".toml"} 

45_MAX_FILE_BYTES_DEFAULT = 64 * 1024 

46_MAX_EVIDENCE_DEFAULT = 32 

47 

48 

49def _split_exclude(exclude: list[str] | None) -> tuple[set[str], list[str]]: 

50 """Erottele dir-nimet ja glob-patternit.""" 

51 excl_dirs: set[str] = set(_DEFAULT_SKIP_DIRS) 

52 excl_globs: list[str] = [] 

53 for item in (exclude or []): 

54 if not item: 

55 continue 

56 if "/" in item or "*" in item: 

57 excl_globs.append(item) 

58 else: 

59 excl_dirs.add(item) 

60 return excl_dirs, excl_globs 

61 

62 

63def _list_mount_files(root: Path, excl_dirs: set[str], excl_globs: list[str]) -> list[str]: 

64 out: list[str] = [] 

65 for dirpath, dirnames, filenames in os.walk(root): 

66 rel_dir = Path(dirpath).relative_to(root) 

67 parts = rel_dir.parts 

68 # Skip if any path segment matches excl_dirs 

69 if parts and any(p in excl_dirs for p in parts): 

70 dirnames.clear() 

71 continue 

72 # Skip explicit nested patterns like ".agent-platform/tmp" 

73 rel_dir_posix = rel_dir.as_posix() if parts else "" 

74 if rel_dir_posix and any(fnmatch.fnmatch(rel_dir_posix, g) or rel_dir_posix.startswith(g.rstrip("/*") + "/") for g in excl_globs): 

75 dirnames.clear() 

76 continue 

77 dirnames[:] = [d for d in dirnames if d not in excl_dirs] 

78 for name in filenames: 

79 rel = (rel_dir / name).as_posix() if parts else name 

80 out.append(rel) 

81 return out 

82 

83 

84def _match_any(rel: str, patterns: list[str]) -> bool: 

85 for pat in patterns: 

86 if fnmatch.fnmatch(rel, pat) or rel == pat: 

87 return True 

88 if pat.endswith("/**") and rel.startswith(pat[:-3]): 

89 return True 

90 if pat.endswith("/*") and rel.startswith(pat[:-2]) and "/" not in rel[len(pat) - 2:]: 

91 return True 

92 return False 

93 

94 

95def _classify(rel: str, include: list[str], doc_paths: set[str], ticket_glob: str, evidence_globs: list[str]) -> str | None: 

96 """Luokittele tiedosto: doc / ticket / evidence.""" 

97 # docs: explicit list or glob 

98 if rel in doc_paths or _match_any(rel, [p for p in include if p.startswith("docs/") and p.endswith(".md") and "**" not in p]): 

99 # docs/ai-context.md, docs/architecture.md, docs/adr/*.md 

100 if rel.startswith("docs/") and rel.endswith(".md") and not rel.startswith("docs/tickets/"): 

101 return "doc" 

102 # tickets 

103 if fnmatch.fnmatch(rel, ticket_glob): 

104 return "ticket" 

105 if fnmatch.fnmatch(rel, "docs/tickets/done/*.md"): 

106 return "ticket" 

107 # evidence 

108 if _match_any(rel, evidence_globs): 

109 return "evidence" 

110 return None 

111 

112 

113def _read_file(root: Path, rel: str, max_bytes: int) -> dict: 

114 full = root / rel 

115 truncated = False 

116 content = "" 

117 if full.is_file(): 

118 data = full.read_bytes() 

119 if len(data) > max_bytes: 

120 data = data[:max_bytes] 

121 truncated = True 

122 content = data.decode("utf-8", errors="replace") 

123 return {"path": rel, "content": content, "truncated": truncated} 

124 

125 

126_PIPELINE_SOURCE_FOOTER = "Source: doc-quality" 

127 

128 

129def _ticket_is_audit_input(root: Path, rel: str, filter_mode: str) -> bool: 

130 """human_backlog_only → ohita pipeline-julkaistut, paitsi jos GATED/REVIEWED.""" 

131 if filter_mode != "human_backlog_only": 

132 return True 

133 full = root / rel 

134 if not full.is_file(): 

135 return False 

136 try: 

137 head = full.read_text(encoding="utf-8", errors="replace")[:8000] 

138 except OSError: 

139 return False 

140 if _PIPELINE_SOURCE_FOOTER in head: 

141 if re.search(r"^\*\*Status:\*\*\s*(GATED|REVIEWED)\b", head, re.M | re.I): 

142 return True 

143 return False 

144 return True 

145 

146 

147def file_collector(step: dict, state: dict) -> dict: 

148 """ 

149 params: 

150 path: str (oletus: state.project_root tai /project) 

151 include: [glob] 

152 exclude: [str|glob] 

153 max_evidence_files: int 

154 max_file_bytes: int 

155 allowed_suffixes: [str] 

156 ticket_filter: 'human_backlog_only' | 'all' 

157 output (state[step.id]): 

158 {doc_files: [{path, content, truncated}], ticket_files: [...], evidence_files: [...], stats: {...}} 

159 """ 

160 step_id = step.get("id", "gather") 

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

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

163 include: list[str] = list(params.get("include") or []) 

164 exclude: list[str] = list(params.get("exclude") or []) 

165 max_evidence = int(params.get("max_evidence_files", _MAX_EVIDENCE_DEFAULT)) 

166 max_bytes = int(params.get("max_file_bytes", _MAX_FILE_BYTES_DEFAULT)) 

167 allowed = {s.lower() for s in (params.get("allowed_suffixes") or [])} or set(_DEFAULT_SUFFIXES) 

168 ticket_filter = str(params.get("ticket_filter", "all")) 

169 

170 excl_dirs, excl_globs = _split_exclude(exclude) 

171 all_files = _list_mount_files(root, excl_dirs, excl_globs) 

172 

173 # Doc paths: explicit literal paths from include list ending .md and not under tickets/ 

174 doc_paths: set[str] = {p for p in include if p.endswith(".md") and "**" not in p and "*" not in p and not p.startswith("docs/tickets/")} 

175 # adr glob 

176 adr_globs = [p for p in include if "*" in p and p.endswith(".md") and not p.startswith("docs/tickets/")] 

177 # ticket glob 

178 ticket_glob = next((p for p in include if p.startswith("docs/tickets/") and p.endswith("*.md")), "docs/tickets/*.md") 

179 # evidence patterns (everything else) 

180 evidence_globs = [ 

181 p for p in include 

182 if p not in doc_paths and not p.startswith("docs/tickets/") and not (p in adr_globs) 

183 ] 

184 

185 docs: list[str] = [] 

186 tickets: list[str] = [] 

187 evidence: list[str] = [] 

188 

189 for rel in sorted(all_files): 

190 suffix = Path(rel).suffix.lower() 

191 if suffix and suffix not in allowed: 

192 continue 

193 # doc? 

194 if rel in doc_paths: 

195 docs.append(rel); continue 

196 if any(fnmatch.fnmatch(rel, g) for g in adr_globs): 

197 docs.append(rel); continue 

198 # ticket? 

199 if fnmatch.fnmatch(rel, ticket_glob) or fnmatch.fnmatch(rel, "docs/tickets/done/*.md"): 

200 if _ticket_is_audit_input(root, rel, ticket_filter): 

201 tickets.append(rel) 

202 continue 

203 # evidence? 

204 if _match_any(rel, evidence_globs): 

205 evidence.append(rel); continue 

206 

207 evidence = evidence[:max_evidence] 

208 

209 def _read_all(paths: list[str]) -> list[dict]: 

210 if not paths: 

211 return [] 

212 out: list[dict] = [] 

213 with ThreadPoolExecutor(max_workers=min(16, len(paths))) as ex: 

214 futs = {ex.submit(_read_file, root, p, max_bytes): p for p in paths} 

215 for fut in as_completed(futs): 

216 out.append(fut.result()) 

217 return sorted(out, key=lambda d: d["path"]) 

218 

219 result = { 

220 "doc_files": _read_all(docs), 

221 "ticket_files": _read_all(tickets), 

222 "evidence_files": _read_all(evidence), 

223 "stats": { 

224 "docs": len(docs), 

225 "tickets": len(tickets), 

226 "evidence": len(evidence), 

227 }, 

228 } 

229 log.info("[file-collector] docs=%d tickets=%d evidence=%d", len(docs), len(tickets), len(evidence)) 

230 return {**state, step_id: result, "project_root": str(root)} 

231 

232 

233# --------------------------------------------------------------------------- 

234# path-verifier 

235# --------------------------------------------------------------------------- 

236 

237 

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

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

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

241 return [] 

242 candidates: list[str] = [rel] 

243 if source_doc: 

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

245 if doc_dir and doc_dir != ".": 

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

247 if joined not in candidates: 

248 candidates.append(joined) 

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

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

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

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

253 prefixed = f"docs/{rel}" 

254 if prefixed not in candidates: 

255 candidates.append(prefixed) 

256 return candidates 

257 

258 

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

260 seen: set[str] = set() 

261 out: list[str] = [] 

262 

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

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

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

266 return 

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

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

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

270 return 

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

272 seen.add(p) 

273 out.append(p) 

274 

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

276 add(p) 

277 if len(out) >= max_paths: 

278 return out[:max_paths] 

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

280 add(p) 

281 if len(out) >= max_paths: 

282 break 

283 return out[:max_paths] 

284 

285 

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

287 paths = _paths_in_text(doc_text) 

288 if not paths: 

289 return "" 

290 lines = [ 

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

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

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

294 ] 

295 for p in paths: 

296 resolved = _path_resolution_candidates(p, source_doc) 

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

298 if hit: 

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

300 else: 

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

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

303 return "\n".join(lines) 

304 

305 

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

307 """ 

308 params: 

309 docs_from: dot-path state-avain (esim. "gather.doc_files") 

310 output (state[step.id]): 

311 {by_path: {<doc.path>: <verification_block_string>}} 

312 """ 

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

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

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

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

317 

318 # Naive dot-path lookup against state 

319 cur: Any = state 

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

321 if isinstance(cur, dict): 

322 cur = cur.get(part) 

323 else: 

324 cur = None 

325 break 

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

327 

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

329 for d in docs_list: 

330 if not isinstance(d, dict): 

331 continue 

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

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

334 if path and content: 

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

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

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

338 

339 

340# --------------------------------------------------------------------------- 

341# finding-prefilter 

342# --------------------------------------------------------------------------- 

343 

344 

345_FALSE_MISSING_PHRASES = ( 

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

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

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

349) 

350_NON_ACTIONABLE_PHRASES = ( 

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

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

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

354 "no concrete doc defect", "no documentation defect", 

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

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

357 "not verified to exist", "are not verified to exist", 

358 "is not verified to exist", 

359) 

360_FUTURE_CLAIM_PHRASES = ( 

361 "is in the future", 

362 "which is in the future", 

363 "future date", 

364 "in the future", 

365) 

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

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

368 

369 

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

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

372 if not lower.strip(): 

373 return True 

374 for phrase in _NON_ACTIONABLE_PHRASES: 

375 if phrase in lower: 

376 return True 

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

378 return True 

379 return False 

380 

381 

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

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

384 

385 Auditors occasionally hallucinate temporal direction — labelling a document 

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

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

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

389 """ 

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

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

392 return False 

393 today = date.today() 

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

395 if not dates_found: 

396 return False 

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

398 for raw in dates_found: 

399 try: 

400 if date.fromisoformat(raw) > today: 

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

402 except ValueError: 

403 continue 

404 return True 

405 

406 

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

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

409 if not paths: 

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

411 p = p.strip() 

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

413 paths.append(p) 

414 seen: set[str] = set() 

415 out: list[str] = [] 

416 for p in paths: 

417 if p not in seen: 

418 seen.add(p) 

419 out.append(p) 

420 return out 

421 

422 

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

424 for c in _path_resolution_candidates(rel, source_doc): 

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

426 return True 

427 return False 

428 

429 

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

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

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

433 return r 

434 return "" 

435 

436 

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

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

439 if _is_non_actionable(detail): 

440 return True, "non_actionable" 

441 if _is_false_future_date(detail): 

442 return True, "false_future_date" 

443 lower = detail.lower() 

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

445 

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

447 return True, "meta_ticket_noise" 

448 

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

450 source_doc = _finding_source_doc(finding) 

451 claimed = _claimed_missing_paths(detail) 

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

453 return True, "false_missing_file" 

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

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

456 return True, "false_missing_file" 

457 

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

459 return True, "false_empty_doc_sample" 

460 

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

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

463 "language mixing", "mixed language"): 

464 if noise in lower: 

465 return True, "low_style_noise" 

466 

467 return False, "" 

468 

469 

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

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

472 m = _SEVERITY_RANK.get(min_severity, 1) 

473 return s <= m 

474 

475 

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

477 """ 

478 params: 

479 min_severity: 'high'|'medium'|'low' 

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

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

482 output (state[step.id]): 

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

484 """ 

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

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

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

488 

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

490 finding = step.get("input") 

491 if not isinstance(finding, dict): 

492 finding = params.get("finding") 

493 if not isinstance(finding, dict): 

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

495 

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

497 is_fp, reason = _is_false_positive(root, finding) 

498 if is_fp: 

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

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

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

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

503 

504 

505# --------------------------------------------------------------------------- 

506# ticket-publisher 

507# --------------------------------------------------------------------------- 

508 

509 

510_TICKET_ORIGIN = "doc-quality-pipeline" 

511 

512 

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

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

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

516 return text 

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

518 if m: 

519 insert_at = m.end() 

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

521 return text 

522 

523 

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

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

526 return re.sub( 

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

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

529 text, 

530 count=1, 

531 flags=re.M | re.I, 

532 ) 

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

534 if m: 

535 insert_at = m.end() 

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

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

538 

539 

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

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

542 title = m.group(1).strip() if m else "doc-quality-issue" 

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

544 

545 

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

547 """ 

548 params: 

549 staging_dir: str (oletus .agent-platform/tmp/doc-quality-tickets) 

550 publish_dir: str (oletus docs/tickets) 

551 status_from: str (DRAFT) 

552 status_to: str (GATED) 

553 source_footer: str (Source: doc-quality) 

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

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

556 """ 

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

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

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

560 

561 ticket_text = step.get("input") 

562 if isinstance(ticket_text, dict): 

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

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

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

566 

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

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

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

570 source_footer = str(params.get("source_footer", "Source: doc-quality")) 

571 

572 body = ticket_text.strip() 

573 today = date.today().isoformat() 

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

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

576 # Status DRAFT → GATED 

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

578 body = re.sub( 

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

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

581 body, 

582 count=1, 

583 flags=re.M | re.I, 

584 ) 

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

586 # Inject Status block after first heading 

587 body = re.sub( 

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

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

590 body, 

591 count=1, 

592 flags=re.M, 

593 ) 

594 body = _ensure_ticket_date(body, today) 

595 body = _ensure_ticket_origin(body) 

596 

597 slug = _slug_from_text(body) 

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

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

600 n = 1 

601 while target.exists(): 

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

603 n += 1 

604 

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

606 try: 

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

608 except OSError as exc: 

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

610 

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

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

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

614 

615 

616# --------------------------------------------------------------------------- 

617# run-finalizer 

618# --------------------------------------------------------------------------- 

619 

620 

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

622 """ 

623 Koostaa loppustatuksen + runlog-rivin + flushaa Langfuse. 

624 params: 

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

626 pipeline_id: str 

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

628 """ 

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

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

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

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

633 

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

635 

636 # Aggregate counts from common state fields if present 

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

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

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

640 if isinstance(process_results, list): 

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

642 if not isinstance(r, dict): 

643 return None 

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

645 if r.get("path"): 

646 return r.get("path") 

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

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

649 

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

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

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

653 

654 merged = state.get("merge_findings") 

655 if isinstance(merged, dict): 

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

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

658 

659 gather = state.get("gather") 

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

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

662 stats.setdefault(k, v) 

663 

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

665 if state.get("errors"): 

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

667 

668 # Append runlog line (best-effort) 

669 runlog_path: str | None = None 

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

671 try: 

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

673 line = { 

674 "pipeline": pipeline_id, 

675 "status": run_status, 

676 "stats": stats, 

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

678 } 

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

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

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

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

683 except OSError as exc: 

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

685 

686 out_state = { 

687 **state, 

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

689 "run_status": run_status, 

690 "stats": stats, 

691 }