From 1e0bf5d6781cce2a13104cd6255ce49b5ae28a99 Mon Sep 17 00:00:00 2001 From: Evan Gong Date: Sat, 31 Oct 2026 12:21:11 +0800 Subject: [PATCH 1/3] fix: require working plans before complex native tools --- docs/architecture.md | 7 ++ openkyrozen/agent/completion.py | 5 +- openkyrozen/agent/engine.py | 30 +++++++- tests/test_agent_engine.py | 129 ++++++++++++++++++++++++++++++-- tests/test_task_consistency.py | 41 ++++++++++ 5 files changed, 205 insertions(+), 7 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index e74d2dd7..134589d9 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -161,6 +161,13 @@ call identity and fragment length but redact argument text because observers include remote UI surfaces; those deltas are progress metadata, not replayable invocation input. +For a complex direct native turn, the engine offers only the permitted typed +`update_plan` tool until an active Working Plan exists. It rejects a batch that +also requests other tools, retries a planless final response within the turn's +step limit, and reports an incomplete turn if planning is unavailable or fails. +Existing Strategic Plan and tool permission checks still apply. Once a plan +completes, the model may finish or establish a new plan before further tools. + Permitted feature dependencies follow the responsibilities above: `agent` coordinates `tasks`, `tools` catalog/model contracts, `security`, provider contracts, `routing`, `memory` and `learning`; `tasks` owns its models/storage ports; `memory` owns memory diff --git a/openkyrozen/agent/completion.py b/openkyrozen/agent/completion.py index aba3deaa..0990dbb4 100644 --- a/openkyrozen/agent/completion.py +++ b/openkyrozen/agent/completion.py @@ -80,7 +80,7 @@ def _complete_turn(self, turn): verified_fix = turn.fix_workflow and turn.fix_workflow.get("stage") in {"verified", "explained"} failed_tools = [] if verified_fix else [record for record in turn.tool_records if not record.get("success")] - turn.runtime_status = "incomplete" if failed_tools else "completed" + turn.runtime_status = "incomplete" if failed_tools or turn.protocol_error_message else "completed" if failed_tools: failures = "\n".join( f"- {record.get('action', 'Tool')}: operation failed; its result was not verified." @@ -122,6 +122,9 @@ def _complete_turn(self, turn): status_line += f"\n- {task['status']}: {task['description']}" turn.final_answer = f"{turn.final_answer.rstrip()}\n\n{status_line}" + if failed_tools or turn.protocol_error_message: + turn.runtime_status = "incomplete" + completion_status = getattr(self, "_turn_completion_status", None) if callable(getattr(completion_status, "set", None)): completion_status.set(turn.runtime_status) diff --git a/openkyrozen/agent/engine.py b/openkyrozen/agent/engine.py index 45fb355c..c66d2f79 100644 --- a/openkyrozen/agent/engine.py +++ b/openkyrozen/agent/engine.py @@ -75,12 +75,28 @@ def _run_native(self, turn: TurnContext) -> None: final_text = "" response = None history_tool_names: set[str] = set() + observed_plan_id = None max_steps = getattr(runtime.execution_context, "max_steps", None) if max_steps is None: max_steps = runtime.MAX_STEPS_PER_TURN for step in range(max_steps + 1): tool_names = self._native_tool_names() + plan = getattr(runtime.tasks, "working_plan", None) if turn.planning_required else None + if plan is not None and plan.status == "active": + observed_plan_id = plan.id + completed_plan = (observed_plan_id is not None and plan is not None + and plan.id == observed_plan_id and plan.status == "completed") + needs_plan = (turn.planning_required and not getattr(runtime.execution_context, "child_run_id", None) + and (plan is None or plan.status != "active")) + if needs_plan: + if completed_plan: + tool_names = runtime._permitted_tool_names() & {"update_plan"} + elif "update_plan" not in runtime._permitted_tool_names(): + turn.protocol_error_message = "A Working Plan is required, but planning is not permitted." + break + else: + tool_names = {"update_plan"} token = runtime._record_foreground_model.set(False) try: try: @@ -106,7 +122,9 @@ def _run_native(self, turn: TurnContext) -> None: unknown_tools = [call.name for call in response.tool_calls if call.name not in tool_names] if unknown_tools: - turn.protocol_error_message = "The provider returned an unknown tool call; no tool was run." + turn.protocol_error_message = ("A Working Plan is required before other tools can run." + if needs_plan else + "The provider returned an unknown tool call; no tool was run.") if runtime._conversation_turn_id.get() is not None: runtime._record_conversation_entry("model.failure", { "error_type": "invalid_native_tool_call", @@ -126,6 +144,16 @@ def _run_native(self, turn: TurnContext) -> None: if not response.tool_calls: final_text = response.text.strip() + if needs_plan and not completed_plan: + final_text = "" + if step < max_steps: + messages.extend((response if response.continuation else { + "role": "assistant", "content": response.text, + }, {"role": "user", "content": + "Create a Working Plan with the update_plan tool before continuing."})) + continue + turn.protocol_error_message = "A Working Plan was not established; no further tools were run." + break if (not final_text or response.finish_reason == FinishReason.LENGTH) and step < 2: messages.extend((response if response.continuation else { "role": "assistant", "content": response.text, diff --git a/tests/test_agent_engine.py b/tests/test_agent_engine.py index 69395ef0..4d60f3ba 100644 --- a/tests/test_agent_engine.py +++ b/tests/test_agent_engine.py @@ -74,6 +74,112 @@ def bind_tool_batch(self, runtime, specs=None): runtime._operation_action = lambda name: name runtime._execute_tool_batch = lambda calls, **kwargs: _execute_tool_batch(runtime, calls, **kwargs) + def complex_native_runtime(self): + provider = SimpleNamespace(get_capabilities=lambda _model: ProviderCapabilities(native_tools=True)) + runtime = SimpleNamespace( + execution_context=SimpleNamespace(provider=provider, discovered_tools=frozenset()), + llm_provider=None, DEEPSEEK_MODEL="model", MAX_STEPS_PER_TURN=2, + _record_foreground_model=contextvars.ContextVar("record", default=False), + _conversation_turn_id=contextvars.ContextVar("turn", default=None), + _last_response_is_native=contextvars.ContextVar("native", default=False), + _last_prompt_tokens=0, _last_completion_tokens=0, + tasks=SimpleNamespace(working_plan=None), + ) + runtime._prepare_turn = lambda turn: ( + setattr(turn, "interaction_mode", "agent"), setattr(turn, "complexity", "complex"), + setattr(turn, "learning_run", {"run_id": "run"}), setattr(turn, "learned_context", ""), + ) + runtime._build_messages = lambda *args, **kwargs: [] + runtime._permitted_tool_names = lambda: {"read_file", "update_plan"} + runtime._prompt_profile = lambda: "compact" + runtime._get_model_response = Mock(return_value=ModelResponse( + tool_calls=(ToolCall("read-1", "read_file", {"path": "note.txt"}),), + finish_reason=FinishReason.TOOL_REQUEST, + )) + runtime._execute_tool_batch = Mock() + runtime._deterministic_tool_summary = lambda records: "no tools" + runtime._complete_turn = lambda turn: turn.protocol_error_message or turn.final_answer + return runtime + + def test_complex_native_turn_cannot_run_tools_before_a_working_plan(self): + runtime = self.complex_native_runtime() + + result = AgentEngine(runtime).run(TurnContext("Complex task")) + + runtime._execute_tool_batch.assert_not_called() + self.assertIn("plan", result.lower()) + self.assertEqual(runtime._get_model_response.call_args.kwargs["tool_names"], {"update_plan"}) + + def test_complex_native_turn_blocks_when_planning_is_not_permitted(self): + runtime = self.complex_native_runtime() + runtime._permitted_tool_names = lambda: {"read_file"} + result = AgentEngine(runtime).run(TurnContext("Complex task")) + self.assertIn("planning is not permitted", result) + runtime._get_model_response.assert_not_called() + runtime._execute_tool_batch.assert_not_called() + + def test_complex_native_turn_rejects_plan_and_other_tool_in_same_batch(self): + runtime = self.complex_native_runtime() + runtime._get_model_response.return_value = ModelResponse(tool_calls=( + ToolCall("plan-1", "update_plan", {"complexity": "complex", "tasks": []}), + ToolCall("read-1", "read_file", {"path": "note.txt"}), + ), finish_reason=FinishReason.TOOL_REQUEST) + result = AgentEngine(runtime).run(TurnContext("Complex task")) + self.assertIn("plan", result.lower()) + runtime._execute_tool_batch.assert_not_called() + + def test_complex_native_turn_cannot_continue_after_failed_plan_update(self): + runtime = self.complex_native_runtime() + runtime._get_model_response.side_effect = [ + ModelResponse(tool_calls=(ToolCall("plan-1", "update_plan", {}),), + finish_reason=FinishReason.TOOL_REQUEST), + ModelResponse(tool_calls=(ToolCall("read-1", "read_file", {"path": "note.txt"}),), + finish_reason=FinishReason.TOOL_REQUEST), + ] + result = AgentEngine(runtime).run(TurnContext("Complex task")) + self.assertIn("plan", result.lower()) + self.assertEqual(runtime._execute_tool_batch.call_count, 1) + self.assertEqual(runtime._execute_tool_batch.call_args.args[0][0]["action"], "update_plan") + + def test_complex_native_turn_retries_final_answer_without_plan_then_stops(self): + runtime = self.complex_native_runtime() + runtime._get_model_response.return_value = ModelResponse(text="Done", finish_reason=FinishReason.FINAL) + result = AgentEngine(runtime).run(TurnContext("Complex task")) + self.assertIn("not established", result) + self.assertEqual(runtime._get_model_response.call_count, 3) + runtime._execute_tool_batch.assert_not_called() + + def test_complex_native_turn_runs_other_tools_after_plan_succeeds(self): + runtime = self.complex_native_runtime() + runtime._get_model_response.side_effect = [ + ModelResponse(tool_calls=(ToolCall("plan-1", "update_plan", {"complexity": "complex", "tasks": []}),), + finish_reason=FinishReason.TOOL_REQUEST), + ModelResponse(tool_calls=(ToolCall("read-1", "read_file", {"path": "note.txt"}),), + finish_reason=FinishReason.TOOL_REQUEST), + ModelResponse(text="Done", finish_reason=FinishReason.FINAL), + ] + def execute(calls, **kwargs): + if calls[0]["action"] == "update_plan": + runtime.tasks.working_plan = SimpleNamespace(id="plan-1", status="active") + runtime._execute_tool_batch.side_effect = execute + self.assertEqual(AgentEngine(runtime).run(TurnContext("Complex task")), "Done") + self.assertEqual([call.args[0][0]["action"] for call in runtime._execute_tool_batch.call_args_list], + ["update_plan", "read_file"]) + self.assertEqual(runtime._get_model_response.call_args_list[0].kwargs["tool_names"], {"update_plan"}) + self.assertIn("read_file", runtime._get_model_response.call_args_list[1].kwargs["tool_names"]) + + def test_complex_native_turn_can_finish_after_its_plan_completes(self): + runtime = self.complex_native_runtime() + runtime.tasks.working_plan = SimpleNamespace(id="plan-1", status="active") + runtime._get_model_response.side_effect = [ + ModelResponse(tool_calls=(ToolCall("read-1", "read_file", {"path": "note.txt"}),), + finish_reason=FinishReason.TOOL_REQUEST), + ModelResponse(text="Done", finish_reason=FinishReason.FINAL), + ] + runtime._execute_tool_batch.side_effect = lambda *args, **kwargs: setattr( + runtime.tasks.working_plan, "status", "completed") + self.assertEqual(AgentEngine(runtime).run(TurnContext("Complex task")), "Done") + def test_compact_native_falls_back_to_permitted_inventory_without_discovery_access(self): runtime = SimpleNamespace( execution_context=SimpleNamespace(discovered_tools=frozenset()), @@ -107,7 +213,8 @@ def test_native_history_keeps_past_schema_but_rejects_revoked_new_calls(self): _last_prompt_tokens=0, _last_completion_tokens=0, ) runtime._prepare_turn = lambda turn: ( - setattr(turn, "interaction_mode", "agent"), setattr(turn, "learning_run", {"run_id": "run-1"}), + setattr(turn, "interaction_mode", "agent"), setattr(turn, "complexity", "simple"), + setattr(turn, "learning_run", {"run_id": "run-1"}), setattr(turn, "learned_context", ""), ) runtime._build_messages = lambda *args, **kwargs: [{"role": "user", "content": "Save this note"}] @@ -166,7 +273,8 @@ def test_compact_native_discovery_adds_only_permitted_schemas_for_next_request(s _last_prompt_tokens=0, _last_completion_tokens=0, ) runtime._prepare_turn = lambda turn: ( - setattr(turn, "interaction_mode", "agent"), setattr(turn, "learning_run", {"run_id": "run-1"}), + setattr(turn, "interaction_mode", "agent"), setattr(turn, "complexity", "simple"), + setattr(turn, "learning_run", {"run_id": "run-1"}), setattr(turn, "learned_context", ""), ) runtime._build_messages = lambda *args, **kwargs: [{"role": "user", "content": "Review Git history"}] @@ -219,7 +327,8 @@ def test_native_loop_stops_at_engine_step_limit(self): _last_prompt_tokens=0, _last_completion_tokens=0, ) runtime._prepare_turn = lambda turn: ( - setattr(turn, "interaction_mode", "agent"), setattr(turn, "learning_run", {"run_id": "run-1"}), + setattr(turn, "interaction_mode", "agent"), setattr(turn, "complexity", "simple"), + setattr(turn, "learning_run", {"run_id": "run-1"}), setattr(turn, "learned_context", ""), ) runtime._build_messages = lambda *args, **kwargs: [{"role": "user", "content": "Read README"}] @@ -251,7 +360,8 @@ def test_native_loop_rejects_unknown_tool_without_execution(self): _last_prompt_tokens=0, _last_completion_tokens=0, ) runtime._prepare_turn = lambda turn: ( - setattr(turn, "interaction_mode", "agent"), setattr(turn, "learning_run", {"run_id": "run-1"}), + setattr(turn, "interaction_mode", "agent"), setattr(turn, "complexity", "simple"), + setattr(turn, "learning_run", {"run_id": "run-1"}), setattr(turn, "learned_context", ""), ) runtime._build_messages = lambda *args, **kwargs: [{"role": "user", "content": "Use missing tool"}] @@ -287,7 +397,8 @@ def test_native_turn_uses_typed_calls_and_correlated_results_without_legacy_loop _last_prompt_tokens=2, _last_completion_tokens=1, ) runtime._prepare_turn = lambda turn: ( - setattr(turn, "interaction_mode", "agent"), setattr(turn, "learning_run", {"run_id": "run-1"}), + setattr(turn, "interaction_mode", "agent"), setattr(turn, "complexity", "simple"), + setattr(turn, "learning_run", {"run_id": "run-1"}), setattr(turn, "learned_context", ""), ) runtime._build_messages = lambda *args, **kwargs: [{"role": "user", "content": "Read README"}] @@ -360,6 +471,8 @@ def chat_response(self, messages, model=None, *, tools=None): raise AssertionError("native prompt includes legacy control syntax") if len(self.requests) == 1: self.assert_tool_schema(tools, "update_plan") + if {item["name"] for item in tools} != {"update_plan"}: + raise AssertionError("complex direct turn offered other tools before planning") return ModelResponse( tool_calls=(ToolCall("plan-1", "update_plan", { "complexity": "complex", @@ -408,7 +521,13 @@ def assert_tool_schema(tools, name): runtime._default_turn.provider = provider runtime.set_interaction_mode("agent") packet = "frozen packet" + prepare_turn = runtime._prepare_turn + def prepare_complex(turn): + prepare_turn(turn) + turn.routing_decision = None + turn.complexity = "complex" with patch.object(runtime, "_schedule_post_turn_learning"), \ + patch.object(runtime, "_prepare_turn", side_effect=prepare_complex), \ patch.object(runtime, "_build_memory_context", return_value=packet) as retrieve_memory, \ patch.object(runtime, "_collect_tool_calls", side_effect=AssertionError("legacy parser used")), \ patch.object(runtime, "_collect_unwrapped_tool_calls", side_effect=AssertionError("legacy parser used")): diff --git a/tests/test_task_consistency.py b/tests/test_task_consistency.py index 4ba64caf..4d38eeb2 100644 --- a/tests/test_task_consistency.py +++ b/tests/test_task_consistency.py @@ -1085,6 +1085,47 @@ def test_completion_keeps_verified_answer_without_second_recap_call(self): self.assertEqual(answer, "Verified result") self.assertEqual(recap.call_count, 0) + def test_protocol_error_marks_turn_incomplete_without_tool_receipts(self): + from types import SimpleNamespace + original_manager, original_tasks = main.subagent_manager, main.tasks + main.subagent_manager = SimpleNamespace(coordinator=None) + main.tasks = SimpleNamespace(tasks=[], working_plan=None) + turn = SimpleNamespace( + final_answer="No work was run.", protocol_error_message="A Working Plan is required.", + tool_records=[], turn_prompt_total=0, turn_completion_total=0, turn_start=time.time(), + tool_calls=[], pending_interaction_reply=None, interaction_mode="agent", fix_workflow=None, + user_input="complex task", response_text="", learning_run=None, learning_receipts=[], + complexity="complex", runtime_status="completed", + ) + try: + with patch.object(main, "_advance_fix_workflow", side_effect=lambda state, _request, answer, *_args: (state, answer)), \ + patch.object(main, "_finish_learning_run", side_effect=lambda _run, _receipts, _task, result, *_args: result): + answer = main._complete_turn(turn) + finally: + main.subagent_manager, main.tasks = original_manager, original_tasks + self.assertEqual(answer, "A Working Plan is required.") + self.assertEqual(turn.runtime_status, "incomplete") + + def test_completed_old_plan_does_not_hide_a_protocol_error(self): + from types import SimpleNamespace + original_manager, original_tasks = main.subagent_manager, main.tasks + main.subagent_manager = SimpleNamespace(coordinator=None) + main.tasks = SimpleNamespace(tasks=[], working_plan=SimpleNamespace(status="completed", task_ids=[])) + turn = SimpleNamespace( + final_answer="No work was run.", protocol_error_message="A Working Plan is required.", + tool_records=[], turn_prompt_total=0, turn_completion_total=0, turn_start=time.time(), + tool_calls=[], pending_interaction_reply=None, interaction_mode="agent", fix_workflow=None, + user_input="complex task", response_text="", learning_run=None, learning_receipts=[], + complexity="complex", runtime_status="completed", + ) + try: + with patch.object(main, "_advance_fix_workflow", side_effect=lambda state, _request, answer, *_args: (state, answer)), \ + patch.object(main, "_finish_learning_run", side_effect=lambda _run, _receipts, _task, result, *_args: result): + main._complete_turn(turn) + finally: + main.subagent_manager, main.tasks = original_manager, original_tasks + self.assertEqual(turn.runtime_status, "incomplete") + def test_simple_turn_completion_ignores_an_unrelated_pending_task_ledger(self): from types import SimpleNamespace original_manager, original_tasks = main.subagent_manager, main.tasks From 082ada19796d3029e69a5eb4b641982362be41e5 Mon Sep 17 00:00:00 2001 From: Evan Gong Date: Sat, 31 Oct 2026 12:24:51 +0800 Subject: [PATCH 2/3] fix: preserve blocked plan completion status --- openkyrozen/agent/completion.py | 2 +- tests/test_task_consistency.py | 23 +++++++++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/openkyrozen/agent/completion.py b/openkyrozen/agent/completion.py index 0990dbb4..58309eed 100644 --- a/openkyrozen/agent/completion.py +++ b/openkyrozen/agent/completion.py @@ -122,7 +122,7 @@ def _complete_turn(self, turn): status_line += f"\n- {task['status']}: {task['description']}" turn.final_answer = f"{turn.final_answer.rstrip()}\n\n{status_line}" - if failed_tools or turn.protocol_error_message: + if turn.protocol_error_message and turn.runtime_status == "completed": turn.runtime_status = "incomplete" completion_status = getattr(self, "_turn_completion_status", None) diff --git a/tests/test_task_consistency.py b/tests/test_task_consistency.py index 4d38eeb2..617cac73 100644 --- a/tests/test_task_consistency.py +++ b/tests/test_task_consistency.py @@ -1126,6 +1126,29 @@ def test_completed_old_plan_does_not_hide_a_protocol_error(self): main.subagent_manager, main.tasks = original_manager, original_tasks self.assertEqual(turn.runtime_status, "incomplete") + def test_protocol_error_preserves_blocked_plan_status(self): + from types import SimpleNamespace + original_manager, original_tasks = main.subagent_manager, main.tasks + main.subagent_manager = SimpleNamespace(coordinator=None) + main.tasks = SimpleNamespace( + tasks=[{"id": "task-1", "status": "blocked", "description": "Wait for access"}], + working_plan=SimpleNamespace(status="active", task_ids=["task-1"]), + ) + turn = SimpleNamespace( + final_answer="Work blocked.", protocol_error_message="A Working Plan is required.", + tool_records=[], turn_prompt_total=0, turn_completion_total=0, turn_start=time.time(), + tool_calls=[], pending_interaction_reply=None, interaction_mode="agent", fix_workflow=None, + user_input="complex task", response_text="", learning_run=None, learning_receipts=[], + complexity="complex", runtime_status="completed", + ) + try: + with patch.object(main, "_advance_fix_workflow", side_effect=lambda state, _request, answer, *_args: (state, answer)), \ + patch.object(main, "_finish_learning_run", side_effect=lambda _run, _receipts, _task, result, *_args: result): + main._complete_turn(turn) + finally: + main.subagent_manager, main.tasks = original_manager, original_tasks + self.assertEqual(turn.runtime_status, "blocked") + def test_simple_turn_completion_ignores_an_unrelated_pending_task_ledger(self): from types import SimpleNamespace original_manager, original_tasks = main.subagent_manager, main.tasks From c98800198a92944b3166a1e43cceef7bf55a4d95 Mon Sep 17 00:00:00 2001 From: Evan Gong Date: Sat, 31 Oct 2026 12:28:35 +0800 Subject: [PATCH 3/3] fix: retain permitted plan revisions in compact mode --- openkyrozen/agent/engine.py | 2 ++ tests/test_agent_engine.py | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/openkyrozen/agent/engine.py b/openkyrozen/agent/engine.py index c66d2f79..0a1c8f25 100644 --- a/openkyrozen/agent/engine.py +++ b/openkyrozen/agent/engine.py @@ -97,6 +97,8 @@ def _run_native(self, turn: TurnContext) -> None: break else: tool_names = {"update_plan"} + elif turn.planning_required and not getattr(runtime.execution_context, "child_run_id", None): + tool_names |= runtime._permitted_tool_names() & {"update_plan"} token = runtime._record_foreground_model.set(False) try: try: diff --git a/tests/test_agent_engine.py b/tests/test_agent_engine.py index 4d60f3ba..a4910245 100644 --- a/tests/test_agent_engine.py +++ b/tests/test_agent_engine.py @@ -168,6 +168,38 @@ def execute(calls, **kwargs): self.assertEqual(runtime._get_model_response.call_args_list[0].kwargs["tool_names"], {"update_plan"}) self.assertIn("read_file", runtime._get_model_response.call_args_list[1].kwargs["tool_names"]) + def test_complex_native_turn_can_revise_plan_in_compact_mode(self): + runtime = self.complex_native_runtime() + runtime._permitted_tool_names = lambda: {"read_file", "discover_tools", "update_plan"} + runtime._get_model_response.side_effect = [ + ModelResponse(tool_calls=(ToolCall("plan-1", "update_plan", {"complexity": "complex", "tasks": []}),), + finish_reason=FinishReason.TOOL_REQUEST), + ModelResponse(tool_calls=(ToolCall("plan-2", "update_plan", {"expected_revision": 1, "operations": []}),), + finish_reason=FinishReason.TOOL_REQUEST), + ModelResponse(text="Done", finish_reason=FinishReason.FINAL), + ] + def execute(calls, **kwargs): + runtime.tasks.working_plan = SimpleNamespace(id="plan-1", status="active") + runtime._execute_tool_batch.side_effect = execute + self.assertEqual(AgentEngine(runtime).run(TurnContext("Complex task")), "Done") + self.assertIn("update_plan", runtime._get_model_response.call_args_list[1].kwargs["tool_names"]) + + def test_complex_native_turn_does_not_retain_revoked_plan_permission(self): + runtime = self.complex_native_runtime() + allowed = {"read_file", "discover_tools", "update_plan"} + runtime._permitted_tool_names = lambda: allowed.copy() + runtime._get_model_response.side_effect = [ + ModelResponse(tool_calls=(ToolCall("plan-1", "update_plan", {}),), finish_reason=FinishReason.TOOL_REQUEST), + ModelResponse(tool_calls=(ToolCall("plan-2", "update_plan", {}),), finish_reason=FinishReason.TOOL_REQUEST), + ] + def execute(calls, **kwargs): + runtime.tasks.working_plan = SimpleNamespace(id="plan-1", status="active") + allowed.remove("update_plan") + runtime._execute_tool_batch.side_effect = execute + result = AgentEngine(runtime).run(TurnContext("Complex task")) + self.assertIn("unknown tool", result) + self.assertEqual(runtime._execute_tool_batch.call_count, 1) + def test_complex_native_turn_can_finish_after_its_plan_completes(self): runtime = self.complex_native_runtime() runtime.tasks.working_plan = SimpleNamespace(id="plan-1", status="active")