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

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- list-dir-entries deterministinen yhden tason hakemistolistaus 

8- aggregate geneerinen fold-mekanismi (13 operaattoria, dedupe, on_error) 

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

10- run-finalizer koostaa stats + flushaa Langfusen 

11- git-checkout kloonaa repon ja valmistelee työpuun (PR/BRANCH/TICKET) 

12- files-write kirjoittaa {path, content} -entryt workspaceen 

13- files-delete poistaa workspace-tiedostot (tracked → git rm, untracked → unlink) 

14 

15Kaikki builtinit noudattavat sopimusta: 

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

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

18""" 

19from __future__ import annotations 

20 

21import json 

22import logging 

23import os 

24import re 

25import shutil 

26import subprocess 

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# Same dual-import pattern as runner.py (flat sys.path in tests vs qualified 

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

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

37try: 

38 from exec_runner import slugify # noqa: F401 

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

40 from langgraph.exec_runner import slugify # noqa: F401 

41 

42# Files-write-helper (validate_and_write_files / FilesOutputError) — sama 

43# dual-import-kuvio kuin runner.py:ssa: flat sys.path (testit) vs. 

44# langgraph.* (main.py). 

45try: 

46 from _files_output import ( # noqa: F401 

47 FilesOutputError, 

48 _resolve_path, 

49 resolve_workspace_root, 

50 validate_and_write_files, 

51 ) 

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

53 from langgraph._files_output import ( # noqa: F401 

54 FilesOutputError, 

55 _resolve_path, 

56 resolve_workspace_root, 

57 validate_and_write_files, 

58 ) 

59 

60# Shared reducer layer (core/langgraph/_reducers.py) — same dual-import pattern 

61# as the other core modules: flat sys.path (tests) vs. langgraph.* (main.py). 

62try: 

63 from _reducers import VALID_OPERATORS, apply_op # noqa: F401 

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

65 from langgraph._reducers import VALID_OPERATORS, apply_op # noqa: F401 

66 

67 

68_DEFAULT_SKIP_DIRS = { 

69 ".git", "node_modules", "__pycache__", ".venv", "venv", 

70 ".mypy_cache", ".pytest_cache", ".ruff_cache", 

71} 

72 

73 

74# --------------------------------------------------------------------------- 

75# ticket-publisher 

76# --------------------------------------------------------------------------- 

77 

78 

79_TICKET_ORIGIN = "pipeline" 

80 

81 

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

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

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

85 return text 

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

87 if m: 

88 insert_at = m.end() 

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

90 return text 

91 

92 

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

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

95 return re.sub( 

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

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

98 text, 

99 count=1, 

100 flags=re.M | re.I, 

101 ) 

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

103 if m: 

104 insert_at = m.end() 

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

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

107 

108 

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

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

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

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

113 

114 

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

116 """ 

117 params: 

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

119 publish_dir: str (oletus docs/tickets) 

120 status_from: str (DRAFT) 

121 status_to: str (GATED) 

122 source_footer: str (Source: pipeline) 

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

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

125 """ 

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

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

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

129 

130 ticket_text = step.get("input") 

131 if isinstance(ticket_text, dict): 

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

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

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

135 

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

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

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

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

140 

141 body = ticket_text.strip() 

142 today = date.today().isoformat() 

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

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

145 # Status DRAFT → GATED 

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

147 body = re.sub( 

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

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

150 body, 

151 count=1, 

152 flags=re.M | re.I, 

153 ) 

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

155 # Inject Status block after first heading 

156 body = re.sub( 

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

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

159 body, 

160 count=1, 

161 flags=re.M, 

162 ) 

163 body = _ensure_ticket_date(body, today) 

164 body = _ensure_ticket_origin(body) 

165 

166 slug = _slug_from_text(body) 

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

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

169 n = 1 

170 while target.exists(): 

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

172 n += 1 

173 

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

175 try: 

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

177 except OSError as exc: 

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

179 

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

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

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

183 

184 

185# --------------------------------------------------------------------------- 

186# run-finalizer 

187# --------------------------------------------------------------------------- 

188 

189 

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

191 """ 

