Coverage for core/langgraph/_envelope.py: 100%

1"""_envelope.py — Core agent output envelope contract and validation. 

2 

3The engine owns a single deterministic definition of the agent output 

4envelope: the status set, the hard-failure error code, the prompt-injection 

5contract text, and the pure validators for the parsed envelope object. 

6 

7This module validates an already-parsed object; JSON text decoding stays in 

8the caller. It intentionally does not import ``_llm`` (which imports this 

9module) to avoid a circular import. 

10""" 

11from __future__ import annotations 

12 

13from collections.abc import Callable 

14from dataclasses import dataclass 

15from typing import Any 

16 

17_ENVELOPE_STATUS_ENUM: tuple[str, ...] = ("ok", "partial", "no_success", "error") 

18 

19ENVELOPE_STATUSES: frozenset[str] = frozenset(_ENVELOPE_STATUS_ENUM) 

20 

21ENVELOPE_ERROR_CODE: str = "AGENT_ERROR" 

22 

23 

24@dataclass(frozen=True) 

25class _EnvelopeField: 

26 """One envelope field: its contract prose and its validator, if any.""" 

27 

28 name: str 

29 contract: str 

30 validate: Callable[[dict], str | None] | None = None 

31 

32 

33def _validate_status(value: dict) -> str | None: 

34 status = value.get("status") 

35 if not isinstance(status, str): 

36 return "status must be a string" 

37 if status not in ENVELOPE_STATUSES: 

38 return f"status must be one of {sorted(ENVELOPE_STATUSES)}" 

39 return None 

40 

41 

42def _validate_error_field(value: dict) -> str | None: 

43 error = value.get("error") 

44 if value.get("status") != "error": 

45 if error is not None: 

46 return "error must be absent or null unless status is 'error'" 

47 return None 

48 if error is None: 

49 return "error is required when status is 'error'" 

50 if not isinstance(error, dict): 

51 return "error must be an object" 

52 return _validate_error_object(error) 

53 

54 

55def _validate_error_object(error: dict) -> str | None: 

56 if not _is_non_empty_string(error.get("code")): 

57 return "error.code must be a non-empty string" 

58 if not _is_non_empty_string(error.get("message")): 

59 return "error.message must be a non-empty string" 

60 if "details" in error and not isinstance(error["details"], dict): 

61 return "error.details must be an object when present" 

62 return None 

63 

64 

65def _validate_calls(value: dict) -> str | None: 

66 calls = value.get("calls") 

67 if calls is None: 

68 return None 

69 if not isinstance(calls, list): 

70 return "calls must be a list or null" 

71 for index, call in enumerate(calls): 

72 error = _validate_call(index, call) 

73 if error is not None: 

74 return error 

75 return None 

76 

77 

78def _validate_iteration(value: dict) -> str | None: 

79 if "iteration" not in value: 

80 return None 

81 if not isinstance(value["iteration"], list): 

82 return "iteration must be a list" 

83 return None 

84 

85 

86def _validate_call(index: int, call: Any) -> str | None: 

87 if not isinstance(call, dict): 

88 return f"calls[{index}] must be an object" 

89 if not _is_non_empty_string(call.get("step")): 

90 return f"calls[{index}].step must be a non-empty string" 

91 if "params" in call and not isinstance(call["params"], dict): 

92 return f"calls[{index}].params must be an object when present" 

93 return None 

94 

95 

96def _is_non_empty_string(value: Any) -> bool: 

97 return isinstance(value, str) and bool(value) 

98 

99 

100def _format_status_enum() -> str: 

101 return ", ".join(f'"{status}"' for status in _ENVELOPE_STATUS_ENUM) 

102 

103 

104_ENVELOPE_FIELDS: tuple[_EnvelopeField, ...] = ( 

105 _EnvelopeField( 

106 name="status", 

107 contract=f'- "status": required. One of {_format_status_enum()}.', 

108 validate=_validate_status, 

109 ), 

110 _EnvelopeField( 

111 name="error", 

112 contract='- "error": required only when "status" is "error"; absent or null ' 

113 "otherwise. When present it is an object with a non-empty string " 

114 '"code", a non-empty string "message", and an optional object "details".', 

115 validate=_validate_error_field, 

116 ), 

117 _EnvelopeField( 

118 name="content", 

119 contract='- "content": optional. Your output. content is an opaque, ' 

120 "unconstrained value. It is not required to be JSON; JSON is merely one " 

121 "possible shape. Core does not type, validate, parse, or interpret it. " 

122 "May be null or absent.", 

123 ), 

124 _EnvelopeField( 

125 name="calls", 

126 contract='- "calls": optional. An ordered list of deterministic-step ' 

127 'invocations, each an object with a non-empty string "step" and an optional ' 

128 'object "params"; preserve order and do not sort; may be null or absent.', 

129 validate=_validate_calls, 

130 ), 

131 _EnvelopeField( 

132 name="iteration", 

133 contract='- "iteration": optional. The LLM-inference-provided list of items ' 

134 "to iterate on the next round; when present it must be a JSON array (an empty " 

135 "array means no iteration); omit it rather than sending null.", 

136 validate=_validate_iteration, 

137 ), 

138) 

139 

140_CONTRACT_HEADER: str = ( 

141 "## Agent output envelope\n" 

142 "\n" 

143 "Respond with exactly one JSON object and nothing else. " 

144 "The object has these fields:\n" 

145 "\n" 

146 "Output ONLY the raw JSON envelope — no markdown code fences and no " 

147 "surrounding prose — because parsing is strict JSON.\n" 

148 "\n" 

149) 

150 

151ENVELOPE_CONTRACT_BLOCK: str = _CONTRACT_HEADER + "\n".join( 

152 field.contract for field in _ENVELOPE_FIELDS 

153) + "\n" 

154 

155 

156def validate_envelope(value: Any) -> str | None: 

157 """Return ``None`` when *value* is a valid envelope, else a short error.""" 

158 if not isinstance(value, dict): 

159 return "envelope must be a JSON object" 

160 

161 for field in _ENVELOPE_FIELDS: 

162 if field.validate is None: 

163 continue 

164 error = field.validate(value) 

165 if error is not None: 

166 return error 

167 return None 

168 

169 

170def is_error_envelope(value: Any) -> bool: 

171 """True iff *value* is a dict whose ``status`` is ``"error"``.""" 

172 return isinstance(value, dict) and value.get("status") == "error"