Coverage for core/langgraph/_envelope.py: 98%
153 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-10-04 15:34 +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)
119ENVELOPE_CONTENT_FIELD: str = "content"
121# Channel-mode persistence stores only envelope metadata at ``state[step_id]``
122# (ADR-0023 D9 / ADR-0043 D6), never the opaque ``content`` payload.
123ENVELOPE_METADATA_KEYS: frozenset[str] = _ENVELOPE_FIELD_NAMES - {
124 ENVELOPE_CONTENT_FIELD
125}
128def project_envelope_metadata(value: Any) -> Any:
129 """Return *value* with the opaque ``content`` payload stripped.
131 Channel-mode persistence calls this after channel routing has read the
132 in-flight result, so ``state[step_id]`` holds only envelope metadata and no
133 materialized copy of ``content`` survives. A non-mapping result (an
134 agent-less builtin payload) passes through unchanged; every non-content
135 field is retained because engine internals read them (``error`` /
136 ``parse_error`` for failure detection, body-item ``verdict`` / ``passed``).
137 """
138 if not isinstance(value, dict):
139 return value
140 return {
141 key: item
142 for key, item in value.items()
143 if key != ENVELOPE_CONTENT_FIELD
144 }
147def project_aggregate_metadata(value: Any) -> Any:
148 """Return an aggregate envelope with per-item ``content`` stripped.
150 A parallel/foreach aggregate persisted at ``state[step_id]`` must hold only
151 envelope metadata at every depth (ADR-0023 D9 / ADR-0043 D6): the routed
152 per-item data lives on the native channel, never in the persisted copy.
153 This applies :func:`project_envelope_metadata` to *value* itself and, when
154 the result carries a ``results`` list, to every item. Non-mapping values
155 and mappings without a ``results`` list pass through unchanged; the
156 projection is idempotent.
157 """
158 projected = project_envelope_metadata(value)
159 if not isinstance(projected, dict):
160 return projected
161 results = projected.get("results")
162 if not isinstance(results, list):
163 return projected
164 return {
165 **projected,
166 "results": [project_envelope_metadata(item) for item in results],
167 }
170_CONTRACT_HEADER: str = (
171 "## Agent output envelope\n"
172 "\n"
173 "Emit the entire envelope — {status, error, content, iteration} — as one "
174 "raw JSON object. The envelope is the single top-level JSON object you "
175 "output, not your task result itself. Your task instructions tell you what "
176 "result to produce (for example a JSON object); put that result in the "
177 "\"content\" field of this envelope. The envelope has these fields:\n"
178 "\n"
179 "Output ONLY the raw JSON envelope — no markdown code fences and no "
180 "surrounding prose.\n"
181 "\n"
182 "Do not include any chain-of-thought or reasoning text outside the "
183 "envelope; keep all reasoning internal and never emit it in the output.\n"
184 "\n"
185 "Tool calls are never part of this envelope. A tool call, its name, its "
186 "arguments, and any tool-call syntax must never appear in this envelope or "
187 "in its \"content\" field; tool calls travel only through the native "
188 "tool-call channel (message.tool_calls).\n"
189 "\n"
190)
192ENVELOPE_CONTRACT_BLOCK: str = _CONTRACT_HEADER + "\n".join(
193 field.contract for field in _ENVELOPE_FIELDS
194) + "\n"
197def _validate_field_set(value: dict) -> str | None:
198 unknown = sorted(key for key in value if key not in _ENVELOPE_FIELD_NAMES)
199 if unknown:
200 return f"envelope must not contain unknown fields: {unknown}"
201 return None
204def validate_envelope(value: Any) -> str | None:
205 """Return ``None`` when *value* is a valid envelope, else a short error."""
206 if not isinstance(value, dict):
207 return "envelope must be a JSON object"
209 error = _validate_field_set(value)
210 if error is not None:
211 return error
213 for field in _ENVELOPE_FIELDS:
214 if field.validate is None:
215 continue
216 error = field.validate(value)
217 if error is not None:
218 return error
219 return None
222def is_error_envelope(value: Any) -> bool:
223 """True iff *value* is a dict whose ``status`` is ``"error"``."""
224 return isinstance(value, dict) and value.get("status") == "error"
227def _strict_parse(text: str) -> tuple[Any, str | None]:
228 """Decode *text* as strict JSON; ``(value, None)`` or ``(None, error)``."""
229 try:
230 return json.loads(text), None
231 except json.JSONDecodeError as error:
232 return None, str(error)
235def _strip_quote_layer(text: str) -> str | None:
236 """Return the string a JSON-string *text* encodes, else ``None``."""
237 value, error = _strict_parse(text)
238 if error is not None or not isinstance(value, str):
239 return None
240 return value
243def _strip_fence_layer(text: str) -> str | None:
244 """Return a whole-output markdown code-fence body, else ``None``."""
245 lines = text.splitlines()
246 if not lines or not _is_fence_opening(lines[0].strip()):
247 return None
248 closing = _last_non_empty_line_index(lines)
249 if closing is None or closing == 0 or lines[closing].strip() != "```":
250 return None
251 return "\n".join(lines[1:closing])
254def _is_fence_opening(line: str) -> bool:
255 if not line.startswith("```"):
256 return False
257 tag = line[3:]
258 return not any(character.isspace() for character in tag) and "`" not in tag
261def _last_non_empty_line_index(lines: list[str]) -> int | None:
262 for index in range(len(lines) - 1, -1, -1):
263 if lines[index].strip():
264 return index
265 return None
268def _strip_one_layer(text: str) -> str | None:
269 """Peel one quote or code-fence wrapping layer; ``None`` if there is none."""
270 quoted = _strip_quote_layer(text)
271 if quoted is not None:
272 return quoted
273 return _strip_fence_layer(text)
276def _recover_json_value(text: str) -> dict | list | None:
277 """Peel up to ``ENVELOPE_MAX_UNWRAP_LAYERS`` wrappers seeking a JSON object or list."""
278 current = text
279 for _ in range(ENVELOPE_MAX_UNWRAP_LAYERS):
280 inner = _strip_one_layer(current)
281 if inner is None:
282 return None
283 value, _error = _strict_parse(inner)
284 if isinstance(value, (dict, list)):
285 return value
286 current = inner
287 return None
290def _classify_object(value: dict, raw: str) -> tuple[dict | None, str | None]:
291 if "status" not in value:
292 return {"status": "ok", "content": raw}, None
293 error = validate_envelope(value)
294 if error is not None:
295 return None, f"ENVELOPE_ERROR: {error}"
296 return value, None
299def parse_agent_envelope(raw: str | None) -> tuple[dict | None, str | None]:
300 """Fit *raw* LLM output into an envelope, or return a technical error.
302 Strict JSON is extended by bounded deterministic recovery: at most
303 ``ENVELOPE_MAX_UNWRAP_LAYERS`` quote/code-fence layers are stripped from an
304 otherwise-valid JSON object or list; a revealed JSON list becomes the
305 ``iteration`` and a revealed object is the envelope. A bare JSON array
306 becomes the ``iteration``.
307 A ``status``-keyed object that fails ``validate_envelope`` is always a
308 technical error, never recovered. Every other output — including prose,
309 malformed JSON, scalars, and objects without ``status`` — becomes
310 ``{"status": "ok", "content": <raw>}`` with the raw output verbatim.
311 """
312 if raw is None or not raw.strip():
313 return None, "empty"
314 text = raw.strip()
315 parsed, _ = _strict_parse(text)
316 if isinstance(parsed, dict):
317 return _classify_object(parsed, raw)
318 recovered = _recover_json_value(text)
319 value = recovered if recovered is not None else parsed
320 if isinstance(value, dict):
321 return _classify_object(value, raw)
322 if isinstance(value, list):
323 return {"status": "ok", "iteration": value}, None
324 return {"status": "ok", "content": raw}, None