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 typing import Any 

14 

15ENVELOPE_STATUSES: frozenset[str] = frozenset({"ok", "partial", "no_success", "error"}) 

16 

17ENVELOPE_ERROR_CODE: str = "AGENT_ERROR" 

18 

19ENVELOPE_CONTRACT_BLOCK: str = """## Agent output envelope 

20 

21Respond with exactly one JSON object and nothing else. The object has these fields: 

22 

23- "status": required. One of "ok", "partial", "no_success", "error". 

24- "error": required only when "status" is "error"; absent or null otherwise. When present it is an object with a non-empty string "code", a non-empty string "message", and an optional object "details". 

25- "payload": optional. Your output as any JSON value (string, object, array, number, boolean, or null); may be null or absent. 

26- "calls": optional. An ordered list of deterministic-step invocations, each an object with a non-empty string "step" and an optional object "params"; preserve order and do not sort; may be null or absent. 

27""" 

28 

29 

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

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

32 if not isinstance(value, dict): 

33 return "envelope must be a JSON object" 

34 

35 for check in (_validate_status, _validate_error_field, _validate_calls): 

36 error = check(value) 

37 if error is not None: 

38 return error 

39 return None 

40 

41 

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

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

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

45 

46 

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

48 status = value.get("status") 

49 if not isinstance(status, str): 

50 return "status must be a string" 

51 if status not in ENVELOPE_STATUSES: 

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

53 return None 

54 

55 

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

57 error = value.get("error") 

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

59 if error is not None: 

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

61 return None 

62 if error is None: 

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

64 if not isinstance(error, dict): 

65 return "error must be an object" 

66 return _validate_error_object(error) 

67 

68 

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

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

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

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

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

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

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

76 return None 

77 

78 

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

80 calls = value.get("calls") 

81 if calls is None: 

82 return None 

83 if not isinstance(calls, list): 

84 return "calls must be a list or null" 

85 for index, call in enumerate(calls): 

86 error = _validate_call(index, call) 

87 if error is not None: 

88 return error 

89 return None 

90 

91 

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

93 if not isinstance(call, dict): 

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

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

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

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

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

99 return None 

100 

101 

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

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