(
events: list[Event],
)
| 137 | # |
| 138 | # Could be used to compare events for testing resumability. |
| 139 | def simplify_resumable_app_events( |
| 140 | events: list[Event], |
| 141 | ) -> list[(str, Union[types.Part, str])]: |
| 142 | results = [] |
| 143 | for event in events: |
| 144 | if event.content: |
| 145 | results.append((event.author, simplify_content(event.content))) |
| 146 | elif event.output and isinstance(event.output, (str, dict)): |
| 147 | # Single_turn agents strip event.content and set event.output instead. |
| 148 | results.append((event.author, event.output)) |
| 149 | elif event.actions.end_of_agent: |
| 150 | results.append((event.author, END_OF_AGENT)) |
| 151 | elif event.actions.agent_state is not None: |
| 152 | results.append((event.author, event.actions.agent_state)) |
| 153 | return results |
| 154 | |
| 155 | |
| 156 | # Simplifies the contents into a list of (author, simplified_content) tuples. |
nothing calls this directly
no test coverage detected