192 Koostaa loppustatuksen + runlog-rivin + flushaa Langfuse. 

193 params: 

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

195 pipeline_id: str 

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

197 """ 

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

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

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

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

202 

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

204 

205 # Aggregate counts from common state fields if present 

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

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

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

209 if isinstance(process_results, list): 

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

211 if not isinstance(r, dict): 

212 return None 

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

214 if r.get("path"): 

215 return r.get("path") 

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

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

218 

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

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

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

222 

223 merged = state.get("merge_findings") 

224 if isinstance(merged, dict): 

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

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

227 

228 gather = state.get("gather") 

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

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

231 stats.setdefault(k, v) 

232 

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

234 if state.get("errors"): 

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

236 

237 # Append runlog line (best-effort) 

238 runlog_path: str | None = None 

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

240 try: 

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

242 line = { 

243 "pipeline": pipeline_id, 

244 "status": run_status, 

245 "stats": stats, 

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

247 } 

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

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

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

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

252 except OSError as exc: 

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

254 

255 out_state = { 

256 **state, 

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

258 "run_status": run_status, 

259 "stats": stats, 

260 } 

261 return out_state 

262 

263# --------------------------------------------------------------------------- 

264# git-checkout 

265# --------------------------------------------------------------------------- 

266 

267 

268_PR_DIFF_MAX_CHARS = 20_000 

269_PR_DIFF_TRUNCATION_SUFFIX = "\n… (truncated)" 

270 

271 

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

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

274 

275 

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

277 value = os.environ.get(name) 

278 if not value: 

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

280 return value 

281 

282 

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

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

285 

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

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

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

289 """ 

290 try: 

291 proc = subprocess.run( 

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

293 cwd=project_dir, 

294 capture_output=True, 

295 text=True, 

296 check=True, 

297 ) 

298 except subprocess.CalledProcessError as exc: 

299 raise ValueError( 

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

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

302 ) from exc 

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

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

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

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

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

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

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

310 raise ValueError( 

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

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

313 ) 

314 

315 

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

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

318 

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

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

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

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

323 """ 

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

325 

326 

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

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

329 

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

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

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

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

334 ADR-0012). 

335 """ 

336 try: 

337 proc = subprocess.run( 

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

339 capture_output=True, 

340 text=True, 

341 check=True, 

342 ) 

343 except subprocess.CalledProcessError as exc: 

344 raise ValueError( 

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

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

347 ) from exc 

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

349 

350 

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

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

353 if len(joined) > _PR_DIFF_MAX_CHARS: 

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

355 return joined 

356 

357 

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

359 url = _require_env("GIT_REPO_URL") 

360 token = _require_env("GIT_TOKEN") 

361 run_profile = _require_env("RUN_PROFILE") 

362 

363 if run_profile == "PR": 

364 branch = _require_env("PR_HEAD_BRANCH") 

365 elif run_profile == "BRANCH": 

366 branch = _require_env("BASE_BRANCH") 

367 elif run_profile != "TICKET": 

368 raise ValueError( 

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

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

371 ) 

372 

373 parsed = urlparse(url) 

374 host = parsed.hostname or "" 

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

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

377 

378 project_dir = "/project" 

379 if os.path.exists(project_dir): 

380 with os.scandir(project_dir) as entries: 

381 for entry in entries: 

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

383 os.unlink(entry.path) 

384 else: 

385 shutil.rmtree(entry.path) 

386 

387 if run_profile == "TICKET": 

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

389 issue_key = _require_env("ISSUE_KEY") 

390 title = _require_env("PR_TITLE") 

391 feature_branch = _ticket_branch(issue_key, title) 

392 

393 if _remote_branch_exists(url, feature_branch): 

394 subprocess.run( 

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

396 check=True, 

397 ) 

398 if not base_branch: 

399 # Resume + empty BASE_BRANCH: the PR base is the remote default. 

400 base_branch = _resolve_default_branch(project_dir) 

401 else: 

