From 71eb208a1f992ffdba119b55e88403e64cb31273 Mon Sep 17 00:00:00 2001 From: johnnykor82 Date: Mon, 8 Jun 2026 11:06:06 +0300 Subject: [PATCH] feat: add context engine request hooks Expose optional post-turn observation and pre-request message preparation hooks so context engines can ingest finalized transcript snapshots and assemble request-only context without invoking host compression semantics. --- agent/context_engine.py | 39 +++- agent/conversation_loop.py | 218 +++++++++++++++--- agent/turn_finalizer.py | 16 ++ tests/agent/test_context_engine.py | 26 +++ .../test_context_engine_host_contract.py | 192 +++++++++++++++ tests/run_agent/test_run_agent.py | 119 ++++++++++ 6 files changed, 576 insertions(+), 34 deletions(-) diff --git a/agent/context_engine.py b/agent/context_engine.py index 79c31fb48e6c..b7a905d9d9b1 100644 --- a/agent/context_engine.py +++ b/agent/context_engine.py @@ -21,7 +21,9 @@ 3. update_from_response() called after each API response with usage data 4. should_compress() checked after each turn 5. compress() called when should_compress() returns True - 6. on_session_end() called at real session boundaries (CLI exit, /reset, + 6. on_turn_complete() called after a user turn finishes with the finalized + in-memory transcript snapshot + 7. on_session_end() called at real session boundaries (CLI exit, /reset, gateway session expiry) — NOT per-turn """ @@ -124,6 +126,41 @@ def should_defer_preflight_to_real_usage(self, rough_tokens: int) -> bool: """ return False + # -- Optional: per-turn observation ------------------------------------ + + def on_turn_complete( + self, + messages: List[Dict[str, Any]], + usage: Dict[str, Any] | None = None, + **kwargs: Any, + ) -> None: + """Observe a finished user turn without requesting compaction. + + Hosts call this after the turn has produced its finalized in-memory + transcript snapshot. Engines should treat ``messages`` as read-only: + return values are ignored and this hook must not rely on transcript + mutation for persistence. + """ + return None + + def prepare_request_messages( + self, + request_messages: List[Dict[str, Any]], + *, + conversation_messages: List[Dict[str, Any]] | None = None, + incoming_message: Dict[str, Any] | None = None, + budget_tokens: int = 0, + **kwargs: Any, + ) -> List[Dict[str, Any]] | None: + """Optionally replace the provider request messages for this call. + + Hosts call this after building the API request message list and before + dispatching it to the provider. A returned list is request-only and + must not be treated as persisted transcript state. Return ``None`` to + leave the request unchanged. + """ + return None + # -- Optional: manual /compress preflight ------------------------------ def has_content_to_compress(self, messages: List[Dict[str, Any]]) -> bool: diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 73bed6b0670d..8140665cac98 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -124,6 +124,172 @@ def _ra(): return run_agent +def _snapshot_messages_for_turn_hook( + messages: List[Dict[str, Any]], +) -> List[Dict[str, Any]]: + """Return a shallow transcript copy for observation hooks.""" + return [dict(msg) if isinstance(msg, dict) else msg for msg in messages] + + +def _notify_context_engine_turn_complete( + agent: Any, + messages: List[Dict[str, Any]], + *, + usage: Optional[Dict[str, Any]], + task_id: str, + turn_id: str, + api_call_count: int, + completed: bool, + interrupted: bool, + failed: bool, + turn_exit_reason: str, +) -> None: + """Best-effort notification for context engines that observe turns.""" + engine = getattr(agent, "context_compressor", None) + hook = getattr(engine, "on_turn_complete", None) + if not callable(hook): + return + try: + from agent.context_engine import ContextEngine + + if getattr(hook, "__func__", None) is ContextEngine.on_turn_complete: + return + except Exception: + pass + + try: + hook( + _snapshot_messages_for_turn_hook(messages), + usage=dict(usage) if usage is not None else None, + session_id=getattr(agent, "session_id", None), + task_id=task_id, + turn_id=turn_id, + api_call_count=api_call_count, + completed=completed, + interrupted=interrupted, + failed=failed, + turn_exit_reason=turn_exit_reason, + conversation_id=getattr(agent, "_gateway_session_key", None), + model=getattr(agent, "model", None), + platform=getattr(agent, "platform", None), + ) + except Exception as exc: + logger.warning( + "Context engine on_turn_complete hook failed: %s", + exc, + exc_info=True, + ) + + +def _context_engine_request_budget(agent: Any) -> int: + engine = getattr(agent, "context_compressor", None) + for attr in ("_budget_tokens", "threshold_tokens", "context_length"): + try: + value = int(getattr(engine, attr, 0) or 0) + except Exception: + value = 0 + if value > 0: + return value + return 0 + + +def _apply_context_engine_request_hook( + agent: Any, + request_messages: List[Dict[str, Any]], + *, + conversation_messages: List[Dict[str, Any]], + incoming_message: Optional[Dict[str, Any]], + task_id: str, + turn_id: str, + api_call_count: int, + budget_tokens: int, +) -> tuple[List[Dict[str, Any]], bool]: + """Let a context engine replace provider request messages, fail-open.""" + engine = getattr(agent, "context_compressor", None) + hook = getattr(engine, "prepare_request_messages", None) + if not callable(hook): + return request_messages, False + try: + from agent.context_engine import ContextEngine + + if getattr(hook, "__func__", None) is ContextEngine.prepare_request_messages: + return request_messages, False + except Exception: + pass + + try: + replacement = hook( + _snapshot_messages_for_turn_hook(request_messages), + conversation_messages=_snapshot_messages_for_turn_hook(conversation_messages), + incoming_message=dict(incoming_message) if isinstance(incoming_message, dict) else None, + budget_tokens=budget_tokens, + session_id=getattr(agent, "session_id", None), + task_id=task_id, + turn_id=turn_id, + api_call_count=api_call_count, + conversation_id=getattr(agent, "_gateway_session_key", None), + model=getattr(agent, "model", None), + provider=getattr(agent, "provider", None), + base_url=getattr(agent, "base_url", None), + platform=getattr(agent, "platform", None), + ) + except Exception as exc: + logger.warning( + "Context engine prepare_request_messages hook failed: %s", + exc, + exc_info=True, + ) + return request_messages, False + + if replacement is None: + return request_messages, False + if not isinstance(replacement, list): + logger.warning( + "Context engine prepare_request_messages returned %s, expected list or None", + type(replacement).__name__, + ) + return request_messages, False + return _snapshot_messages_for_turn_hook(replacement), True + + +def _normalize_request_messages_for_api( + api_messages: List[Dict[str, Any]], +) -> None: + """Normalize API-copy messages for stable provider requests.""" + for am in api_messages: + if isinstance(am.get("content"), str): + am["content"] = am["content"].strip() + for am in api_messages: + tcs = am.get("tool_calls") + if not tcs: + continue + new_tcs = [] + for tc in tcs: + if isinstance(tc, dict) and "function" in tc: + try: + args_obj = json.loads(tc["function"]["arguments"]) + tc = {**tc, "function": { + **tc["function"], + "arguments": json.dumps( + args_obj, separators=(",", ":"), + sort_keys=True, + ), + }} + except Exception: + tc["function"]["arguments"] = _repair_tool_call_arguments( + tc["function"]["arguments"], + tc["function"].get("name", "?"), + ) + new_tcs.append(tc) + am["tool_calls"] = new_tcs + + # Proactively strip any surrogate characters before the API call. + # Models served via Ollama (Kimi K2.5, GLM-5, Qwen) can return + # lone surrogates (U+D800-U+DFFF) that crash json.dumps() inside + # the OpenAI SDK. Sanitizing here prevents the 3-retry cycle. + _sanitize_messages_surrogates(api_messages) + + def _nous_entitlement_message(capability: str) -> str: try: from hermes_cli.nous_account import ( @@ -441,6 +607,7 @@ def run_conversation( length_continue_retries = 0 truncated_tool_call_retries = 0 truncated_response_parts: List[str] = [] + last_usage_dict: Optional[Dict[str, Any]] = None compression_attempts = 0 _turn_exit_reason = "unknown" # Diagnostic: why the loop ended @@ -676,12 +843,26 @@ def run_conversation( for idx, pfm in enumerate(agent.prefill_messages): api_messages.insert(sys_offset + idx, pfm.copy()) + api_messages, _ = _apply_context_engine_request_hook( + agent, + api_messages, + conversation_messages=messages, + incoming_message=messages[current_turn_user_idx] + if 0 <= current_turn_user_idx < len(messages) + else None, + task_id=effective_task_id, + turn_id=turn_id, + api_call_count=api_call_count, + budget_tokens=_context_engine_request_budget(agent), + ) + # Apply Anthropic prompt caching for Claude models on native # Anthropic, OpenRouter, and third-party Anthropic-compatible # gateways. Auto-detected: if ``_use_prompt_caching`` is set, # inject cache_control breakpoints (system + last 3 messages) # to reduce input token costs by ~75% on multi-turn - # conversations. + # conversations. Runs after request-only context replacement so + # provider annotations target the final API message list. if agent._use_prompt_caching: api_messages = apply_anthropic_cache_control( api_messages, @@ -711,38 +892,7 @@ def run_conversation( # (llama.cpp, vLLM, Ollama) and improves cache hit rates for # cloud providers. Operates on api_messages (the API copy) so # the original conversation history in `messages` is untouched. - for am in api_messages: - if isinstance(am.get("content"), str): - am["content"] = am["content"].strip() - for am in api_messages: - tcs = am.get("tool_calls") - if not tcs: - continue - new_tcs = [] - for tc in tcs: - if isinstance(tc, dict) and "function" in tc: - try: - args_obj = json.loads(tc["function"]["arguments"]) - tc = {**tc, "function": { - **tc["function"], - "arguments": json.dumps( - args_obj, separators=(",", ":"), - sort_keys=True, - ), - }} - except Exception: - tc["function"]["arguments"] = _repair_tool_call_arguments( - tc["function"]["arguments"], - tc["function"].get("name", "?"), - ) - new_tcs.append(tc) - am["tool_calls"] = new_tcs - - # Proactively strip any surrogate characters before the API call. - # Models served via Ollama (Kimi K2.5, GLM-5, Qwen) can return - # lone surrogates (U+D800-U+DFFF) that crash json.dumps() inside - # the OpenAI SDK. Sanitizing here prevents the 3-retry cycle. - _sanitize_messages_surrogates(api_messages) + _normalize_request_messages_for_api(api_messages) # Calculate approximate request size for logging total_chars = sum(len(str(msg)) for msg in api_messages) @@ -1587,6 +1737,7 @@ def _perform_api_call(next_api_kwargs): "cache_write_tokens": canonical_usage.cache_write_tokens, "reasoning_tokens": canonical_usage.reasoning_tokens, } + last_usage_dict = usage_dict agent.context_compressor.update_from_response(usage_dict) # Cache discovered context length after successful call. @@ -4214,6 +4365,7 @@ def _perform_api_call(next_api_kwargs): original_user_message=original_user_message, _should_review_memory=_should_review_memory, _turn_exit_reason=_turn_exit_reason, + last_usage=last_usage_dict, ) diff --git a/agent/turn_finalizer.py b/agent/turn_finalizer.py index 20db3fcef9f6..add1a5875ecb 100644 --- a/agent/turn_finalizer.py +++ b/agent/turn_finalizer.py @@ -42,6 +42,7 @@ def finalize_turn( original_user_message, _should_review_memory, _turn_exit_reason, + last_usage=None, ): """Run the post-loop finalization and return the turn ``result`` dict. @@ -305,6 +306,21 @@ def finalize_turn( except Exception as exc: logger.warning("post_llm_call hook failed: %s", exc) + from agent.conversation_loop import _notify_context_engine_turn_complete + + _notify_context_engine_turn_complete( + agent, + messages, + usage=last_usage, + task_id=effective_task_id, + turn_id=turn_id, + api_call_count=api_call_count, + completed=completed, + interrupted=interrupted, + failed=failed, + turn_exit_reason=_turn_exit_reason, + ) + # Extract reasoning from the CURRENT turn only. Walk backwards # but stop at the user message that started this turn — anything # earlier is from a prior turn and must not leak into the reasoning diff --git a/tests/agent/test_context_engine.py b/tests/agent/test_context_engine.py index 32acec010c88..5533cc6c68bb 100644 --- a/tests/agent/test_context_engine.py +++ b/tests/agent/test_context_engine.py @@ -124,6 +124,32 @@ def test_should_compress_preflight_default_false(self): engine = StubEngine() assert engine.should_compress_preflight([]) is False + def test_on_turn_complete_default_noop(self): + engine = StubEngine() + messages = [{"role": "user", "content": "hello"}] + + result = engine.on_turn_complete( + messages, + usage={"prompt_tokens": 12, "completion_tokens": 3, "total_tokens": 15}, + session_id="session-123", + ) + + assert result is None + + def test_prepare_request_messages_default_noop(self): + engine = StubEngine() + request_messages = [{"role": "user", "content": "hello"}] + + result = engine.prepare_request_messages( + request_messages, + conversation_messages=request_messages, + incoming_message=request_messages[-1], + budget_tokens=128000, + session_id="session-123", + ) + + assert result is None + # --------------------------------------------------------------------------- # StubEngine behavior diff --git a/tests/agent/test_context_engine_host_contract.py b/tests/agent/test_context_engine_host_contract.py index bb6fb4c41089..a7417feec8f8 100644 --- a/tests/agent/test_context_engine_host_contract.py +++ b/tests/agent/test_context_engine_host_contract.py @@ -26,6 +26,7 @@ from __future__ import annotations +import logging from unittest.mock import MagicMock @@ -195,6 +196,197 @@ def test_update_from_response_forwards_canonical_cache_buckets(): assert usage_dict["output_tokens"] == 500 +def test_on_turn_complete_receives_snapshot_and_metadata(): + """Host forwards a finalized turn snapshot to context engines.""" + from agent.conversation_loop import _notify_context_engine_turn_complete + + captured: dict = {} + + class Engine: + def on_turn_complete(self, messages, usage=None, **kwargs): + captured["messages"] = messages + captured["usage"] = usage + captured["kwargs"] = kwargs + + agent = _bare_agent() + agent.context_compressor = Engine() + messages = [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": "hi"}, + ] + usage = {"prompt_tokens": 10, "completion_tokens": 2, "total_tokens": 12} + + _notify_context_engine_turn_complete( + agent, + messages, + usage=usage, + task_id="task-123", + turn_id="turn-456", + api_call_count=1, + completed=True, + interrupted=False, + failed=False, + turn_exit_reason="text_response(finish_reason=stop)", + ) + + assert captured["messages"] == messages + assert captured["messages"] is not messages + assert captured["messages"][0] is not messages[0] + captured["messages"][0]["content"] = "mutated" + assert messages[0]["content"] == "hello" + assert captured["usage"] == usage + assert captured["usage"] is not usage + assert captured["kwargs"]["session_id"] == "test-session" + assert captured["kwargs"]["task_id"] == "task-123" + assert captured["kwargs"]["turn_id"] == "turn-456" + assert captured["kwargs"]["api_call_count"] == 1 + assert captured["kwargs"]["completed"] is True + assert captured["kwargs"]["interrupted"] is False + assert captured["kwargs"]["failed"] is False + assert captured["kwargs"]["turn_exit_reason"] == "text_response(finish_reason=stop)" + assert captured["kwargs"]["model"] == "fake-model" + assert captured["kwargs"]["platform"] == "telegram" + + +def test_on_turn_complete_failure_is_non_fatal(caplog): + """A context engine observation failure must not break the user turn.""" + from agent.conversation_loop import _notify_context_engine_turn_complete + + class Engine: + def on_turn_complete(self, messages, usage=None, **kwargs): + raise RuntimeError("observer offline") + + agent = _bare_agent() + agent.context_compressor = Engine() + + with caplog.at_level(logging.WARNING): + _notify_context_engine_turn_complete( + agent, + [{"role": "assistant", "content": "done"}], + usage=None, + task_id="task-123", + turn_id="turn-456", + api_call_count=1, + completed=True, + interrupted=False, + failed=False, + turn_exit_reason="text_response(finish_reason=stop)", + ) + + assert "Context engine on_turn_complete hook failed" in caplog.text + + +def test_prepare_request_messages_replaces_request_only(): + """Context engines may replace provider request messages without mutating history.""" + from agent.conversation_loop import _apply_context_engine_request_hook + + captured: dict = {} + + class Engine: + context_length = 200_000 + + def prepare_request_messages(self, request_messages, **kwargs): + captured["request_messages"] = request_messages + captured["kwargs"] = kwargs + request_messages[0]["content"] = "mutated copy" + return [ + {"role": "system", "content": "prepared system"}, + {"role": "user", "content": "prepared user"}, + ] + + agent = _bare_agent() + agent.context_compressor = Engine() + request_messages = [{"role": "user", "content": "hello"}] + conversation_messages = [{"role": "user", "content": "hello"}] + + prepared, changed = _apply_context_engine_request_hook( + agent, + request_messages, + conversation_messages=conversation_messages, + incoming_message=conversation_messages[-1], + task_id="task-123", + turn_id="turn-456", + api_call_count=1, + budget_tokens=12345, + ) + + assert changed is True + assert prepared == [ + {"role": "system", "content": "prepared system"}, + {"role": "user", "content": "prepared user"}, + ] + assert prepared is not captured["request_messages"] + assert request_messages == [{"role": "user", "content": "hello"}] + assert conversation_messages == [{"role": "user", "content": "hello"}] + assert captured["kwargs"]["conversation_messages"] == conversation_messages + assert captured["kwargs"]["conversation_messages"] is not conversation_messages + assert captured["kwargs"]["incoming_message"] == conversation_messages[-1] + assert captured["kwargs"]["incoming_message"] is not conversation_messages[-1] + assert captured["kwargs"]["session_id"] == "test-session" + assert captured["kwargs"]["task_id"] == "task-123" + assert captured["kwargs"]["turn_id"] == "turn-456" + assert captured["kwargs"]["api_call_count"] == 1 + assert captured["kwargs"]["budget_tokens"] == 12345 + assert captured["kwargs"]["model"] == "fake-model" + assert captured["kwargs"]["platform"] == "telegram" + + +def test_prepare_request_messages_none_preserves_original_request(): + """Returning None means identity/no replacement.""" + from agent.conversation_loop import _apply_context_engine_request_hook + + class Engine: + def prepare_request_messages(self, request_messages, **kwargs): + return None + + agent = _bare_agent() + agent.context_compressor = Engine() + request_messages = [{"role": "user", "content": "hello"}] + + prepared, changed = _apply_context_engine_request_hook( + agent, + request_messages, + conversation_messages=[], + incoming_message=None, + task_id="task-123", + turn_id="turn-456", + api_call_count=1, + budget_tokens=0, + ) + + assert changed is False + assert prepared is request_messages + + +def test_prepare_request_messages_failure_is_non_fatal(caplog): + """A request-preparation failure must fail open to the original request.""" + from agent.conversation_loop import _apply_context_engine_request_hook + + class Engine: + def prepare_request_messages(self, request_messages, **kwargs): + raise RuntimeError("retrieval offline") + + agent = _bare_agent() + agent.context_compressor = Engine() + request_messages = [{"role": "user", "content": "hello"}] + + with caplog.at_level(logging.WARNING): + prepared, changed = _apply_context_engine_request_hook( + agent, + request_messages, + conversation_messages=[], + incoming_message=None, + task_id="task-123", + turn_id="turn-456", + api_call_count=1, + budget_tokens=0, + ) + + assert changed is False + assert prepared is request_messages + assert "Context engine prepare_request_messages hook failed" in caplog.text + + def test_discover_context_engines_includes_plugin_registered_engines(monkeypatch): """Plugin-registered context engines appear in the ``hermes plugins`` picker.""" from hermes_cli import plugins_cmd diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index d215c7b193ad..49bdb05a5dd0 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -3373,6 +3373,125 @@ def _setup_agent(self, agent): agent.compression_enabled = False agent.save_trajectories = False + def test_notifies_context_engine_on_turn_complete(self, agent): + self._setup_agent(agent) + agent.context_compressor.on_turn_complete = MagicMock() + agent.client.chat.completions.create.return_value = _mock_response( + content="done", + usage={"prompt_tokens": 10, "completion_tokens": 4, "total_tokens": 14}, + ) + + with ( + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + patch.object(agent, "_sync_external_memory_for_turn"), + patch("hermes_cli.plugins.invoke_hook", return_value=[]), + ): + result = agent.run_conversation("hello", task_id="task-123") + + assert result["completed"] is True + agent.context_compressor.on_turn_complete.assert_called_once() + messages_arg = agent.context_compressor.on_turn_complete.call_args.args[0] + usage_arg = agent.context_compressor.on_turn_complete.call_args.kwargs["usage"] + kwargs = agent.context_compressor.on_turn_complete.call_args.kwargs + + assert [m["role"] for m in messages_arg] == ["user", "assistant"] + assert messages_arg[-1]["content"] == "done" + assert usage_arg["prompt_tokens"] == 10 + assert usage_arg["completion_tokens"] == 4 + assert usage_arg["total_tokens"] == 14 + assert kwargs["session_id"] == agent.session_id + assert kwargs["task_id"] == "task-123" + assert kwargs["completed"] is True + assert kwargs["interrupted"] is False + assert kwargs["failed"] is False + assert kwargs["turn_exit_reason"] == "text_response(finish_reason=stop)" + + def test_context_engine_prepare_request_messages_is_request_only(self, agent): + self._setup_agent(agent) + agent.context_compressor.prepare_request_messages = MagicMock( + return_value=[ + {"role": "system", "content": "prepared system"}, + {"role": "user", "content": "prepared user"}, + ] + ) + agent.client.chat.completions.create.return_value = _mock_response( + content="done", + usage={"prompt_tokens": 8, "completion_tokens": 2, "total_tokens": 10}, + ) + + with ( + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + patch.object(agent, "_sync_external_memory_for_turn"), + patch("hermes_cli.plugins.invoke_hook", return_value=[]), + ): + result = agent.run_conversation("hello", task_id="task-123") + + assert result["completed"] is True + sent_messages = agent.client.chat.completions.create.call_args.kwargs["messages"] + assert sent_messages == [ + {"role": "system", "content": "prepared system"}, + {"role": "user", "content": "prepared user"}, + ] + assert [m["role"] for m in result["messages"]] == ["user", "assistant"] + assert result["messages"][0]["content"] == "hello" + assert result["messages"][-1]["content"] == "done" + agent.context_compressor.prepare_request_messages.assert_called_once() + hook_kwargs = agent.context_compressor.prepare_request_messages.call_args.kwargs + assert hook_kwargs["conversation_messages"][0]["content"] == "hello" + assert hook_kwargs["incoming_message"]["content"] == "hello" + assert hook_kwargs["task_id"] == "task-123" + assert hook_kwargs["session_id"] == agent.session_id + + def test_context_engine_prepare_request_messages_runs_before_prompt_cache(self, agent): + self._setup_agent(agent) + agent._use_prompt_caching = True + agent._cache_ttl = "5m" + agent._use_native_cache_layout = False + agent.context_compressor.prepare_request_messages = MagicMock( + return_value=[ + {"role": "system", "content": "prepared system"}, + {"role": "user", "content": "prepared user"}, + ] + ) + agent.client.chat.completions.create.return_value = _mock_response( + content="done", + usage={"prompt_tokens": 8, "completion_tokens": 2, "total_tokens": 10}, + ) + cached_seen = {} + + def fake_cache_control(messages, **kwargs): + cached_seen["messages"] = [dict(m) for m in messages] + cached = [dict(m) for m in messages] + cached[0]["cache_control"] = {"type": "ephemeral"} + return cached + + with ( + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + patch.object(agent, "_sync_external_memory_for_turn"), + patch("hermes_cli.plugins.invoke_hook", return_value=[]), + patch( + "agent.conversation_loop.apply_anthropic_cache_control", + side_effect=fake_cache_control, + ) as cache_mock, + ): + result = agent.run_conversation("hello", task_id="task-123") + + assert result["completed"] is True + cache_mock.assert_called_once() + assert cached_seen["messages"] == [ + {"role": "system", "content": "prepared system"}, + {"role": "user", "content": "prepared user"}, + ] + sent_messages = agent.client.chat.completions.create.call_args.kwargs["messages"] + assert sent_messages[0]["cache_control"] == {"type": "ephemeral"} + assert sent_messages[0]["content"] == "prepared system" + def test_stop_finish_reason_returns_response(self, agent): self._setup_agent(agent) resp = _mock_response(content="Final answer", finish_reason="stop")