Coverage for core/langgraph/_builtins

1"""

2_builtins_functions.py — deterministic, pure engine builtins for the function

3mechanism (ADR-0025 call-expansion round).

5Implements the two engine-owned pure builtins registered in the deterministic

6builtin register (00-process-contract § Deterministic builtin surface):

8- coverage-check deterministic set comparison: declared units vs the fused

9 read-map package set → verdict + gap list. Pure (no LLM).

10- question-block-form deterministic human-interface question-block formation on

11 exhaustion/stage-escalation → {status, question_block}.

12 Pure (no LLM).

14Both follow the builtin contract `fn(step, state) -> state` and write their

15result at `state[step["id"]]`.

16"""

17from __future__ import annotations

19import logging

20from typing import Any

22log = logging.getLogger(__name__)

24_IDENTITY_KEYS = ("path", "id", "unit", "name")

25_LIST_KEYS = ("units", "read_map", "paths", "covered", "items", "read_map_units")

28def _hashable(value: Any) -> Any:

29 """Return a hashable identity for a scalar; fall back to repr for containers."""

30 if isinstance(value, (dict, list)):

31 return repr(value)

32 return value

35def _identity(unit: Any) -> Any:

36 """The identity of a declared unit or a package entry.

38 A dict is identified by its first present identity key (path/id/unit/name);

39 a scalar is identified as itself.

40 """

41 if isinstance(unit, dict):

42 for key in _IDENTITY_KEYS:

43 if key in unit:

44 return _hashable(unit[key])

45 return repr(unit)

46 return _hashable(unit)

49def _collect_package_identities(package: Any, acc: set) -> set:

50 """Collect the set of unit identities a package covers (its read-map set).

52 A package dict contributes its own identity keys plus any list-shaped

53 read-map/unit/covered entries; nested entries are visited recursively.

54 """

55 if isinstance(package, dict):

56 for key in _IDENTITY_KEYS:

57 if key in package:

58 acc.add(_hashable(package[key]))

59 for key in _LIST_KEYS:

60 if key in package:

61 _collect_package_identities(package[key], acc)

62 elif isinstance(package, list):

63 for item in package:

64 _collect_package_identities(item, acc)

65 else:

66 acc.add(_hashable(package))

67 return acc

70# ---------------------------------------------------------------------------

71# coverage-check

72# ---------------------------------------------------------------------------

75def coverage_check(step: dict, state: dict) -> dict:

76 """Deterministic coverage verdict (CS-UC-03/04).

78 input (resolved) shape:

79 declared: [<unit>, ...]

80 packages: [<package>, ...]

82 output (state[step.id]):

83 declared: <declared> (passthrough)

84 packages: <packages> (passthrough)

85 verdict: "COVERED" | "NOT_COVERED"

86 gaps: [<unit>, ...] (declared units not covered by any package)

88 Empty declared set → NOT_COVERED (Q5 — no vacuous truth): an empty declared

89 set never induces a pass; it is an escalation signal.

90 """

91 step_id = step.get("id", "coverage")

92 inputs = step.get("input") or {}

93 declared = inputs.get("declared") or []

94 packages = inputs.get("packages") or []

96 covered: set = set()

97 for pkg in packages:

98 _collect_package_identities(pkg, covered)

100 gaps: list[Any] = []

101 for unit in declared:

102 if _identity(unit) not in covered:

103 gaps.append(unit)

104

105 verdict = "COVERED" if declared and not gaps else "NOT_COVERED"

106 log.info(

107 "[coverage-check] declared=%d covered=%d gaps=%d verdict=%s",

108 len(declared), len(covered), len(gaps), verdict,

109 )

110 return {**state, step_id: {

111 "declared": declared,

112 "packages": packages,

113 "verdict": verdict,

114 "gaps": gaps,

115 }}

116

117

118# ---------------------------------------------------------------------------

119# question-block-form

120# ---------------------------------------------------------------------------

121

122

123def _fmt_val(value: Any) -> str:

124 if isinstance(value, dict):

125 ident = _identity(value)

126 return str(ident)

127 return str(value)

128

129

130def _collect_problems(inputs: dict) -> tuple[list[str], list[str]]:

131 """Collect (problems, open_questions) from the exhaustion inputs.

132

133 Recognized input keys (all optional): coverage, decompose, research, refine,

134 close_readers, close_synthesis.

135 """

136 problems: list[str] = []

137 questions: list[str] = []

138

139 coverage = inputs.get("coverage")

140 if isinstance(coverage, dict):

141 verdict = coverage.get("verdict")

142 if verdict and verdict != "COVERED":

143 problems.append(f"coverage {verdict}")

144 for gap in coverage.get("gaps") or []:

145 questions.append(f"How to cover unit '{_fmt_val(gap)}'?")

146

147 for key in ("decompose", "research", "refine", "close_readers", "close_synthesis"):

