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_iteration(value: dict) -> str | None: 

66 if "iteration" not in value: 

67 return None 

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

69 return "iteration must be a list" 

70 return None 

71 

72 

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

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

75 

76 

77def _format_status_enum() -> str: 

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

79 

80 

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

82 _EnvelopeField( 

83 name="status", 

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

85 validate=_validate_status, 

86 ), 

87 _EnvelopeField( 

88 name="error", 

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

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

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

92 validate=_validate_error_field, 

93 ), 

94 _EnvelopeField( 

95 name="content", 

96 contract='- "content": optional. Put the result your task instructions asked you to produce here. content is an opaque, ' 

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

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

99 "May be null or absent.", 

100 ), 

101 _EnvelopeField( 

102 name="iteration", 

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

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

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

106 validate=_validate_iteration, 

107 ), 

108) 

109 

110_ENVELOPE_FIELD_NAMES: frozenset[str] = frozenset(field.name for field in _ENVELOPE_FIELDS) 

111 

112_CONTRACT_HEADER: str = ( 

113 "## Agent output envelope\n" 

114 "\n" 

115 "Your task instructions tell you what result to produce (for example a " 

116 "JSON object). Put that result in the \"content\" field of this envelope: " 

117 "the envelope is the single top-level JSON object you output, not your " 

118 "task result itself. The envelope has these fields:\n" 

119 "\n" 

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

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

122 "\n" 

123 "Tool calls are never part of this envelope. A tool call, its name, its " 

124 "arguments, and any tool-call syntax must never appear in this envelope or " 

125 "in its \"content\" field; tool calls travel only through the native " 

126 "tool-call channel (message.tool_calls).\n" 

127 "\n" 

128) 

129 

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

131 field.contract for field in _ENVELOPE_FIELDS 

132) + "\n" 

133 

134 

135def _validate_field_set(value: dict) -> str | None: 

136 unknown = sorted(key for key in value if key not in _ENVELOPE_FIELD_NAMES) 

137 if unknown: 

138 return f"envelope must not contain unknown fields: {unknown}" 

139 return None 

140 

141 

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

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

144 if not isinstance(value, dict): 

145 return "envelope must be a JSON object" 

146 

147 error = _validate_field_set(value) 

148 if error is not None: 

149 return error 

150 

151 for field in _ENVELOPE_FIELDS: 

152 if field.validate is None: 

153 continue 

154 error = field.validate(value) 

155 if error is not None: 

156 return error 

157 return None 

158 

159 

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

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

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