fix: require working plans for complex native turns

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

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 turn.protocol_error_message and turn.runtime_status == "completed":
+        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)

openkyrozen/agent/engine.py

@@ -75,12 +75,30 @@ 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"}
+            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:
@@ -106,7 +124,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 +146,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,

tests/test_agent_engine.py

@@ -74,6 +74,144 @@ 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_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")
+        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 +245,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 +305,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 +359,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 +392,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 +429,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 +503,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 +553,13 @@ def assert_tool_schema(tools, name):
             runtime._default_turn.provider = provider
             runtime.set_interaction_mode("agent")
             packet = "<memory_context>frozen packet</memory_context>"
+            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")):