148 src = inputs.get(key)

149 if isinstance(src, dict) and src.get("escalate"):

150 problems.append(f"{key} escalated (agent judged the material insufficient)")

151 q = src.get("question")

152 if q:

153 questions.append(str(q))

154

155 for key in ("refine", "close_readers", "close_synthesis"):

156 src = inputs.get(key)

157 if isinstance(src, dict) and src.get("exhausted"):

158 problems.append(

159 f"{key} exhausted at {src.get('iterations', '?')} iterations "

160 "(bound reached without PASS)"

161 )

162

163 if not problems:

164 problems.append("agentic answering effort exhausted")

165 if not questions:

166 questions.append("Provide the missing input or apply refinements, then re-run.")

167 return problems, questions

168

169

170def question_block_form(step: dict, state: dict) -> dict:

171 """Deterministic human-interface question-block formation (CS-UC-07/LR-UC-03).

172

173 input (resolved): the exhaustion sources — coverage, decompose, research,

174 audit, refine, close_readers, close_synthesis (all optional; input is

175 passed through into the output).

176

177 output (state[step.id]):

178 status: "exhausted"

179 question_block: {problems, open_questions, how_to_proceed, input}

180 """

181 step_id = step.get("id", "qb")

182 inputs = step.get("input") or {}

183 problems, questions = _collect_problems(inputs)

184 how_to_proceed = (

185 "Provide the answers/input above, or apply the refinements, then re-run "

186 "(idempotent) with the outcome in the ticket context."

187 )

188 question_block = {

189 "problems": problems,

190 "open_questions": questions,

191 "how_to_proceed": how_to_proceed,

192 "input": inputs,

193 }

194 return {**state, step_id: {"status": "exhausted", "question_block": question_block}}

195

196

197# ---------------------------------------------------------------------------

198# frontier-aggregate

199# ---------------------------------------------------------------------------

200

201

202def _frontier_uniq(items: Any, acc: list, seen: set) -> None:

203 """Union `items` into `acc`, deduping by path (first wins). A dict is keyed

204 by its `path`; a scalar is keyed by value. Tolerates None/non-iterables."""

205 for it in items or []:

206 if isinstance(it, dict):

207 path = it.get("path")

208 if path and path not in seen:

209 seen.add(path)

210 acc.append(it)

211 elif it is not None and it not in seen:

212 seen.add(it)

213 acc.append(it)

214

215

216def frontier_aggregate(step: dict, state: dict) -> dict:

217 """Deterministic pure union of a BFS level's unit results into the next

218 frontier plus the accumulated file_set/drops (context-file-set, ADR-0012).

219

220 A BFS level fans a dir frontier out to two agent passes (folder-judge and

221 file-examiner); this unions both results lists with the prior BFS

222 accumulator ({file_set, drops, depth} carried in `loop.current`) into the

223 next frontier + the accumulated set. Pure — no side effects; writes its

224 result at state[step.id]. Children are emitted as a deduped list of

225 directory paths (the next frontier, iterated by the next-level foreach).

226

227 input:

228 folder_results: list of {children, drops} — the folder-judge pass results

229 file_results: list of {file_set, drops} — the file-examiner pass results

230 file_set: prior accumulated file_set (list) — loop.current.file_set

231 drops: prior accumulated drops (list) — loop.current.drops

232 depth: prior BFS depth (int) — loop.current.depth

233

234 output (state[step.id]):

235 {frontier: [path,...], file_set: [...], drops: [...],

236 depth: int, empty: bool, stats: {...}}

237 """

238 step_id = step.get("id", "frontier_aggregate")

239 inputs = step.get("input") or {}

240 folder_results = inputs.get("folder_results")

241 file_results = inputs.get("file_results")

242 if folder_results is None:

243 folder_results = []

244 if file_results is None:

245 file_results = []

246 if not isinstance(folder_results, list) or not isinstance(file_results, list):

247 raise ValueError(

248 f"frontier-aggregate: 'folder_results'/'file_results' must be lists "

249 "(ADR-0012)"

250 )

251

252 file_set: list[Any] = []

253 drops: list[Any] = []

254 children: list[Any] = []

255 seen_files: set[Any] = set()

256 seen_drops: set[Any] = set()

257 seen_children: set[Any] = set()

258

259 _frontier_uniq(inputs.get("file_set") or [], file_set, seen_files)

260 _frontier_uniq(inputs.get("drops") or [], drops, seen_drops)

261

262 for res in file_results:

263 if not isinstance(res, dict):

264 raise ValueError(

265 f"frontier-aggregate: 'file_results' item must be a dict, got "

266 f"{type(res).__name__} (ADR-0012)"

267 )

268 _frontier_uniq(res.get("file_set") or [], file_set, seen_files)

