feat: expose guarded native Working Plan revisions

docs/tasks-and-automation.md

@@ -13,6 +13,8 @@ Task IDs and records are actor/workspace/session scoped. The web task API expose
 
 Simple work does not require a plan, and a plan is optional for medium work. Complex direct work uses a durable Working Plan with runtime-owned stable IDs and explicit acceptance criteria. The typed `update_plan` boundary applies revision-checked batches of add, update, cancel, reorder, dependency, split, merge, replace, block, and unblock operations; legacy `complexity`/`tasks` input is adapted through the same validation path. Splitting, merging, or replacing retains superseded tasks as cancelled history and records both directions of the relationship. Blocking requires a reason and prevents scheduling until explicitly unblocked with a reason. Each structural revision and its optional rationale are written atomically with a complete plan/task snapshot in the event history. A changed task must still be pending and unexecuted; executed task records and evidence are preserved. Scheduling follows plan order and waits for every dependency to succeed. A plan completes only when every task succeeds with acceptance evidence; recovery also reconciles a crash between the final task write and the plan status write. Clearing an active plan cancels its unfinished steps. The foreground scheduler runs consecutive calls concurrently only when every call is catalog-marked safe and read-only; all other calls preserve response order, and task receipts are reconciled in that order. Assistant answers remain available when a complex turn ends with unfinished work; its runtime outcome is reported separately as completed, incomplete, blocked, or cancelled. This outcome uses only that turn's Working Plan, so an unrelated task ledger cannot make a simple answer appear unfinished. The native foreground loop uses typed provider calls; the legacy text protocol remains available through its compatibility adapter.
 
+In native Agent mode, the model can call the structured `update_plan` tool when its effective capabilities include `write`. Ask and Plan modes never receive that schema. Tool authorization still intersects the active capability token, and an accepted Strategic Plan remains bound to its objective, revision, paths, effects, and approval state; plan editing does not grant permission to perform the revised tasks.
+
 The TUI and Web chat show the current Working Plan revision, its user-facing revision reason, and tasks in scheduler order. Progress is updated from typed task events; successful tasks are shown as verified, and cancelled replacements remain marked as superseded. The Web view restores the active plan snapshot when a saved session is reopened. These observer projections contain task descriptions, statuses, supersession links, and rationale only; execution evidence and tool arguments remain private.
 
 The tool executor makes at most one automatic retry, and only when a typed error marks a normal-risk read or an allowlisted built-in network lookup as retriable. It rechecks task binding and child cancellation before dispatching the retry. Filesystem, process, Git, stateful, custom network, and other mutation-capable calls are never automatically retried. After interruption or restart, unfinished side effects remain uncertain; inspect the current workspace and recorded receipts before requesting the mutation again.

openkyrozen/agent/prompts.py

@@ -54,7 +54,9 @@ def _permitted_tool_names(self, agent_config: dict[str, Any] | None = None) -> s
     active = getattr(self, "_execution_capability_token", None)
     capabilities = configured & active.capabilities if active else configured
     capabilities = mode_capabilities(capabilities, self._active_interaction_mode.get())
-    permitted = {name for name in self.AVAILABLE_TOOLS if name != "update_plan" and self.tool_registry.get_spec(name).visibility == 'public' and tool_capability(name,self.AVAILABLE_TOOLS) in capabilities}
+    permitted = {name for name in self.AVAILABLE_TOOLS if self.tool_registry.get_spec(name).visibility == 'public' and tool_capability(name,self.AVAILABLE_TOOLS) in capabilities}
+    if self._active_interaction_mode.get() != "agent":
+        permitted.discard("update_plan")
     delegated = self.execution_context.delegation_tools
     if self.execution_context.child_run_id and delegated is not None:
         permitted &= set(delegated)
@@ -88,7 +90,7 @@ def _discover_tools(self, args: str) -> str:
 
 def _agent_prompt_tools_list(self, agent_config: dict[str, Any]) -> str:
     """Build the inventory within config, active token and interaction bounds."""
