fix: reconcile numbered plans

docs/self-evolution.md

@@ -9,7 +9,7 @@ Historical verification snapshot: `51be33361422e55e1f2f00c33a0e0f8c56132a91`
 (the post-#54 `main` revision, captured before this #55 documentation-only
 update). Snapshot date: 2026-09-04.
 
-Current repository test count at this snapshot: **190 unittest cases**.
+Current repository test count at this snapshot: **192 unittest cases**.
 
 ## Verified surface
 

main.py

@@ -4378,16 +4378,15 @@ def _tasks_from_plan(text: str) -> None:
     if not plan_match:
         return
     plan_body = plan_match.group(1).strip()
-    for index, line in enumerate(plan_body.splitlines()):
+    descriptions = []
+    for line in plan_body.splitlines():
         line = line.strip()
         if not line:
             continue
         # Remove leading numbering "1." or "1)" etc.
         cleaned = re.sub(r"^\s*\d+[.)]?\s*", "", line).strip()
-        if cleaned:
-            tasks.add_task(cleaned, task_id=f"plan-{index + 1}")
-        else:
-            tasks.add_task(line, task_id=f"plan-{index + 1}")
+        descriptions.append(cleaned or line)
+    tasks.add_ordered_plan(descriptions, task_id_prefix="plan")
 
 
 def _build_task_progress_hint() -> str:

task_engine.py

@@ -88,6 +88,12 @@ def _ordered_plan_info(task: dict[str, Any]) -> dict[str, Any] | None:
     return {"id": marker["id"], "index": marker["index"], "total": marker["total"]}
 
 
+def _ordered_plan_id(items: list[str]) -> str | None:
+    if not items or len({_task_key(item) for item in items}) != len(items):
+        return None
+    return _task_key("\n".join(items))
+
+
 def _ordered_action_compatible(description: str, action: str) -> bool:
     """Keep ordered inference bounded to descriptions with an execution hint."""
     action = _receipt_action(action)
@@ -192,6 +198,19 @@ def add_task(self, description: str, *, dependencies: list[str] | None = None,
         self._event(task, "task.created", {"description": description})
         return len(self.tasks) - 1
 
+    def add_ordered_plan(self, descriptions: list[str], *, task_id_prefix: str | None = None) -> None:
+        """Add a numbered plan with the same durable ordering as a plain TaskList."""
+        items = [str(description).strip() for description in descriptions if str(description).strip()]
+        plan_id = _ordered_plan_id(items)
+        for position, description in enumerate(items):
+            checkpoint = None
+            if plan_id is not None:
+                checkpoint = {"ordered_plan": {
+                    "id": plan_id, "index": position, "total": len(items),
+                }}
+            task_id = f"{task_id_prefix}-{position + 1}" if task_id_prefix else None
+            self.add_task(description, task_id=task_id, checkpoint=checkpoint)
+
     def set_status(self, idx: int, status: str, *, evidence: dict[str, Any] | None = None) -> bool:
         if not 0 <= idx < len(self.tasks):
             return False
@@ -493,8 +512,8 @@ def from_llm_block(self, text: str) -> None:
         existing_by_key = {_task_key(item["description"]): item for item in self.tasks}
         plain_items = [item for item in raw_tasks if isinstance(item, str) and item.strip()]
         ordered_plan_id = None
-        if len(plain_items) == len(raw_tasks) and len({_task_key(item) for item in plain_items}) == len(plain_items):
-            ordered_plan_id = _task_key("\n".join(item.strip() for item in plain_items))
+        if len(plain_items) == len(raw_tasks):
+            ordered_plan_id = _ordered_plan_id([item.strip() for item in plain_items])
         for position, item in enumerate(raw_tasks):
             checkpoint = None
             if isinstance(item, str):
@@ -565,11 +584,14 @@ def recover(self) -> list[dict[str, Any]]:
                 workspace_id=self.workspace_id, session_id=self.session_id,
                 user_id=self.user_id,
             )
-            self.tasks = [
-                task for task in all_rows
-                if (marker := _ordered_plan_info(task)) is not None
-                and marker["id"] in ordered_plan_ids
-            ]
+            self.tasks = sorted(
+                (
+                    task for task in all_rows
+                    if (marker := _ordered_plan_info(task)) is not None
+                    and marker["id"] in ordered_plan_ids
+                ),
+                key=lambda task: _ordered_plan_info(task)["index"],
+            )
         else:
             self.tasks = unfinished
         return unfinished

tests/test_task_consistency.py

@@ -469,6 +469,103 @@ def artifact_context(self, _run):
             ).stdout)
             self.assertIn("plain-string plan", reply)
 
