fix: recover and execute durable tasks

Author: EvanProgramming

README.md

@@ -381,7 +381,15 @@ The migration creates a `.v1-backup` copy and writes the v2 database under `~/.k
 
 ### v2 durable tasks and learning
 
-Tasks persist across process restarts and use `pending`, `running`, `succeeded`, `failed`, `blocked`, and `cancelled` states. `TaskDone` is only a completion request; a successful tool result, test, file check, or explicit confirmation must provide evidence before a task can succeed.
+Tasks persist across process restarts and use `pending`, `running`, `succeeded`, `failed`, `blocked`, and `cancelled` states. `TaskDone` is only a completion request; a successful tool result, test, file check, or explicit confirmation must provide evidence before a task can succeed. API-created tasks with an explicit safe `action` and string `args` are picked up by the durable worker after server restart; failed or blocked tasks require an explicit resume request.
+
+```bash
+curl -X POST http://127.0.0.1:8000/api/v2/tasks \
+  -H 'Content-Type: application/json' \
+  -d '{"description":"write a marker","action":"write_file","args":"task-marker.txt|completed","acceptance":["marker written"]}'
+# Resume a failed/blocked task:
+curl -X POST http://127.0.0.1:8000/api/v2/tasks/<task-id>/resume
+```
 
 Run a frozen clean-versus-evolved benchmark with identical case order and runner settings:
 

event_store.py

@@ -402,6 +402,8 @@ def list_tasks(self, *, workspace_id: str = "default", session_id: str | None =
         result = []
         for row in rows:
             item = dict(row)
+            if item["status"] == "done":
+                item["status"] = "succeeded"
             item["dependencies"] = self._loads(item["dependencies"], [])
             item["acceptance"] = self._loads(item["acceptance"], [])
             item["checkpoint"] = self._loads(item["checkpoint"], {})

main.py

@@ -93,7 +93,7 @@ def _terminal_supports_unicode() -> bool:
 from rich import print as rprint
 
 from memory import MemoryBank
-from task_engine import TaskManager
+from task_engine import TaskManager, TaskWorker
 from event_store import stable_hash
 from learning_engine import LearningEngine
 from skill_registry import SkillRegistry
@@ -2431,6 +2431,22 @@ def _run_tool(action: str, args: str) -> str:
     return result
 
 
+def _execute_durable_task(task: dict[str, Any]) -> dict[str, Any]:
+    """Run one explicitly stored task action through the normal CLI gates."""
+    checkpoint = task.get("checkpoint", {})
+    action = TOOL_ALIASES.get(str(checkpoint.get("action", "")).strip(),
+                              str(checkpoint.get("action", "")).strip())
+    args = checkpoint.get("args", "")
+    if not action:
+        return {"success": False,
+                "result": "No executable action stored; create the task with action and string args."}
+    if not isinstance(args, str):
+        return {"success": False, "action": action, "result": "Task args must be a plain string."}
+    result = str(_run_tool(action, args))[:2000]
+    return {"success": not _is_tool_error(result), "action": action, "args": args, "result": result,
+            "acceptance": str(checkpoint.get("acceptance") or "durable task action completed")}
+
+
 def _select_model(user_input: str) -> str:
     """Auto-select the best DeepSeek model based on task complexity.
     Simple queries → fast model (deepseek-chat / V4).
@@ -3599,6 +3615,14 @@ def _print_banner() -> None:
     console.print(f"  [{_MUTED}]Platform: {platform_tag} {_DOT} Python {sys.version_info.major}.{sys.version_info.minor}[/{_MUTED}]")
 
 
+def _run_recovered_tasks(*, max_tasks: int = 20) -> list[dict[str, Any]]:
+    """Recover pending CLI tasks and execute only the bounded safe queue."""
+    recovered = tasks.recover()
+    if recovered:
+        console.print(f"[{_MUTED}]Recovered {len(recovered)} durable task(s).[/{_MUTED}]")
+    return TaskWorker(tasks, _execute_durable_task, max_tasks=max_tasks).run_until_idle(max_tasks=max_tasks)
+
+
 def main() -> None:
     # Bytecode cache cleared already at module level (see top of file)
     global _provider_config, llm_provider, DEEPSEEK_MODEL, DEEPSEEK_MODEL_SIMPLE, DEEPSEEK_MODEL_COMPLEX, MODEL_NAME
@@ -3653,6 +3677,9 @@ def main() -> None:
         sys.exit(1)
     # Set workspace root for sandbox
     _set_workspace_root(os.getcwd())
+    task_results = _run_recovered_tasks()
+    for item in task_results:
+        console.print(f"[{_SUCCESS}]Durable task {item['task']['id']}: {item['status']}[/{_SUCCESS}]")
     # Load project files synchronously to avoid ChromaDB thread conflicts
     _load_project_files_into_memory()
     console.print(f"[{_MUTED}]Project files loaded into memory.[/{_MUTED}]")
@@ -3711,6 +3738,21 @@ def main() -> None:
             _show_self_learning_menu()
             continue
 
+        if user_input.lower().startswith("/tasks"):
+            parts = user_input.split(maxsplit=2)
+            if len(parts) == 1 or (len(parts) > 1 and parts[1].lower() == "list"):
+                console.print(tasks.format())
+            elif len(parts) == 3 and parts[1].lower() == "resume":
+                resumed = tasks.resume(parts[2].strip())
+                if not resumed:
+                    console.print(f"[{_ERROR}]Failed or blocked task not found.[/{_ERROR}]")
+                else:
+                    result = TaskWorker(tasks, _execute_durable_task, max_tasks=1).run_once()
+                    console.print(json.dumps(result or {"status": "queued"}, ensure_ascii=False, indent=2, default=str))
+            else:
+                console.print("Usage: /tasks list | /tasks resume <task-id>")
+            continue
+
         if user_input.lower().startswith("/agent"):
             parts = user_input.split(maxsplit=1)
             if len(parts) == 1:

server.py

@@ -34,9 +34,10 @@
 import main as _agent
 from providers import get_cost_summary, reset_cost_tracker
 from memory import MemoryBank
-from task_engine import TaskManager
+from task_engine import TaskManager, TaskWorker
 from scheduler import JobScheduler
 from tools import allowed_tool_names, resolve_capabilities, tool_capability
+from capability_tokens import issue_capability_token
 
 try:
     from fastapi import FastAPI, Request, HTTPException, Depends
@@ -144,6 +145,39 @@ def _audit(event: str, detail: str = "", user: str = "anonymous") -> None:
                           user_id=_SERVER_ACTOR_ID)
 
 
+def _execute_durable_task(task: dict[str, Any]) -> dict[str, Any]:
+    """Execute only an explicitly stored, Web-profile action."""
+    checkpoint = task.get("checkpoint", {})
+    action = str(checkpoint.get("action", "")).strip()
+    args = checkpoint.get("args", "")
+    if not action:
+        return {"success": False,
+                "result": "No executable action stored; create the task with action and string args."}
+    action = _agent.TOOL_ALIASES.get(action, action)
+    if not isinstance(args, str):
+        return {"success": False, "action": action,
+                "result": "Task args must be a plain string."}
+    if action not in _allowed_server_tools("web"):
+        return {"success": False, "action": action,
+                "result": f"Error: task action '{action}' is not allowed by the Web capability profile."}
+    _agent._execution_capability_token = issue_capability_token(
+        "surface:web:durable-task", _server_capabilities("web"), ttl_seconds=300,
+    )
+    result = str(_agent._run_tool(action, args))[:2000]
+    return {
+        "success": not result.lower().startswith("error:"), "action": action, "args": args,
+        "result": result,
+        "acceptance": str(checkpoint.get("acceptance") or "durable task action completed"),
+    }
+
+
+_task_worker = TaskWorker(
+    TaskManager(_agent.memory_bank.store, workspace_id=_agent.memory_bank.workspace_id,
+                user_id=_SERVER_ACTOR_ID),
+    _execute_durable_task,
+)
+
+
 def _run_scheduled_job(job: dict[str, Any]) -> None:
     payload = job.get("payload", {})
     if payload.get("type") == "learning_cycle":
@@ -152,6 +186,9 @@ def _run_scheduled_job(job: dict[str, Any]) -> None:
         if _agent._SELF_LEARNING_FLAGS.get("outcome_verified_evolution", True):
             _agent._review_evolution_runs()
         return
+    if payload.get("type") == "task_worker":
+        _task_worker.run_once()
+        return
     if payload.get("type") == "chat":
         session_id = _normalise_session_id(payload.get("session_id"))
         session = _get_or_create_session(session_id, _SERVER_ACTOR_ID)
@@ -661,13 +698,36 @@ async def api_v2_create_task(request: Request):
     session = _normalise_session_id(body.get("session_id"))
     manager = TaskManager(_agent.memory_bank.store, workspace_id=_agent.memory_bank.workspace_id,
                           session_id=session, user_id=_SERVER_ACTOR_ID)
+    checkpoint: dict[str, Any] = {}
+    if body.get("action") is not None:
+        action = _agent.TOOL_ALIASES.get(str(body.get("action")).strip(), str(body.get("action")).strip())
+        if action not in _allowed_server_tools("web"):
+            raise HTTPException(403, f"Task action '{action}' is not allowed by the Web capability profile")
+        if not isinstance(body.get("args", ""), str):
+            raise HTTPException(400, "args must be a plain string")
+        checkpoint = {"action": action, "args": body.get("args", "")}
+        if isinstance(body.get("acceptance"), list) and body["acceptance"]:
+            checkpoint["acceptance"] = body["acceptance"][0]
     index = manager.add_task(
         str(body["description"]),
         dependencies=body.get("dependencies") if isinstance(body.get("dependencies"), list) else None,
         acceptance=body.get("acceptance") if isinstance(body.get("acceptance"), list) else None,
         priority=int(body.get("priority", 0)),
+        checkpoint=checkpoint,
     )
-    return {"task": manager.tasks[index]}
+    if not any(job.get("payload", {}).get("type") == "task_worker" for job in _scheduler.list_jobs()):
+        _scheduler.schedule_every("durable-task-worker", 1.0, payload={"type": "task_worker"},
+                                  job_id="job_durable_task_worker", delay_seconds=0)
+    return {"task": manager.tasks[index], "queued": True}
+
+
+@app.post("/api/v2/tasks/{task_id}/resume", dependencies=[Depends(require_api_access)])
+async def api_v2_resume_task(task_id: str):
+    manager = _task_worker.manager
+    task = manager.resume(task_id)
+    if task is None:
+        raise HTTPException(404, "Failed or blocked task not found in this scope")
+    return {"task": task, "queued": True}
 
 
 @app.get("/api/v2/learning", dependencies=[Depends(require_api_access)])
@@ -1148,6 +1208,12 @@ async def startup():
     _agent._load_project_files_into_memory()
     _load_plugins()
     _trigger_hook("on_startup", agent=_agent)
+    recovered = _task_worker.recover()
+    if recovered:
+        _audit("TASK_RECOVERY", f"recovered={len(recovered)} pending_or_resumable tasks")
+    if not any(job.get("payload", {}).get("type") == "task_worker" for job in _scheduler.list_jobs()):
+        _scheduler.schedule_every("durable-task-worker", 1.0, payload={"type": "task_worker"},
+                                  job_id="job_durable_task_worker", delay_seconds=0)
     _scheduler.start()
     _audit("STARTUP", "server started")
     print("[Server] Ready — http://127.0.0.1:8000 (set KYROZEN_SERVER_TOKEN for remote access)")

task_engine.py

@@ -87,7 +87,7 @@ def _event(self, task: dict[str, Any], event_type: str, payload: Any) -> None:
 
     def add_task(self, description: str, *, dependencies: list[str] | None = None,
                  acceptance: list[dict[str, Any]] | None = None, priority: int = 0,
-                 task_id: str | None = None) -> int:
+                 task_id: str | None = None, checkpoint: dict[str, Any] | None = None) -> int:
         description = str(description).strip()
         if not description:
             raise ValueError("Task description cannot be empty")
@@ -104,7 +104,7 @@ def add_task(self, description: str, *, dependencies: list[str] | None = None,
         task = {
             "id": task_id, "description": description, "status": "pending", "priority": priority,
             "dependencies": dependencies or [], "acceptance": acceptance or [], "attempts": 0,
-            "checkpoint": {}, "evidence": [], "created_at": utc_now(), "updated_at": utc_now(),
+            "checkpoint": checkpoint or {}, "evidence": [], "created_at": utc_now(), "updated_at": utc_now(),
         }
         self.tasks.append(task)
         self._persist(task)
@@ -167,6 +167,78 @@ def active_task_id(self) -> str | None:
                 return task["id"]
         return None
 
+    def _replace_task(self, task: dict[str, Any]) -> int:
+        for index, existing in enumerate(self.tasks):
+            if existing["id"] == task["id"]:
+                self.tasks[index] = task
+                return index
+        self.tasks.append(task)
+        return len(self.tasks) - 1
+
+    def claim_next(self) -> dict[str, Any] | None:
+        """Atomically claim one pending task in this exact durable scope."""
+        with self.store._lock, self.store.connection() as db:
+            row = db.execute(
+                "SELECT * FROM tasks WHERE user_id=? AND workspace_id=? AND session_id IS ? AND status=? "
+                "ORDER BY priority DESC, updated_at LIMIT 1",
+                (self.user_id, self.workspace_id, self.session_id, "pending"),
+            ).fetchone()
+            if row is None:
+                return None
+            now = utc_now()
+            claimed = db.execute(
+                "UPDATE tasks SET status=?,attempts=attempts+1,updated_at=? WHERE id=? AND user_id=? "
+                "AND workspace_id=? AND session_id IS ? AND status=?",
+                ("running", now, row["id"], self.user_id, self.workspace_id, self.session_id, "pending"),
+            ).rowcount
+            if claimed != 1:
+                return None
+        task = dict(row)
+        task["status"] = "running"
+        task["attempts"] = int(task.get("attempts", 0)) + 1
+        for key in ("dependencies", "acceptance", "checkpoint", "evidence"):
+            task[key] = self.store._loads(task[key], [] if key in {"dependencies", "acceptance", "evidence"} else {})
+        task["updated_at"] = now
+        self._replace_task(task)
+        self._event(task, "task.claimed", {"attempt": task["attempts"]})
+        return task
+
+    def update_checkpoint(self, task_id: str, checkpoint: dict[str, Any]) -> bool:
+        """Persist a bounded resumable checkpoint for one task."""
+        if not isinstance(checkpoint, dict):
+            raise ValueError("checkpoint must be an object")
+        if not self.tasks:
+            self.recover()
+        index = next((i for i, task in enumerate(self.tasks) if task["id"] == str(task_id)), None)
+        if index is None:
+            return False
+        task = self.tasks[index]
+        task["checkpoint"] = {**task.get("checkpoint", {}), **checkpoint}
+        task["updated_at"] = utc_now()
+        self._persist(task)
+        self._event(task, "task.checkpoint", {"checkpoint": task["checkpoint"]})
+        return True
+
+    def resume(self, task_id: str) -> dict[str, Any] | None:
+        """Move one failed or blocked task back to pending in this scope."""
+        if not self.tasks:
+            self.recover()
+        index = next((i for i, task in enumerate(self.tasks) if task["id"] == str(task_id)), None)
+        if index is None:
+            self.recover()
+            index = next((i for i, task in enumerate(self.tasks) if task["id"] == str(task_id)), None)
+        if index is None:
+            return None
+        task = self.tasks[index]
+        if task.get("status") not in {"failed", "blocked"}:
+            return task if task.get("status") == "pending" else None
+        previous = task["status"]
+        task["status"] = "pending"
+        task["updated_at"] = utc_now()
+        self._persist(task)
+        self._event(task, "task.resumed", {"previous_status": previous})
+        return task
+
     def reconcile_completions(self, task_ids: set[str] | None = None) -> None:
         for idx in list(self._pending_completions):
             if task_ids is not None and self.tasks[idx]["id"] not in task_ids:
@@ -256,3 +328,53 @@ def recover(self) -> list[dict[str, Any]]:
                 self._event(task, "task.recovered", {"previous_status": "running"})
         self.tasks = rows
         return rows
+
+
+class TaskWorker:
+    """Execute at most a bounded number of durable task actions per call."""
+
+    def __init__(self, manager: TaskManager, executor: Any, *, max_tasks: int = 1):
+        self.manager = manager
+        self.executor = executor
+        self.max_tasks = max(1, min(int(max_tasks), 20))
+
+    def recover(self) -> list[dict[str, Any]]:
+        return self.manager.recover()
+
+    def run_once(self) -> dict[str, Any] | None:
+        task = self.manager.claim_next()
+        if task is None:
+            return None
+        task_id = task["id"]
+        try:
+            outcome = self.executor(task)
+            if not isinstance(outcome, dict):
+                outcome = {"success": False, "result": str(outcome)}
+        except Exception as exc:
+            outcome = {"success": False, "result": f"Error: {exc}"}
+        success = bool(outcome.get("success"))
+        result = str(outcome.get("result", ""))[:2000]
+        action = str(outcome.get("action") or task.get("checkpoint", {}).get("action") or "task.execute")
+        acceptance = str(outcome.get("acceptance") or "durable task action completed") if success else None
+        evidence = self.manager.record_evidence(
+            task_id=task_id, action=action, args=outcome.get("args") or task.get("checkpoint", {}).get("args"),
+            result=result, success=success, acceptance=acceptance,
+            receipt_id=outcome.get("receipt_id"),
+        )
+        index = next(i for i, item in enumerate(self.manager.tasks) if item["id"] == task_id)
+        status = "succeeded" if success and self.manager.set_status(index, "succeeded") else "failed"
+        if status == "failed" and self.manager.tasks[index].get("status") != "failed":
+            self.manager.set_status(index, "failed")
+        self.manager._event(self.manager.tasks[index], "task.execution_completed", {
+            "success": status == "succeeded", "result": result, "evidence": evidence,
+        })
+        return {"task": self.manager.tasks[index], "status": status, "result": result, "evidence": evidence}
+
+    def run_until_idle(self, *, max_tasks: int | None = None) -> list[dict[str, Any]]:
+        results = []
+        for _ in range(max(1, min(int(max_tasks or self.max_tasks), 20))):
+            result = self.run_once()
+            if result is None:
+                break
+            results.append(result)
+        return results