Coverage for core/langgraph/_envelope.py: 99%
139 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-28 00:19 +0000
1"""_envelope.py — Core agent output envelope contract and validation.
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.
7The module also owns the envelope's parse ladder: it fits raw LLM output into
8the locked form via strict JSON, bounded wrapper recovery (object or list) (at most
9``ENVELOPE_MAX_UNWRAP_LAYERS`` quote/code-fence layers), and a raw-content
10fallback. It intentionally does not import ``_llm`` (which imports this
11module) to avoid a circular import.
12"""
13from __future__ import annotations
15import json
16from collections.abc import Callable
17from dataclasses import dataclass
18from typing import Any
20_ENVELOPE_STATUS_ENUM: tuple[str, ...] = ("ok", "partial", "no_success", "error")
22ENVELOPE_STATUSES: frozenset[str] = frozenset(_ENVELOPE_STATUS_ENUM)
24ENVELOPE_ERROR_CODE: str = "AGENT_ERROR"
26ENVELOPE_MAX_UNWRAP_LAYERS: int = 1
29@dataclass(frozen=True)
30class _EnvelopeField:
31 """One envelope field: its contract prose and its validator, if any."""
33 name: str
34 contract: str
35 validate: Callable[[dict], str | None] | None = None
38def _validate_status(value: dict) -> str | None:
39 status = value.get("status")
40 if not isinstance(status, str):
41 return "status must be a string"
42 if status not in ENVELOPE_STATUSES:
43 return f"status must be one of {sorted(ENVELOPE_STATUSES)}"
44 return None
47def _validate_error_field(value: dict) -> str | None:
48 error = value.get("error")
49 if value.get("status") != "error":
50 if error is not None:
51 return "error must be absent or null unless status is 'error'"
52 return None
53 if error is None:
54 return "error is required when status is 'error'"
55 if not isinstance(error, dict):
56 return "error must be an object"
57 return _validate_error_object(error)
60def _validate_error_object(error: dict) -> str | None:
61 if not _is_non_empty_string(error.get("code")):
62 return "error.code must be a non-empty string"
63 if not _is_non_empty_string(error.get("message")):
64 return "error.message must be a non-empty string"
65 if "details" in error and not isinstance(error["details"], dict):
66 return "error.details must be an object when present"
67 return None
70def _validate_iteration(value: dict) -> str | None:
71 if "iteration" not in value:
72 return None
73 if not isinstance(value["iteration"], list):
74 return "iteration must be a list"
75 return None
78def _is_non_empty_string(value: Any) -> bool:
79 return isinstance(value, str) and bool(value)
82def _format_status_enum() -> str:
83 return ", ".join(f'"{status}"' for status in _ENVELOPE_STATUS_ENUM)
86_ENVELOPE_FIELDS: tuple[_EnvelopeField, ...] = (
87 _EnvelopeField(
88 name="status",
89 contract=f'- "status": required. The inference status; one of {_format_status_enum()}.',
90 validate=_validate_status,
91 ),
92 _EnvelopeField(
93 name="error",
94 contract='- "error": the structured failure information; required only when '
95 '"status" is "error"; absent or null otherwise. When present it is an object '
96 'with a non-empty string "code", a non-empty string "message", and an '
97 'optional object "details".',
98 validate=_validate_error_field,
99 ),
100 _EnvelopeField(
101 name="content",
102 contract='- "content": optional. The value handed to the next actor in the pipeline; put the result your task instructions asked you to produce here. content is an opaque, '
103 "unconstrained value. It is not required to be JSON; JSON is merely one "
104 "possible shape. Core does not type, validate, parse, or interpret it. "
105 "May be null or absent.",
106 ),
107 _EnvelopeField(
108 name="iteration",
109 contract='- "iteration": optional. The iteration/topology (fan-out) signal; the '
110 "LLM-inference-provided list of items to iterate on the next round; when "
111 "present it must be a JSON array (an empty array means no iteration); omit it "
112 "rather than sending null.",
113 validate=_validate_iteration,
114 ),
115)
117_ENVELOPE_FIELD_NAMES: frozenset[str] = frozenset(field.name for field in _ENVELOPE_FIELDS)
119_CONTRACT_HEADER: str = (
120 "## Agent output envelope\n"
121 "\n"
122 "Emit the entire envelope — {status, error, content, iteration} — as one "
123 "raw JSON object. The envelope is the single top-level JSON object you "
124 "output, not your task result itself. Your task instructions tell you what "
125 "result to produce (for example a JSON object); put that result in the "
126 "\"content\" field of this envelope. The envelope has these fields:\n"
127 "\n"
128 "Output ONLY the raw JSON envelope — no markdown code fences and no "
129 "surrounding prose.\n"
130 "\n"
131 "Do not include any chain-of-thought or reasoning text outside the "
132 "envelope; keep all reasoning internal and never emit it in the output.\n"
133 "\n"
134 "Tool calls are never part of this envelope. A tool call, its name, its "
135 "arguments, and any tool-call syntax must never appear in this envelope or "
136 "in its \"content\" field; tool calls travel only through the native "
137 "tool-call channel (message.tool_calls).\n"
138 "\n"
139)
141ENVELOPE_CONTRACT_BLOCK: str = _CONTRACT_HEADER + "\n".join(
142 field.contract for field in _ENVELOPE_FIELDS
143) + "\n"
146def _validate_field_set(value: dict) -> str | None:
147 unknown = sorted(key for key in value if key not in _ENVELOPE_FIELD_NAMES)
148 if unknown:
149 return f"envelope must not contain unknown fields: {unknown}"
150 return None
153def validate_envelope(value: Any) -> str | None:
154 """Return ``None`` when *value* is a valid envelope, else a short error."""
155 if not isinstance(value, dict):
156 return "envelope must be a JSON object"
158 error = _validate_field_set(value)
159 if error is not None:
160 return error
162 for field in _ENVELOPE_FIELDS:
163 if field.validate is None:
164 continue
165 error = field.validate(value)
166 if error is not None:
167 return error
168 return None
171def is_error_envelope(value: Any) -> bool:
172 """True iff *value* is a dict whose ``status`` is ``"error"``."""
173 return isinstance(value, dict) and value.get("status") == "error"
176def _strict_parse(text: str) -> tuple[Any, str | None]:
177 """Decode *text* as strict JSON; ``(value, None)`` or ``(None, error)``."""
178 try:
179 return json.loads(text), None
180 except json.JSONDecodeError as error:
181 return None, str(error)
184def _strip_quote_layer(text: str) -> str | None:
185 """Return the string a JSON-string *text* encodes, else ``None``."""
186 value, error = _strict_parse(text)
187 if error is not None or not isinstance(value, str):
188 return None
189 return value
192def _strip_fence_layer(text: str) -> str | None:
193 """Return a whole-output markdown code-fence body, else ``None``."""
194 lines = text.splitlines()
195 if not lines or not _is_fence_opening(lines[0].strip()):
196 return None
197 closing = _last_non_empty_line_index(lines)
198 if closing is None or closing == 0 or lines[closing].strip() != "```":
199 return None
200 return "\n".join(lines[1:closing])
203def _is_fence_opening(line: str) -> bool:
204 if not line.startswith("```"):
205 return False
206 tag = line[3:]
207 return not any(character.isspace() for character in tag) and "`" not in tag
210def _last_non_empty_line_index(lines: list[str]) -> int | None:
211 for index in range(len(lines) - 1, -1, -1):
212 if lines[index].strip():
213 return index
214 return None
217def _strip_one_layer(text: str) -> str | None:
218 """Peel one quote or code-fence wrapping layer; ``None`` if there is none."""
219 quoted = _strip_quote_layer(text)
220 if quoted is not None:
221 return quoted
222 return _strip_fence_layer(text)
225def _recover_json_value(text: str) -> dict | list | None:
226 """Peel up to ``ENVELOPE_MAX_UNWRAP_LAYERS`` wrappers seeking a JSON object or list."""
227 current = text
228 for _ in range(ENVELOPE_MAX_UNWRAP_LAYERS):
229 inner = _strip_one_layer(current)
230 if inner is None:
231 return None
232 value, _error = _strict_parse(inner)
233 if isinstance(value, (dict, list)):
234 return value
235 current = inner
236 return None
239def _classify_object(value: dict, raw: str) -> tuple[dict | None, str | None]:
240 if "status" not in value:
241 return {"status": "ok", "content": raw}, None
242 error = validate_envelope(value)
243 if error is not None:
244 return None, f"ENVELOPE_ERROR: {error}"
245 return value, None
248def parse_agent_envelope(raw: str | None) -> tuple[dict | None, str | None]:
249 """Fit *raw* LLM output into an envelope, or return a technical error.
251 Strict JSON is extended by bounded deterministic recovery: at most
252 ``ENVELOPE_MAX_UNWRAP_LAYERS`` quote/code-fence layers are stripped from an
253 otherwise-valid JSON object or list; a revealed JSON list becomes the
254 ``iteration`` and a revealed object is the envelope. A bare JSON array
255 becomes the ``iteration``.
256 A ``status``-keyed object that fails ``validate_envelope`` is always a
257 technical error, never recovered. Every other output — including prose,
258 malformed JSON, scalars, and objects without ``status`` — becomes
259 ``{"status": "ok", "content": <raw>}`` with the raw output verbatim.
260 """
261 if raw is None or not raw.strip():
262 return None, "empty"
263 text = raw.strip()
264 parsed, _ = _strict_parse(text)
265 if isinstance(parsed, dict):
266 return _classify_object(parsed, raw)
267 recovered = _recover_json_value(text)
268 value = recovered if recovered is not None else parsed
269 if isinstance(value, dict):
270 return _classify_object(value, raw)
271 if isinstance(value, list):
272 return {"status": "ok", "iteration": value}, None
273 return {"status": "ok", "content": raw}, None