feat(automation): run timeout, retry/backoff policy, and force-stop action

coworker/automation/models.py

@@ -7,7 +7,7 @@
 
 import time
 import uuid
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, fields
 from typing import Any, Optional
 
 # Indexed by cron day-of-week: 0 and 7 are Sunday, 1 is Monday … 6 is Saturday. Must start
@@ -138,6 +138,10 @@ class ScheduledTask:
     last_status: Optional[str] = None
     run_count: int = 0
     max_runs: Optional[int] = None
+    timeout_seconds: Optional[float] = 900.0  # 15 minutes default run timeout
+    max_retries: int = 0  # max retries on error (0 = disabled)
+    retry_backoff_seconds: float = 60.0  # base backoff duration for retries
+    retry_count: int = 0  # consecutive error retry attempt count
     # Sidebar unread tracking (UX-023): runs started after this mark count as
     # "unseen"; opening the automation's detail advances it. 0.0 = never opened.
     seen_runs_at: float = 0.0
@@ -155,7 +159,9 @@ def to_dict(self) -> dict:
     def from_dict(cls, d: dict) -> "ScheduledTask":
         d = dict(d)
         d["schedule"] = Schedule.from_dict(d.get("schedule") or {})
-        return cls(**d)
+        valid_fields = {f.name for f in fields(cls)}
+        filtered = {k: v for k, v in d.items() if k in valid_fields}
+        return cls(**filtered)
 
     # -- standing rules (§25) --------------------------------------------------
     def standing_rules(self) -> dict[str, set[str]]:
@@ -204,6 +210,10 @@ def public(self) -> dict[str, Any]:
             "last_run": self.last_run,
             "last_status": self.last_status,
             "run_count": self.run_count,
+            "timeout_seconds": self.timeout_seconds,
+            "max_retries": self.max_retries,
+            "retry_backoff_seconds": self.retry_backoff_seconds,
+            "retry_count": self.retry_count,
             "notify_on_completion": self.notify_on_completion,
             # UX-023: lets the detail freeze the pre-open mark for its "new" pills.
             "seen_runs_at": self.seen_runs_at,
@@ -223,11 +233,11 @@ class TaskRun:
     run_id: str = field(default_factory=lambda: "run-" + uuid.uuid4().hex[:10])
     started_at: float = field(default_factory=_now)
     finished_at: Optional[float] = None
-    status: str = "running"  # running | ok | error | skipped
+    status: str = "running"  # running | ok | error | skipped | timed_out | cancelled
     result_text: Optional[str] = None
     artifacts: list[str] = field(default_factory=list)
     error: Optional[str] = None
-    trigger: str = "schedule"  # schedule | manual | catchup
+    trigger: str = "schedule"  # schedule | manual | catchup | retry
     session_id: str = ""  # the run's own conversation thread — persisted + continuable
 
     def __post_init__(self) -> None:
@@ -239,4 +249,6 @@ def to_dict(self) -> dict:
 
     @classmethod
     def from_dict(cls, d: dict) -> "TaskRun":
-        return cls(**d)
+        valid_fields = {f.name for f in fields(cls)}
+        filtered = {k: v for k, v in d.items() if k in valid_fields}
+        return cls(**filtered)

coworker/automation/scheduler.py

@@ -4,12 +4,18 @@
 startup, then resume), and **skip-on-overlap** (don't stack a run if the previous is still
 going). The actual execution is injected as `runner(task, trigger) -> TaskRun` so this stays
 independent of the engine/manager.