402 if base_branch: 

403 subprocess.run( 

404 _git_argv("clone", "--depth", "1", "--branch", base_branch, url, project_dir), 

405 check=True, 

406 ) 

407 else: 

408 # Empty BASE_BRANCH → clone the default branch, then resolve the 

409 # default branch name as a repo fact (PR base field + state). 

410 subprocess.run( 

411 _git_argv("clone", "--depth", "1", url, project_dir), 

412 check=True, 

413 ) 

414 base_branch = _resolve_default_branch(project_dir) 

415 subprocess.run( 

416 _git_argv("checkout", "-b", feature_branch), 

417 cwd=project_dir, 

418 check=True, 

419 ) 

420 result = {**state, "checkout_done": True, "project_dir": project_dir, 

421 "ticket_branch": feature_branch, "ticket_base_branch": base_branch} 

422 return result 

423 

424 subprocess.run( 

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

426 check=True, 

427 ) 

428 

429 result = {**state, "checkout_done": True, "project_dir": project_dir} 

430 

431 if run_profile == "PR": 

432 base_branch = _require_env("PR_BASE_BRANCH") 

433 diff_base = f"origin/{base_branch}" 

434 

435 def _git(*args: str) -> subprocess.CompletedProcess: 

436 try: 

437 return subprocess.run( 

438 _git_argv(*args), 

439 cwd=project_dir, 

440 capture_output=True, 

441 text=True, 

442 check=True, 

443 ) 

444 except subprocess.CalledProcessError as exc: 

445 detail = (exc.stderr or exc.stdout or "").strip() 

446 raise ValueError( 

447 f"git-checkout: 'git {' '.join(args)}' failed " 

448 f"(exit {exc.returncode}): {detail or 'no output'}" 

449 ) from exc 

450 

451 _git("fetch", "--depth", "1", "origin", f"{base_branch}:refs/remotes/origin/{base_branch}") 

452 changed = _git("diff", "--name-only", f"{diff_base}..HEAD") 

453 diff_out = _git("diff", f"{diff_base}..HEAD") 

454 changed_files = [line for line in changed.stdout.splitlines() if line.strip()] 

455 pr_diff = _join_pr_diff([diff_out.stdout]) 

456 result = {**result, "pr_diff": pr_diff, "changed_files": changed_files} 

457 

458 return result 

459 

460 

461# --------------------------------------------------------------------------- 

462# files-write 

463# --------------------------------------------------------------------------- 

464 

465 

466def _files_write_error(step_id: str, message: str) -> FilesOutputError: 

467 return FilesOutputError(f"files-write step '{step_id}': {message}") 

468 

469 

470def _write_validated_entries(step_id: str, entries: list, workspace_root: Path) -> None: 

471 try: 

472 validate_and_write_files(entries, workspace_root) 

473 except FilesOutputError as exc: 

474 raise _files_write_error(step_id, str(exc)) from exc 

475 

476 

477def files_write(step: dict, state: dict) -> dict: 

478 """Kirjoittaa template-resolvautuneet {path, content}-entryt workspaceen. 

479 

480 params.entries on yksi {path, content}-objekti tai lista; polut ovat 

481 workspace-suhteisia, kirjoituskohde workspace-juuri (state.project_root → 

482 PROJECT_ROOT-env → /project, ADR-0024). Validointi ja kirjoitus 

483 uudelleenkäyttävät validate_and_write_filesia (1 MiB/entry, 10 MiB/step, 

484 polkupako/.git/ulkopako-esto, fail-fast ADR-0012). 

485 """ 

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

487 entries = params.get("entries") 

488 if entries is None or entries == []: 

489 raise _files_write_error( 

490 step["id"], 

491 "'params.entries' is required — single " 

492 "{path, content} object or a list", 

493 ) 

494 if isinstance(entries, dict): 

495 entries = [entries] 

496 elif not isinstance(entries, list): 

497 raise _files_write_error( 

498 step["id"], 

499 f"'params.entries' must be a single " 

500 f"{{path, content}} object or a list, got {type(entries).__name__}", 

501 ) 

