| 1 | from types import SimpleNamespace |
| 2 | |
| 3 | import pytest |
| 4 | |
| 5 | from agent import LoopData |
| 6 | from extensions.python._functions.agent.Agent.hist_add_ai_response.end._10_log_plain_responses import ( |
| 7 | LogPlainResponses, |
| 8 | ) |
| 9 | from extensions.python.response_stream._10_log_from_stream import ( |
| 10 | LogFromStream as StreamLog, |
| 11 | ) |
| 12 | from extensions.python.reasoning_stream._10_log_from_stream import ( |
| 13 | LogFromStream as ReasoningLog, |
| 14 | ) |
| 15 | from extensions.python.response_stream_end._15_log_from_stream_end import ( |
| 16 | LogFromStream as StreamLogEnd, |
| 17 | ) |
| 18 | from extensions.python.response_stream._20_live_response import LiveResponse |
| 19 | from helpers.dirty_json import DirtyJson |
| 20 | from helpers.log import Log |
| 21 | |
| 22 | |
| 23 | def _agent_with_generating_log(): |
| 24 | log = Log() |
| 25 | item = log.log(type="agent", heading="A0: Calling LLM...", id="msg-1") |
| 26 | agent = SimpleNamespace( |
| 27 | loop_data=SimpleNamespace(params_temporary={"log_item_generating": item}) |
| 28 | ) |
| 29 | return agent, item |
| 30 | |
| 31 | |
| 32 | def test_log_keyword_updates_share_kvp_value_limit(): |
| 33 | item = Log().log(type="agent", heading="A0: Reasoning") |
| 34 | |
| 35 | item.update(reasoning="r" * 6_000, finished=True) |
| 36 | |
| 37 | assert len(item.kvps["reasoning"]) <= 5_000 |
| 38 | assert "Characters hidden" in item.kvps["reasoning"] |
| 39 | assert item.kvps["finished"] is True |
| 40 | |
| 41 | |
| 42 | def test_live_log_updates_omit_unchanged_collections(monkeypatch): |
| 43 | import helpers.log as log_module |
| 44 | |
| 45 | full_dirty: list[str | None] = [] |
| 46 | context_dirty: list[tuple[str, str | None, bool]] = [] |
| 47 | monkeypatch.setattr( |
| 48 | log_module, |
| 49 | "_MARK_DIRTY_ALL", |
| 50 | lambda *, reason=None: full_dirty.append(reason), |
| 51 | ) |
| 52 | monkeypatch.setattr( |
| 53 | log_module, |
| 54 | "_MARK_DIRTY_FOR_CONTEXT", |
| 55 | lambda context_id, *, reason=None, include_collections=True: context_dirty.append( |
| 56 | (context_id, reason, include_collections) |
| 57 | ), |
| 58 | ) |
| 59 | |
| 60 | log = Log() |
| 61 | log.context = SimpleNamespace(id="ctx", streaming_agent=None) |
| 62 | item = log.log(type="agent", heading="Calling LLM") |
| 63 | item.update(content="stream update") |
| 64 | log.set_progress("Receiving") |
| 65 | |
| 66 | assert full_dirty == ["log.Log._notify_state_monitor"] |
| 67 | assert context_dirty == [ |
| 68 | ("ctx", "log.Log._update_item", False), |
| 69 | ("ctx", "log.Log.set_progress", False), |
| 70 | ] |
| 71 | |
| 72 | |
| 73 | def test_responses_plain_text_completion_finishes_generating_log_as_response(): |
| 74 | agent, item = _agent_with_generating_log() |
| 75 | data = { |
| 76 | "args": (agent, "Plain final answer."), |
| 77 | "kwargs": {"id": "msg-1", "llm_result": SimpleNamespace(mode="responses")}, |
| 78 | } |
| 79 | |
| 80 | LogPlainResponses(agent=agent).execute(data=data) |
| 81 | |
| 82 | assert item.type == "response" |
| 83 | assert item.heading == "" |
| 84 | assert item.content == "Plain final answer." |
| 85 | assert item.update_progress == "none" |
| 86 | assert item.kvps["finished"] is True |
| 87 | assert agent.loop_data.params_temporary["log_item_response"] is item |
| 88 | |
| 89 | |
| 90 | def test_responses_tool_json_keeps_generating_log_as_agent_step(): |
| 91 | agent, item = _agent_with_generating_log() |
| 92 | data = { |
| 93 | "args": ( |
| 94 | agent, |
| 95 | '{"tool_name":"search_engine","tool_args":{"query":"today news"}}', |
| 96 | ), |
| 97 | "kwargs": {"id": "msg-1", "llm_result": SimpleNamespace(mode="responses")}, |
| 98 | } |
| 99 | |
| 100 | LogPlainResponses(agent=agent).execute(data=data) |
| 101 | |
| 102 | assert item.type == "agent" |
| 103 | assert item.heading == "A0: Calling LLM..." |
| 104 | assert item.content == "" |
| 105 | assert "log_item_response" not in agent.loop_data.params_temporary |
| 106 | |
| 107 | |
| 108 | def test_responses_plain_json_completion_finishes_generating_log_as_response(): |
| 109 | agent, item = _agent_with_generating_log() |
| 110 | data = { |
| 111 | "args": (agent, '{"status":"ok"}'), |
| 112 | "kwargs": {"id": "msg-1", "llm_result": SimpleNamespace(mode="responses")}, |
| 113 | } |
| 114 | |
| 115 | LogPlainResponses(agent=agent).execute(data=data) |
| 116 | |
| 117 | assert item.type == "response" |
| 118 | assert item.content == '{"status":"ok"}' |
| 119 | assert agent.loop_data.params_temporary["log_item_response"] is item |
| 120 | |
| 121 | |
| 122 | def test_responses_plain_text_completion_does_not_replace_live_response_log(): |
| 123 | agent, item = _agent_with_generating_log() |
| 124 | live_response = Log().log(type="response", content="Already live") |
| 125 | agent.loop_data.params_temporary["log_item_response"] = live_response |
| 126 | data = { |
| 127 | "args": (agent, "Plain final answer."), |
| 128 | "kwargs": {"id": "msg-1", "llm_result": SimpleNamespace(mode="responses")}, |
| 129 | } |
| 130 | |
| 131 | LogPlainResponses(agent=agent).execute(data=data) |
| 132 | |
| 133 | assert item.type == "agent" |
| 134 | assert item.content == "" |
| 135 | assert agent.loop_data.params_temporary["log_item_response"] is live_response |
| 136 | |
| 137 | |
| 138 | @pytest.mark.asyncio |
| 139 | async def test_live_response_renders_single_action_wrapper(): |
| 140 | log = Log() |
| 141 | generating = log.log(type="agent", id="msg-1") |
| 142 | loop_data = SimpleNamespace(params_temporary={"log_item_generating": generating}) |
| 143 | agent = SimpleNamespace( |
| 144 | context=SimpleNamespace(log=log), |
| 145 | agent_name="A0", |
| 146 | ) |
| 147 | |
| 148 | await LiveResponse(agent=agent).execute( |
| 149 | loop_data=loop_data, |
| 150 | parsed={ |
| 151 | "actions": [ |
| 152 | {"tool_name": "response", "tool_args": {"text": "wrapper works"}} |
| 153 | ] |
| 154 | }, |
| 155 | ) |
| 156 | |
| 157 | response = loop_data.params_temporary["log_item_response"] |
| 158 | assert response.type == "response" |
| 159 | assert response.content == "wrapper works" |
| 160 | assert response.id == "msg-1" |
| 161 | |
| 162 | |
| 163 | @pytest.mark.asyncio |
| 164 | async def test_live_response_renders_legacy_message_argument(): |
| 165 | log = Log() |
| 166 | generating = log.log(type="agent", id="msg-1") |
| 167 | loop_data = SimpleNamespace(params_temporary={"log_item_generating": generating}) |
| 168 | agent = SimpleNamespace( |
| 169 | context=SimpleNamespace(log=log), |
| 170 | agent_name="A0", |
| 171 | ) |
| 172 | |
| 173 | await LiveResponse(agent=agent).execute( |
| 174 | loop_data=loop_data, |
| 175 | parsed={"tool_name": "response", "tool_args": {"message": "legacy works"}}, |
| 176 | ) |
| 177 | |
| 178 | response = loop_data.params_temporary["log_item_response"] |
| 179 | assert response.content == "legacy works" |
| 180 | assert response.id == "msg-1" |
| 181 | |
| 182 | |
| 183 | @pytest.mark.asyncio |
| 184 | @pytest.mark.parametrize( |
| 185 | ("stream", "expected_step"), |
| 186 | [ |
| 187 | ( |
| 188 | '{"tool_name":"code_execution_tool","tool_args":', |
| 189 | "Using code_execution_tool...", |
| 190 | ), |
| 191 | ( |
| 192 | '{"tool_name":"code_execution_tool","tool_args":' |
| 193 | '{"runtime":"python","code":', |
| 194 | "Writing Python code... ", |
| 195 | ), |
| 196 | ( |
| 197 | '{"tool_name":"code_execution_tool","tool_args":' |
| 198 | '{"runtime":"python","code":"pri', |
| 199 | "Writing Python code... (3)", |
| 200 | ), |
| 201 | ], |
| 202 | ) |
| 203 | async def test_stream_log_tolerates_partial_tool_arguments( |
| 204 | stream: str, |
| 205 | expected_step: str, |
| 206 | ): |
| 207 | log = Log() |
| 208 | loop_data = LoopData() |
| 209 | agent = SimpleNamespace( |
| 210 | context=SimpleNamespace(log=log), |
| 211 | agent_name="A0", |
| 212 | ) |
| 213 | |
| 214 | await StreamLog(agent=agent).execute( |
| 215 | loop_data=loop_data, |
| 216 | text=stream, |
| 217 | parsed=DirtyJson.parse_string(stream), |
| 218 | ) |
| 219 | |
| 220 | item = loop_data.params_temporary["log_item_generating"] |
| 221 | assert item.kvps["step"] == expected_step |
| 222 | |
| 223 | |
| 224 | @pytest.mark.asyncio |
| 225 | async def test_stream_log_uses_native_reasoning_when_thoughts_are_absent(): |
| 226 | log = Log() |
| 227 | loop_data = LoopData() |
| 228 | agent = SimpleNamespace( |
| 229 | context=SimpleNamespace(log=log), |
| 230 | agent_name="A0", |
| 231 | ) |
| 232 | |
| 233 | await ReasoningLog(agent=agent).execute( |
| 234 | loop_data=loop_data, |
| 235 | text="Native reasoning summary", |
| 236 | ) |
| 237 | await StreamLog(agent=agent).execute( |
| 238 | loop_data=loop_data, |
| 239 | text='{"tool_name":"response","tool_args":{"text":"Done"}}', |
| 240 | parsed={"tool_name": "response", "tool_args": {"text": "Done"}}, |
| 241 | ) |
| 242 | await StreamLogEnd(agent=agent).execute(loop_data=loop_data) |
| 243 | |
| 244 | item = loop_data.params_temporary["log_item_generating"] |
| 245 | assert item.heading == "A0: Using response" |
| 246 | assert item.kvps["thoughts"] == ["Native reasoning summary"] |
| 247 | assert item.kvps["reasoning"] == "Native reasoning summary" |