269 _frontier_uniq(res.get("drops") or [], drops, seen_drops)

270

271 for res in folder_results:

272 if not isinstance(res, dict):

273 raise ValueError(

274 f"frontier-aggregate: 'folder_results' item must be a dict, got "

275 f"{type(res).__name__} (ADR-0012)"

276 )

277 _frontier_uniq(res.get("drops") or [], drops, seen_drops)

278 for child in res.get("children") or []:

279 path = child if isinstance(child, str) else (

280 child.get("path") if isinstance(child, dict) else None

281 )

282 if path and path not in seen_children:

283 seen_children.add(path)

284 children.append(path)

285

286 prior_depth = inputs.get("depth") or 0

287 try:

288 depth = int(prior_depth) + 1

289 except (TypeError, ValueError):

290 depth = 1

291

292 return {

293 **state,

294 step_id: {

295 "frontier": children,

296 "file_set": file_set,

297 "drops": drops,

298 "depth": depth,

299 "empty": not children,

300 "stats": {

301 "units": len(folder_results),

302 "files": len(file_set),

303 "drops": len(drops),

304 "children": len(children),

305 },

306 },

307 }

308

309

310def frontier_seed(step: dict, state: dict) -> dict:

311 """Deterministic pure seed of the context-file-set BFS frontier (ADR-0012).

312

313 Receives the resolved root of the file-exclusion scan and the one-level

314 `entries` produced by list-dir-entries, and fans the seed into the first

315 BFS level: child dirs become the next frontier (absolute paths built from

316 the `path` param), and the seed's own files become `root_files` (passed

317 through unchanged). Pure — no side effects; writes its result at

318 state[step.id].

319

320 input:

321 path: the listed dir root (path-generic; never hardcoded)

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

323

324 output (state[step.id]):

325 {frontier: [<abs child-dir path>, ...],

326 root_files: [<passed-through file entry dict>, ...]}

327 """

328 step_id = step.get("id", "frontier_split")

329 inputs = step.get("input") or {}

330 path = inputs.get("path") or ""

331 entries = inputs.get("entries") or []

332

333 frontier: list[str] = []

334 root_files: list[dict] = []

335 for entry in entries:

336 if not isinstance(entry, dict):

337 continue

338 if entry.get("kind") == "dir":

339 frontier.append(f"{path}/{entry['name']}")

340 elif entry.get("kind") == "file":

341 root_files.append(entry)

342

343 return {**state, step_id: {"frontier": frontier, "root_files": root_files}}

344

345

346# ---------------------------------------------------------------------------

347# fail-hard

348# ---------------------------------------------------------------------------

349

350

351def _hard_failure(code: str, message: str, details: Any | None = None) -> Exception:

352 try:

353 from runner import RunHardFailure

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

355 from langgraph.runner import RunHardFailure

356 return RunHardFailure(code, message, details)

357

358

359def _compose_fail_message(code: str, details: Any) -> str:

360 """Compose a concise human-facing reason from the auditor refine output.

361

362 Only when `details` carries the expected `{verdict, findings}` shape; any

363 other shape (or no `details`) falls back to a generic reason. The message is

364 derived solely from the supplied inputs — never a hardcoded path, never

365 project-specific (ADR-0033).

366 """

367 if isinstance(details, dict) and details.get("verdict") is not None:

368 findings = details.get("findings")

369 if isinstance(findings, list) and findings:

370 findings_txt = ", ".join(str(f) for f in findings)

371 return f"audit under-minimum: verdict={details['verdict']}; findings={findings_txt}"

372 return f"hard failure: {code}"

373

374

375def fail_hard(step: dict, state: dict) -> dict:

376 """Deterministic control-flow builtin (ADR-0033): unconditionally hard-fails.

377

378 Pure-raised: control-flow only, no side effects, no state mutation — it never

379 returns state; it raises `RunHardFailure`, halting the run (phase failed,

380 exit 1).

381

382 input (resolved) shape (all optional):

383 code: str — hard-failure code (passed through unchanged)

384 message: str — optional human-facing reason (if absent/empty, composed

385 from `details`)

386 details: any — optional payload (passed through unchanged)

387

388 output: none — raises RunHardFailure(code, message, details) instead.

389 """

390 inputs = step.get("input") or {}

391 if not isinstance(inputs, dict):

392 inputs = {}

393 code = inputs.get("code") or "HARD_FAILURE"

394 message = inputs.get("message") or _compose_fail_message(code, inputs.get("details"))

395 raise _hard_failure(code, message, inputs.get("details"))

396

397

398# ---------------------------------------------------------------------------

399# Registry helper

400# ---------------------------------------------------------------------------

401

402

403def register_all(executor: Any) -> None:

404 """Register the function-engine builtins on the executor (hyphen/underscore)."""

Coverage for core/langgraph/_builtins_functions.py: 77%

171 statements