502 

503 workspace_root = resolve_workspace_root(state) 

504 _write_validated_entries(step["id"], entries, workspace_root) 

505 written = [entry["path"] for entry in entries] 

506 return {**state, step["id"]: {"written": written, "count": len(written)}} 

507 

508 

509# --------------------------------------------------------------------------- 

510# files-delete 

511# --------------------------------------------------------------------------- 

512 

513 

514def _is_tracked(workspace_root: Path, relative: str) -> bool: 

515 """Onko tiedosto git-jäljitetty workspace-juuresta suhteellisena polkuna. 

516 

517 Ei-repo- tai ei-git-ympäristö → False (untracked-käsittely: pelkkä 

518 tiedostojärjestelmän poisto — enginein add -A-commit ei tarvitse 

519 stagingia poissaolevalle untracked-tiedostolle). 

520 """ 

521 try: 

522 proc = subprocess.run( 

523 ["git", "-c", "safe.directory=*", "ls-files", "--", relative], 

524 cwd=workspace_root, 

525 capture_output=True, 

526 text=True, 

527 check=False, 

528 ) 

529 except FileNotFoundError: 

530 return False 

531 return proc.returncode == 0 and bool(proc.stdout.strip()) 

532 

533 

534def _git_rm(workspace_root: Path, relative: str) -> None: 

535 """Poista tracked-tiedosto ja stageaa poisto (git rm --force -semantiikka). 

536 

537 --force on pakollinen: poisto on nimenomainen tarkoitus, ja 

538 approval-uusinta-ajo KIRJOITTAA split-proposalin uudelleen 

539 (manifestin files-write ajaa ennen delete_ephemeral:a) — työpuussa on 

540 siis muokattu versio, jonka tavallinen git rm kieltäytyisi poistamasta 

541 (approval-run kuolisi deadlock-korjauslooppiin). 

542 """ 

543 try: 

544 subprocess.run( 

545 ["git", "-c", "safe.directory=*", "rm", "--force", "--", relative], 

546 cwd=workspace_root, 

547 capture_output=True, 

548 text=True, 

549 check=True, 

550 ) 

551 except subprocess.CalledProcessError as exc: 

552 raise FilesOutputError( 

553 f"files-delete: 'git rm {relative}' failed (exit {exc.returncode}): " 

554 f"{(exc.stderr or exc.stdout or '').strip()} (ADR-0012)" 

555 ) from exc 

556 except FileNotFoundError as exc: 

557 raise FilesOutputError( 

558 "files-delete: git is not available — cannot stage the deletion " 

559 f"of tracked file '{relative}' (ADR-0012)" 

560 ) from exc 

561 

562 

563def files_delete(step: dict, state: dict) -> dict: 

564 """Poistaa workspace-suhteiset tiedostot (tracked → git rm, untracked → unlink). 

565 

566 params.paths on yksi workspace-suhteinen polkustring tai lista niistä; 

567 poistokohde workspace-juuri (state.project_root → PROJECT_ROOT-env → 

568 /project, ADR-0024). Kaikki polut validoidaan ENSIN (muoto, polkupako, 

569 .git/ulkopako-esto, olemassaolo — fail-fast ADR-0012) ja poistetaan 

570 vasta sitten all-or-nothing: yksikin virheellinen polku → mitään ei 

571 poisteta. Tracked-tiedosto poistetaan git rm:llä (poisto stageattuna — 

572 enginein myöhempi add -A-commit poimii sen), untracked pelkällä 

573 tiedostojärjestelmän poistolla. Tulos: {"deleted": [<polut>], "count"}. 

574 """ 

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

576 paths = params.get("paths") 

577 if isinstance(paths, str): 

578 paths = [paths] 

579 elif not isinstance(paths, list) or not paths: 

580 raise FilesOutputError( 

581 "files-delete: 'params.paths' is required — a single " 

582 "workspace-relative path string or a list of them (ADR-0012)" 

583 ) 

584 

585 workspace_root = resolve_workspace_root(state) 

