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

1""" 

2_data_structure.py — Deterministic communication-context resolution and 

3rendering. 

4 

5Projects a step's declared readable channels (`input_channels`) out of the 

6invocation read overlay and renders them as a distinct communication-context 

7block, separate from the execution context (`input`/`params`). 

8""" 

9from __future__ import annotations 

10 

11import json 

12from typing import Any 

13 

14# Reserved resolved-step key carrying a step's projected COMMUNICATION context 

15# (the declared readable channels' invocation-overlay values). Distinct from 

16# the EXECUTION context (input/params). 

17COMMUNICATION_CONTEXT_KEY = "_communication_context" 

18 

19 

20def resolve_communication_context(step: dict, read_view: dict) -> dict[str, Any]: 

21 """Project a step's declared input channels out of the invocation overlay. 

22 

23 Reads the ordered ``input_channels`` declaration from the step and returns 

24 ``{channel_name: read_view.get(channel_name)}`` in declaration order. An 

25 absent or non-list declaration yields ``{}`` — never whole state. 

26 """ 

27 channels = step.get("input_channels") 

28 if not isinstance(channels, list): 

29 return {} 

30 return {channel: read_view.get(channel) for channel in channels} 

31 

32 

33def render_communication_block(communication: dict) -> str: 

34 return ( 

35 "## Communication context\n\n```json\n" 

36 + json.dumps(communication, indent=2, ensure_ascii=False) 

37 + "\n```" 

38 )