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

1""" 

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

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

4 

5Implements the engine-owned pure builtin registered in the deterministic 

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

7 

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

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

10 Pure (no LLM). 

11 

12It follows the builtin contract `fn(step, state) -> state` and writes its 

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

14""" 

15from __future__ import annotations 

16 

17import logging 

18from typing import Any 

19 

20log = logging.getLogger(__name__) 

21 

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

23 

24 

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

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

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

28 return repr(value) 

29 return value 

30 

31 

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

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

34 

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

36 a scalar is identified as itself. 

37 """ 

38 if isinstance(unit, dict): 

39 for key in _IDENTITY_KEYS: 

40 if key in unit: 

41 return _hashable(unit[key]) 

42 return repr(unit) 

43 return _hashable(unit) 

44 

45 

46# --------------------------------------------------------------------------- 

47# question-block-form 

48# --------------------------------------------------------------------------- 

49 

50 

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

52 if isinstance(value, dict): 

53 ident = _identity(value) 

54 return str(ident) 

55 return str(value) 

56 

57 

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

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

60 

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

62 close_readers, close_synthesis. 

63 """ 

64 problems: list[str] = [] 

65 questions: list[str] = [] 

66 

67 coverage = inputs.get("coverage") 

68 if isinstance(coverage, dict): 

69 verdict = coverage.get("verdict") 

70 if verdict and verdict != "COVERED": 

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

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

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

74 

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

76 src = inputs.get(key) 

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

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

79 q = src.get("question") 

80 if q: 

81 questions.append(str(q)) 

82 

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

84 src = inputs.get(key) 

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

86 problems.append( 

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

88 "(bound reached without PASS)" 

89 ) 

90 

91 if not problems: 

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

93 if not questions: 

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

95 return problems, questions 

96 

97 

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

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

100 

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

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

103 passed through into the output). 

104 

105 output (state[step.id]): 

106 status: "exhausted" 

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

108 """ 

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

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

111 problems, questions = _collect_problems(inputs) 

112 how_to_proceed = ( 

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

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

115 ) 

116 question_block = { 

117 "problems": problems, 

118 "open_questions": questions, 

119 "how_to_proceed": how_to_proceed, 

120 "input": inputs, 

121 } 

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

123 

124 

125# --------------------------------------------------------------------------- 

126# Registry helper 

127# --------------------------------------------------------------------------- 

128 

129 

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

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

132 executor.register("question-block-form", question_block_form)