586 root = workspace_root.resolve(strict=False) 

587 prepared: list[tuple[Path, str, str]] = [] 

588 for index, raw_path in enumerate(paths): 

589 ref = f"path {index}" 

590 if not isinstance(raw_path, str): 

591 raise FilesOutputError( 

592 f"files-delete: {ref} must be a string path, " 

593 f"got {type(raw_path).__name__} (ADR-0012)" 

594 ) 

595 if not raw_path.strip(): 

596 raise FilesOutputError( 

597 f"files-delete: {ref} is empty or whitespace-only — " 

598 "a path is required (ADR-0012)" 

599 ) 

600 if Path(raw_path).is_absolute(): 

601 raise FilesOutputError( 

602 f"files-delete: {ref} '{raw_path}' is absolute — only " 

603 "workspace-relative paths are allowed (ADR-0012)" 

604 ) 

605 resolved = _resolve_path(root, raw_path, root, ref) 

606 if not resolved.is_file(): 

607 raise FilesOutputError( 

608 f"files-delete: {ref} '{raw_path}' is missing or not a " 

609 "file — nothing deleted (ADR-0012)" 

610 ) 

611 prepared.append((resolved, raw_path, ref)) 

612 

613 deleted: list[str] = [] 

614 for resolved, raw_path, ref in prepared: 

615 if _is_tracked(workspace_root, raw_path): 

616 _git_rm(workspace_root, raw_path) 

617 else: 

618 try: 

619 resolved.unlink() 

620 except OSError as exc: 

621 raise FilesOutputError( 

622 f"files-delete: {ref} failed to delete '{raw_path}': " 

623 f"{exc} (ADR-0012)" 

624 ) from exc 

625 deleted.append(raw_path) 

626 

627 return {**state, step["id"]: {"deleted": deleted, "count": len(deleted)}} 

628 

629 

630# --------------------------------------------------------------------------- 

631# list-dir-entries 

632# --------------------------------------------------------------------------- 

633 

634def _hard_failure(code: str, message: str, details: dict | None = None) -> "RunHardFailure": 

635 try: 

636 from runner import RunHardFailure 

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

638 from langgraph.runner import RunHardFailure 

639 return RunHardFailure(code, message, details) 

640 

641 

642def _entry_size(child: Path) -> int: 

643 """Deterministic size for a single top-level entry (dirs use their stat size).""" 

644 try: 

645 return child.stat().st_size 

646 except OSError as exc: 

647 raise _hard_failure( 

648 "LIST_DIR_UNREADABLE", 

649 f"list-dir-entries: cannot stat entry '{child.name}': {exc} (ADR-0012)", 

650 {"path": str(child)}, 

651 ) from exc 

652 

653 

654def list_dir_entries(step: dict, state: dict) -> dict: 

655 """ 

656 Deterministic ONE-level directory listing — structural skip only (ADR-0012). 

657 

658 Lists exactly one level of the resolved `path`; nested subdirectories are 

659 NOT expanded. Only `_DEFAULT_SKIP_DIRS` top-level entries are dropped; no 

660 semantic name/path filtering, no content read. A missing/unreadable/non-dir 

661 root is a hard error (ADR-0012) — never a silent empty list. 

662 

663 params: 

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

665 

666 output (state[step.id]): 

667 {entries: [{path, name, size, kind: "dir"|"file"}], 

668 stats: {total_files, total_dirs, bytes}} 

669 """ 

670 step_id = step.get("id", "list_dir_entries") 

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

672 root = Path( 

673 params.get("path") 

674 or state.get("project_root") 

675 or os.getenv("PROJECT_ROOT", "/project") 

676 ) 

677 if not root.is_dir(): 

678 raise _hard_failure( 

679 "LIST_DIR_MISSING", 

680 f"list-dir-entries: path {str(root)!r} is missing or not a directory (ADR-0012)", 

681 {"step_id": step_id, "path": str(root)}, 

682 ) 

683 try: 

684 children = sorted(root.iterdir(), key=lambda p: p.name) 