+
+Features (Issue #621):
+- Run timeout (per-task or default e.g. 15 min), releasing overlap guard on expiry
+- Error retry with exponential backoff for runs ending in error
+- Force stop action to cancel stuck in-flight runs from UI
 """
 
 from __future__ import annotations
 
 import asyncio
 import logging
+import time
 from typing import Awaitable, Callable, Optional
 
 from .models import ScheduledTask, TaskRun
@@ -28,15 +34,20 @@ def __init__(
         *,
         tick_seconds: float = 30.0,
         extra_tick: Optional[Callable[[], Awaitable[None]]] = None,
+        default_timeout: float = 900.0,
+        on_timeout: Optional[Callable[[ScheduledTask, TaskRun], Awaitable[None]]] = None,
     ) -> None:
         self.store = store
         self.runner = runner
         self.tick_seconds = tick_seconds
         # An extra per-tick coroutine (self-wake resumption: resume sessions whose wakes are due).
         self.extra_tick = extra_tick
+        self.default_timeout = default_timeout
+        self.on_timeout = on_timeout
         self._task: Optional[asyncio.Task] = None
         self._running_ids: set[str] = set()  # overlap guard
         self._spawned: set[asyncio.Task] = set()  # keep spawned runs referenced
+        self._active_runs: dict[str, asyncio.Task] = {}  # task_id -> running asyncio.Task
 
     def start(self) -> None:
         if self._task is None:
@@ -59,6 +70,7 @@ async def stop(self) -> None:
             except asyncio.CancelledError:
                 pass
         self._spawned.clear()
+        self._active_runs.clear()
 
     async def _loop(self) -> None:
         # First pass = run-once-catch-up for anything missed while the server was down.
@@ -84,9 +96,14 @@ async def _tick(self, *, trigger: str) -> None:
             # already clear — the task runs twice.
             if not self._claim(task.id):
                 continue
-            spawned = asyncio.create_task(self._run_claimed(task, trigger=trigger))
+            run_trigger = "retry" if task.retry_count > 0 else trigger
+            spawned = asyncio.create_task(self._run_claimed(task, trigger=run_trigger))
             self._spawned.add(spawned)
+            self._active_runs[task.id] = spawned
             spawned.add_done_callback(self._spawned.discard)
+            spawned.add_done_callback(
+                lambda _, tid=task.id: self._active_runs.pop(tid, None)
+            )
         if self.extra_tick is not None:
             try:
                 await self.extra_tick()
@@ -100,29 +117,110 @@ def _claim(self, task_id: str) -> bool:
         self._running_ids.add(task_id)
         return True
 
+    def force_stop(self, task_id: str) -> bool:
+        """Cancel an in-flight run for task_id, immediately releasing the overlap guard."""
+        self._running_ids.discard(task_id)
+        active = self._active_runs.pop(task_id, None)
+        if active is not None and not active.done():
+            active.cancel()
+            return True
+        return False
+
     async def run_task(self, task: ScheduledTask, *, trigger: str) -> Optional[TaskRun]:
         if not self._claim(task.id):
             return None
-        return await self._run_claimed(task, trigger=trigger)
+        current = asyncio.current_task()
+        if current is not None:
+            self._active_runs[task.id] = current
+        try:
+            return await self._run_claimed(task, trigger=trigger)
+        finally:
+            self._active_runs.pop(task.id, None)
 
     async def _run_claimed(
         self, task: ScheduledTask, *, trigger: str
     ) -> Optional[TaskRun]:
+        timeout = (
+            task.timeout_seconds
+            if task.timeout_seconds is not None
+            else self.default_timeout
+        )
+        run = None
         try:
-            run = await self.runner(task, trigger)
+            if timeout and timeout > 0:
+                run = await asyncio.wait_for(
+                    self.runner(task, trigger), timeout=timeout
+                )
+            else:
+                run = await self.runner(task, trigger)
+        except asyncio.TimeoutError:
+            logger.warning("task %s run timed out after %ss", task.id, timeout)
+            run = TaskRun(
+                task_id=task.id,
+                status="timed_out",
+                error=f"Task run timed out after {timeout}s",
+                trigger=trigger,
+                finished_at=time.time(),
+            )
+            self.store.add_run(run)
+            if self.on_timeout is not None:
+                try:
+                    await self.on_timeout(task, run)
+                except Exception:
+                    logger.exception(
+                        "scheduler on_timeout callback failed for %s", task.id
+                    )
+        except asyncio.CancelledError:
+            logger.info("task %s was cancelled / force stopped", task.id)
+            run = TaskRun(
+                task_id=task.id,
+                status="cancelled",
+                error="Force stopped by user",
+                trigger=trigger,
+                finished_at=time.time(),
+            )
+            self.store.add_run(run)
         except Exception as exc:
             logger.exception("task %s run failed", task.id)
             run = TaskRun(
-                task_id=task.id, status="error", error=str(exc), trigger=trigger
+                task_id=task.id,
+                status="error",
+                error=str(exc),
+                trigger=trigger,
+                finished_at=time.time(),
             )
             self.store.add_run(run)
         finally:
             self._running_ids.discard(task.id)
-        # advance the task (run_count/last_run) → save recomputes next_run.
+            self._active_runs.pop(task.id, None)
+
+        # advance the task (run_count/last_run/status/retry).
         fresh = self.store.get(task.id)
         if fresh is not None:
             fresh.run_count += 1
             fresh.last_run = run.started_at if run else None
             fresh.last_status = run.status if run else "error"
-            self.store.save(fresh)
+
+            # Retry on error with exponential backoff (not cancelled or timed-out)
+            if (
+                run
+                and run.status == "error"
+                and fresh.max_retries > 0
+                and fresh.retry_count < fresh.max_retries
+            ):
+                backoff = fresh.retry_backoff_seconds * (2**fresh.retry_count)
+                fresh.retry_count += 1
+                fresh.next_run = time.time() + backoff
+                logger.info(
+                    "task %s failed (%s); scheduled retry %d/%d in %.1fs",
+                    fresh.id,
+                    run.error,
+                    fresh.retry_count,
+                    fresh.max_retries,
+                    backoff,
+                )
+                self.store.save(fresh, recompute_next_run=False)
+            else:
+                fresh.retry_count = 0
+                self.store.save(fresh, recompute_next_run=True)
         return run

coworker/automation/store.py

@@ -100,9 +100,14 @@ def _init(self) -> None:
             self._conn.commit()
 
     # -- tasks ------------------------------------------------------------------
-    def save(self, task: ScheduledTask) -> ScheduledTask:
+    def save(
+        self, task: ScheduledTask, *, recompute_next_run: bool = True
+    ) -> ScheduledTask:
         task.updated_at = _epoch_now()
-        task.next_run = compute_next_run(task) if task.enabled else None
+        if recompute_next_run:
+            task.next_run = compute_next_run(task) if task.enabled else None
+        else:
+            task.next_run = task.next_run if task.enabled else None
         with self._lock:
             self._conn.execute(
                 "INSERT OR REPLACE INTO scheduled_tasks (id, enabled, next_run, data) VALUES (?, ?, ?, ?)",

coworker/automation/tools.py

@@ -9,7 +9,7 @@
 
 from __future__ import annotations
 
-from typing import Any, Callable, Optional
+from typing import Any, Callable
 
 import aisuite as ai
 
@@ -86,6 +86,14 @@
                         "required": ["tool", "target", "access"],
                     },
                 },
+                "timeout_seconds": {
+                    "type": "number",
+                    "description": "Optional run timeout in seconds (default 900s / 15 min).",
+                },
+                "max_retries": {
+                    "type": "integer",
+                    "description": "Optional maximum error retries with exponential backoff (default 0).",
+                },
             },
             "required": ["title", "instructions"],
         },
@@ -105,6 +113,9 @@
                 "instructions": {"type": "string"},
                 "cron": {"type": "string"},
                 "title": {"type": "string"},
+                "timeout_seconds": {"type": "number"},
+                "max_retries": {"type": "integer"},
+                "retry_backoff_seconds": {"type": "number"},
             },
             "required": ["id"],
         },
@@ -155,7 +166,14 @@ def scheduling_tools(
     default_workspace: str,
 ) -> list[Callable[..., Any]]:
     def create_scheduled_task(
-        title, instructions, cron=None, fire_at=None, timezone="local", permissions=None
+        title,
+        instructions,
+        cron=None,
+        fire_at=None,
+        timezone="local",
+        permissions=None,
+        timeout_seconds=900.0,
+        max_retries=0,
     ):
         from croniter import croniter
 
@@ -185,6 +203,8 @@ def create_scheduled_task(
             origin_session_id=origin.get("session_id", ""),
             agent=origin.get("agent", "cowork"),
             always_allowed_tools=grants,
+            timeout_seconds=float(timeout_seconds) if timeout_seconds is not None else 900.0,
+            max_retries=int(max_retries) if max_retries is not None else 0,
         )
         store.save(task)
         return {
@@ -195,13 +215,22 @@ def create_scheduled_task(
             "next_run": task.next_run,
             "workspace": workspace,
             "always_allowed": grants,
+            "timeout_seconds": task.timeout_seconds,
+            "max_retries": task.max_retries,
         }
 
     def list_scheduled_tasks():
         return {"tasks": [t.public() for t in store.list()]}
 
     def update_scheduled_task(
-        id, enabled=None, instructions=None, cron=None, title=None
+        id,
+        enabled=None,
+        instructions=None,
+        cron=None,
+        title=None,
+        timeout_seconds=None,
+        max_retries=None,
+        retry_backoff_seconds=None,
     ):
         from croniter import croniter
 
@@ -219,6 +248,12 @@ def update_scheduled_task(
             task.instructions = instructions
         if title is not None:
             task.title = title
+        if timeout_seconds is not None:
+            task.timeout_seconds = float(timeout_seconds)
+        if max_retries is not None:
+            task.max_retries = int(max_retries)
+        if retry_backoff_seconds is not None:
+            task.retry_backoff_seconds = float(retry_backoff_seconds)
         store.save(task)
         return {"ok": True, "task": task.public()}
 

coworker/server/app.py

@@ -2044,6 +2044,10 @@ def automation_run(task_id: str) -> dict[str, Any]:
     def automation_run_finalize(task_id: str, run_id: str) -> dict[str, Any]:
         return manager.finalize_manual_run(task_id, run_id)
 
+    @app.post("/v1/automations/{task_id}/stop")
+    def automation_stop(task_id: str) -> dict[str, Any]:
+        return manager.force_stop_automation(task_id)
+
     @app.websocket("/ws/session/{session_id}")
     async def ws_session(ws: WebSocket, session_id: str) -> None:
         if not _websocket_authenticated(ws):