-    permitted = self._permitted_tool_names(agent_config) - {"update_plan"}
+    permitted = self._permitted_tool_names(agent_config)
     detailed = permitted
     if self._prompt_profile() == "compact":
         detailed = permitted & ({"list_dir", "find_files", "search_files", "read_file", "write_file", "edit_file", "apply_patch", "run_cmd", "discover_tools"}

tests/test_agent_engine.py

@@ -354,20 +354,33 @@ def chat_response(self, messages, model=None, *, tools=None):
                 from openkyrozen.providers.models import provider_identity
                 validate_tool_history(messages, provider_id=provider_identity(self), model=model, native=True)
                 validate_replied_tool_definitions(messages, tools)
-                self.requests.append((messages, tools))
+                self.requests.append((list(messages), tools))
                 system_prompt = messages[0]["content"]
                 if any(marker in system_prompt for marker in ("Action:", "TaskList:", "TaskDone:", "DefineTool:")):
                     raise AssertionError("native prompt includes legacy control syntax")
                 if len(self.requests) == 1:
-                    self.assert_tool_schema(tools, "read_file")
+                    self.assert_tool_schema(tools, "update_plan")
                     return ModelResponse(
-                        tool_calls=(ToolCall("read-1", "read_file", {"path": "note.txt"}),),
+                        tool_calls=(ToolCall("plan-1", "update_plan", {
+                            "complexity": "complex",
+                            "tasks": [{"description": "Read note.txt", "acceptance": ["contents verified"]}],
+                        }),),
                         finish_reason=FinishReason.TOOL_REQUEST,
                         continuation=provider_continuation(self, model, "fixture", {"id": "response-1"}),
                         metadata={"provider": self.name, "model": model},
                     )
+                if len(self.requests) == 2:
+                    self.assert_tool_schema(tools, "update_plan")
+                    if not isinstance(messages[-1], ToolReply) or messages[-1].call_id != "plan-1":
+                        raise AssertionError("native plan result was not correlated to its call")
+                    return ModelResponse(
+                        tool_calls=(ToolCall("read-1", "read_file", {"path": "note.txt"}),),
+                        finish_reason=FinishReason.TOOL_REQUEST,
+                        continuation=provider_continuation(self, model, "fixture", {"id": "response-2"}),
+                        metadata={"provider": self.name, "model": model},
+                    )
                 if not isinstance(messages[-1], ToolReply) or messages[-1].call_id != "read-1":
-                    raise AssertionError("native tool result was not correlated to its call")
+                    raise AssertionError("native read result was not correlated to its call")
                 return ModelResponse(
                     text="The file contains verified text.", finish_reason=FinishReason.FINAL,
                     metadata={"provider": self.name, "model": model},
@@ -403,11 +416,14 @@ def assert_tool_schema(tools, name):
 
         self.assertIn("verified text", answer)
         retrieve_memory.assert_called_once()
-        self.assertEqual(provider.calls_attempted, 2)
-        self.assertEqual(len(provider.requests), 2)
+        self.assertEqual(provider.calls_attempted, 3)
+        self.assertEqual(len(provider.requests), 3)
         self.assertTrue(all(any(packet in item.get("content", "") for item in request[0]
                                 if isinstance(item, dict)) for request in provider.requests))
-        self.assertEqual(provider.requests[1][0][-1].call_id, "read-1")
+        self.assertEqual(provider.requests[1][0][-1].call_id, "plan-1")
+        self.assertEqual(provider.requests[2][0][-1].call_id, "read-1")
+        self.assertEqual(app.runtime.tasks.working_plan.complexity, "complex")
+        self.assertEqual(app.runtime.tasks.tasks[0]["description"], "Read note.txt")
 
 
 if __name__ == "__main__":

tests/test_working_plans.py

@@ -1023,16 +1023,28 @@ def test_runtime_call_persists_a_typed_working_plan(self):
             finally:
                 app.close()
 
-    def test_structured_update_plan_is_not_advertised_as_a_legacy_text_action(self):
+    def test_structured_update_plan_is_agent_only_and_not_a_legacy_text_action(self):
         with tempfile.TemporaryDirectory() as directory:
             path = Path(directory) / "state.sqlite3"
             memory = build_memory(path, user_id="alice", index=None)
             app = build_application(memory=memory, tools=ToolAdapters(directory), user_id="alice")
             try:
+                app.runtime._active_interaction_mode.set("agent")
                 prompt_tools = app.runtime._agent_prompt_tools_list({"capabilities": ["read", "write"]})
-                self.assertNotIn("update_plan", prompt_tools)
+                self.assertIn("update_plan", prompt_tools)
+                self.assertIn("update_plan", app.runtime._permitted_tool_names({"capabilities": ["read", "write"]}))
+                self.assertNotIn("update_plan", app.runtime._permitted_tool_names({"capabilities": ["read"]}))
+                from openkyrozen.agent.engine import AgentEngine
+                native_names = AgentEngine(app.runtime)._native_tool_names()
+                self.assertIn("update_plan", native_names)
+                native_schema = next(schema for schema in app.runtime.tool_registry.provider_schemas(names=native_names)
+                                     if schema["name"] == "update_plan")
+                self.assertIn("operations", native_schema["parameters"]["properties"])
+                app.runtime._active_interaction_mode.set("ask")
                 self.assertNotIn("update_plan", app.runtime._permitted_tool_names({"capabilities": ["read", "write"]}))
-                self.assertTrue(app.runtime._discover_tools("update_plan").startswith("Error:"))
+                app.runtime._active_interaction_mode.set("agent")
+                self.assertTrue(app.runtime._discover_tools("update_plan").startswith("- update_plan:"))
+                self.assertNotIn("update_plan", app.runtime._build_tools_list(frozenset({"read", "write"})))
                 self.assertIn("update_plan", app.runtime.tool_registry.specs)
             finally:
                 app.close()