685 except OSError as exc: 

686 raise _hard_failure( 

687 "LIST_DIR_UNREADABLE", 

688 f"list-dir-entries: cannot list {str(root)!r}: {exc} (ADR-0012)", 

689 {"step_id": step_id, "path": str(root)}, 

690 ) from exc 

691 

692 entries: list[dict] = [] 

693 total_files = 0 

694 total_dirs = 0 

695 total_bytes = 0 

696 for child in children: 

697 if child.name in _DEFAULT_SKIP_DIRS: 

698 continue 

699 is_dir = child.is_dir() 

700 size = _entry_size(child) 

701 if is_dir: 

702 total_dirs += 1 

703 else: 

704 total_files += 1 

705 total_bytes += size 

706 entries.append({ 

707 "path": child.relative_to(root).as_posix(), 

708 "name": child.name, 

709 "size": size, 

710 "kind": "dir" if is_dir else "file", 

711 }) 

712 

713 entries.sort(key=lambda e: e["path"]) 

714 log.info("[list-dir-entries] entries=%d files=%d dirs=%d", len(entries), total_files, total_dirs) 

715 return { 

716 **state, 

717 step_id: { 

718 "entries": entries, 

719 "stats": { 

720 "total_files": total_files, 

721 "total_dirs": total_dirs, 

722 "bytes": total_bytes, 

723 }, 

724 }, 

725 } 

726 

727 

728# --------------------------------------------------------------------------- 

729# aggregate (core/docs/builtins/aggregate.md) 

730# --------------------------------------------------------------------------- 

731 

732_DEFAULT_OPERATOR = "union" 

733_DEFAULT_ON_ERROR = "keep" 

734_ON_ERROR_POLICIES = frozenset({"error", "skip", "keep"}) 

735_RESERVED_OUTPUT_KEYS = frozenset({"acc"}) 

736_MISSING_PATH = object() 

737 

738 

739def _resolve_dot_path(obj: Any, path: str) -> Any: 

740 """Canonical dot-path resolution; ``_MISSING_PATH`` when unresolved. 

741 

742 Delegates to the runner's strict resolver so an unresolved path is 

743 distinguishable from a path that resolves to ``None`` — the on_error 

744 triggers depend on that difference (ADR-0012). 

745 """ 

746 try: 

747 from runner import _dot_get_strict 

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

749 from langgraph.runner import _dot_get_strict 

750 try: 

751 return _dot_get_strict(obj, path) 

752 except ValueError: 

753 return _MISSING_PATH 

754 

755 

756def _aggregate_failure(message: str, step_id: str, **details: Any) -> "RunHardFailure": 

757 return _hard_failure( 

758 "AGGREGATE_ERROR", 

759 f"aggregate: {message} (ADR-0012)", 

760 {"step_id": step_id, **details}, 

761 ) 

762 

763 

764def _results_param(input_spec: dict, step_id: str) -> Any: 

765 """Resolve the required ``results``/``sources`` alias pair (ADR-0012).""" 

766 results = input_spec.get("results") 

767 sources = input_spec.get("sources") 

768 if results is None and sources is None: 

769 raise _aggregate_failure("missing required 'results'/'sources'", step_id) 

770 if results is not None and sources is not None and results != sources: 

771 raise _aggregate_failure("'results' and 'sources' disagree", step_id) 

772 return results if results is not None else sources 

773 

774 

775def _operator(field: dict, top_op: Any, step_id: str, target: str) -> str: 

776 op = field.get("op") or top_op or _DEFAULT_OPERATOR 

777 if op not in VALID_OPERATORS: 

778 raise _aggregate_failure( 

779 f"unknown operator {op!r}", step_id, target=target, operator=op, 

780 ) 

781 return op 

782 

783 

784def _error_policy(field: dict, top_on_error: Any, step_id: str, target: str) -> str: 

785 policy = field.get("on_error") or top_on_error or _DEFAULT_ON_ERROR 

786 if policy not in _ON_ERROR_POLICIES: 

787 raise _aggregate_failure(