+    def test_plan_only_tasks_reconcile_in_order_after_restart(self):
+        with tempfile.TemporaryDirectory() as directory:
+            store = MemoryBank(Path(directory) / "state.sqlite3").store
+            manager = TaskManager(store, workspace_id="project", session_id="plan-only")
+            original_tasks = main.tasks
+            main.tasks = manager
+            try:
+                main._tasks_from_plan(
+                    "Plan:\n1. Inspect the repository\n2. Read the README\n"
+                    "3. Run the tests\n4. Write the README\n5. Review the diff\n"
+                    "6. Commit the change\n7. Report the final results\n"
+                )
+            finally:
+                main.tasks = original_tasks
+
+            self.assertEqual(len(manager.tasks), 7)
+            self.assertTrue(all((task.get("checkpoint") or {}).get("ordered_plan")
+                                for task in manager.tasks))
+            receipts = [
+                ("list_dir", "."), ("read_file", "README.md"), ("run_cmd", "python -m unittest"),
+                ("write_file", "README.md|quick start"), ("git_diff", "."),
+                ("git_commit", "audit: plan-only"),
+            ]
+            for action, args in receipts[:3]:
+                evidence = manager.record_evidence(action=action, args=args, result="verified", success=True)
+                self.assertIn("task_id", evidence)
+
+            reopened = TaskManager(store, workspace_id="project", session_id="plan-only")
+            self.assertEqual(len(reopened.recover()), 4)
+            for action, args in receipts[3:]:
+                evidence = reopened.record_evidence(action=action, args=args, result="verified", success=True)
+                self.assertIn("task_id", evidence)
+            self.assertEqual([task["status"] for task in reopened.tasks[:6]], ["succeeded"] * 6)
+            self.assertEqual(reopened.tasks[6]["status"], "pending")
+            self.assertTrue(all(len(task["evidence"]) == 1 for task in reopened.tasks[:6]))
+
+    def test_cli_plan_only_completes_verified_work_without_taskdone(self):
+        class StubLearning:
+            def feedback_signal(self, _text):
+                return None
+
+            def route_profile(self, _text, _profile=None):
+                return "coder"
+
+            def begin_run(self, profile, _task, provider_model=None):
+                return {"run_id": "plan-cli-run", "profile": profile, "provider_model": provider_model}
+
+            def artifact_context(self, _run):
+                return "", []
+
+        responses = [
+            "Plan:\n1. Inspect the repository status\n2. Create the audit file\n"
+            "3. Run the test suite\n4. Stage the audit file\n5. Commit the audit file\n"
+            'Action: {"action":"git_status","args":"."}',
+            'Action: {"action":"write_file","args":"audit-plan.txt|PLAN_OK"}',
+            'Action: {"action":"run_cmd","args":"python -c pass"}',
+            'Action: {"action":"git_add","args":"audit-plan.txt"}',
+            'Action: {"action":"git_commit","args":"audit: plan-only"}',
+            "Committed the Plan-only work.",
+        ]
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            subprocess.run(["git", "init", "-q"], cwd=root, check=True)
+            subprocess.run(["git", "config", "user.email", "test@example.invalid"], cwd=root, check=True)
+            subprocess.run(["git", "config", "user.name", "OpenKyrozen Test"], cwd=root, check=True)
+            store = MemoryBank(root / "state.sqlite3").store
+            original_root = main._get_workspace_root()
+            original_tasks = main.tasks
+            original_learning = main.learning_engine
+            main._set_workspace_root(root)
+            main.tasks = TaskManager(store, workspace_id="project", session_id="plan-cli")
+            main.learning_engine = StubLearning()
+            try:
+                with patch.object(main, "_classify_complexity", return_value="complex"), \
+                     patch.object(main, "_build_messages", return_value=[]), \
+                     patch.object(main, "_build_memory_context", return_value=""), \
+                     patch.object(main, "_call_llm_with_spinner", side_effect=responses), \
+                     patch.object(main, "_get_llm_response", return_value=""), \
+                     patch.object(main, "_update_tasks_panel"), \
+                     patch.object(main, "_finish_learning_run", side_effect=lambda run, receipts, task,
+                                  result, records, tokens, started: result):
+                    reply = main._chat_turn("Create and commit the audit file.", clear_tasks=True)
+            finally:
+                main._set_workspace_root(original_root)
+                main.tasks = original_tasks
+                main.learning_engine = original_learning
+
+            rows = store.list_tasks(workspace_id="project", session_id="plan-cli")
+            self.assertEqual(len(rows), 5)
+            self.assertEqual({row["status"] for row in rows}, {"succeeded"})
+            self.assertEqual((root / "audit-plan.txt").read_text(encoding="utf-8"), "PLAN_OK")
+            self.assertIn("audit: plan-only", subprocess.run(
+                ["git", "log", "-1", "--pretty=%s"], cwd=root,
+                capture_output=True, text=True, check=True,
+            ).stdout)
+            self.assertIn("Plan-only work", reply)
+
     def test_web_natural_plan_receipts_match_and_survive_restart(self):
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)