From 3ec52759d96b0dc28d4326a130f710d4cac82bb9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=87a=C4=9Fda=C5=9F=20Y=C3=BCrekli?= <25122236+cagdasyurekli@users.noreply.github.com> Date: Sun, 6 Sep 2026 02:37:16 +0200 Subject: [PATCH] feat(automation): run timeout, retry/backoff policy, and force-stop action (#621) --- coworker/automation/models.py | 22 ++- coworker/automation/scheduler.py | 110 ++++++++++++- coworker/automation/store.py | 9 +- coworker/automation/tools.py | 41 ++++- coworker/server/app.py | 4 + coworker/server/manager.py | 74 ++++++++- surfaces/gui/src/api.ts | 8 + surfaces/gui/src/components/ScheduledView.tsx | 31 +++- tests/test_automation.py | 155 ++++++++++++++++++ 9 files changed, 433 insertions(+), 21 deletions(-) diff --git a/coworker/automation/models.py b/coworker/automation/models.py index 186e158fd7..47851cf162 100644 --- a/coworker/automation/models.py +++ b/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) diff --git a/coworker/automation/scheduler.py b/coworker/automation/scheduler.py index c2a5487ad6..c08a9b8bfb 100644 --- a/coworker/automation/scheduler.py +++ b/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 diff --git a/coworker/automation/store.py b/coworker/automation/store.py index 3c0a558a33..0f89c58191 100644 --- a/coworker/automation/store.py +++ b/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 (?, ?, ?, ?)", diff --git a/coworker/automation/tools.py b/coworker/automation/tools.py index caf192ab61..bc0ab49861 100644 --- a/coworker/automation/tools.py +++ b/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()} diff --git a/coworker/server/app.py b/coworker/server/app.py index c2e60e1592..28e2d75fe8 100644 --- a/coworker/server/app.py +++ b/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): diff --git a/coworker/server/manager.py b/coworker/server/manager.py index cf9b740faa..0150a22b89 100644 --- a/coworker/server/manager.py +++ b/coworker/server/manager.py @@ -290,7 +290,10 @@ def __init__( # The scheduler also resumes self-wake'd sessions each tick (extra_tick). self.task_store = TaskStore(base / "automation.db") self.scheduler = Scheduler( - self.task_store, self._run_scheduled_task, extra_tick=self._scheduler_tick + self.task_store, + self._run_scheduled_task, + extra_tick=self._scheduler_tick, + on_timeout=self._on_task_timeout, ) # Agent teams: two append-only stores, one record discipline. The journal is # case-keyed (knowledge outlives boards/teams); the board log is space-scoped, @@ -4991,6 +4994,33 @@ async def _notify_task_done(self, task, run: TaskRun) -> None: except Exception: pass + async def _on_task_timeout(self, task: ScheduledTask, run: TaskRun) -> None: + """Called by Scheduler when a task run times out (Issue #621).""" + if self.inbox: + try: + self.inbox.add_notification( + run.session_id or task.task_session_id, + f"⏰ Automation '{task.title}' timed out", + body=( + f"Task run timed out after {task.timeout_seconds or 900}s. " + "The overlap guard has been released." + ), + ) + except Exception: + logger.exception("Failed to add timeout notification to inbox") + await self.broadcast_event( + { + "type": "automation_run_timeout", + "data": { + "task_id": task.id, + "task_title": task.title, + "session_id": run.session_id, + "run_id": run.run_id, + "timeout_seconds": task.timeout_seconds, + }, + } + ) + # -- automation REST -------------------------------------------------------- def list_automations(self) -> dict[str, Any]: # Unseen = runs started after the task's seen mark (UX-023 sidebar badges). @@ -5004,7 +5034,7 @@ def list_automations(self) -> dict[str, Any]: { **t.public(), "unseen_runs": len(unseen), - "unseen_failed": bool(unseen) and unseen[0].status == "error", + "unseen_failed": bool(unseen) and unseen[0].status in ("error", "timed_out"), } ) return {"tasks": tasks} @@ -5069,6 +5099,21 @@ def create_automation(self, payload: dict[str, Any]) -> dict[str, Any]: # rendered the grants, the submit IS the consent. Same validation as the # agent tool — only target-bound write grants survive. always_allowed_tools=grant_entries(payload.get("permissions")), + timeout_seconds=( + float(payload["timeout_seconds"]) + if payload.get("timeout_seconds") is not None + else 900.0 + ), + max_retries=( + int(payload["max_retries"]) + if payload.get("max_retries") is not None + else 0 + ), + retry_backoff_seconds=( + float(payload["retry_backoff_seconds"]) + if payload.get("retry_backoff_seconds") is not None + else 60.0 + ), ) task.workspace = self._provision_scratch(task.task_session_id) self.task_store.save(task) @@ -5086,6 +5131,12 @@ def update_automation( task.instructions = changes["instructions"] if changes.get("title") is not None: task.title = changes["title"] + if changes.get("timeout_seconds") is not None: + task.timeout_seconds = float(changes["timeout_seconds"]) + if changes.get("max_retries") is not None: + task.max_retries = int(changes["max_retries"]) + if changes.get("retry_backoff_seconds") is not None: + task.retry_backoff_seconds = float(changes["retry_backoff_seconds"]) if changes.get("cron") is not None: from croniter import croniter @@ -5105,6 +5156,25 @@ def update_automation( engine.permissions.task_rules = task.standing_rules() return {"ok": True, "task": task.public()} + def force_stop_automation(self, task_id: str) -> dict[str, Any]: + """Force-stop an in-flight automation run (Issue #621).""" + task = self.task_store.get(task_id) + if task is None: + return {"ok": False, "error": "not found"} + stopped = self.scheduler.force_stop(task_id) + for sid, engine in list(self._engines.items()): + owner = self.task_store.task_for_run_session(sid) + if owner is not None and owner.id == task_id: + if hasattr(engine, "stop"): + engine.stop() + for r in self.task_store.runs(task_id, limit=5): + if r.status == "running": + r.status = "cancelled" + r.error = "Force stopped by user" + r.finished_at = _epoch() + self.task_store.add_run(r) + return {"ok": True, "task_id": task_id, "stopped": stopped} + def delete_automation(self, task_id: str) -> dict[str, Any]: return {"ok": self.task_store.delete(task_id), "id": task_id} diff --git a/surfaces/gui/src/api.ts b/surfaces/gui/src/api.ts index a69e0ceca8..f7c3bec617 100644 --- a/surfaces/gui/src/api.ts +++ b/surfaces/gui/src/api.ts @@ -2111,6 +2111,14 @@ export async function finalizeAutomationRun(id: string, runId: string) { return res.json(); } +/** Force stop a running automation task (Issue #621). */ +export async function stopAutomation(id: string): Promise<{ ok: boolean; stopped?: boolean }> { + const res = await fetch(`${httpBase()}/v1/automations/${encodeURIComponent(id)}/stop`, { + method: "POST", + }); + return res.json(); +} + export async function allowUser( name: string, userId: string, diff --git a/surfaces/gui/src/components/ScheduledView.tsx b/surfaces/gui/src/components/ScheduledView.tsx index bab346fa4c..5733911196 100644 --- a/surfaces/gui/src/components/ScheduledView.tsx +++ b/surfaces/gui/src/components/ScheduledView.tsx @@ -7,6 +7,7 @@ import { getAutomations, markAutomationSeen, announceAutomationsChanged, + stopAutomation, updateAutomation, type Automation, type AutomationRun, @@ -367,6 +368,12 @@ function TaskDetail({ onBack(); }; + const isRunning = runs.some((r) => r.status === "running") || task.last_status === "running"; + const forceStop = async () => { + await stopAutomation(id); + refresh(); + }; + return ( + {isRunning ? ( + + ) : ( + + )} + )} {r.artifacts.length > 0 && · {tt("automations.file_count", { count: r.artifacts.length })}} diff --git a/tests/test_automation.py b/tests/test_automation.py index 4503f22caf..532b80fc20 100644 --- a/tests/test_automation.py +++ b/tests/test_automation.py @@ -476,3 +476,158 @@ async def dead(message): assert event["data"]["session_id"] == run.session_id assert event["data"]["trigger"] == "schedule" assert dead not in manager._event_clients # dropped, not fatal + + +# -- Issue #621: Timeout, retry/backoff, force_stop ---------------------------- +@pytest.mark.asyncio +async def test_scheduler_run_timeout_and_releases_overlap_guard(tmp_path): + store = TaskStore(tmp_path / "auto.db") + t = _task(timeout_seconds=0.05) + store.save(t) + + timeout_called = asyncio.Event() + timed_out_task_id = None + + async def on_timeout(task, run): + nonlocal timed_out_task_id + timed_out_task_id = task.id + timeout_called.set() + + async def slow_runner(task, trigger): + await asyncio.sleep(1.0) + return TaskRun(task_id=task.id, status="ok") + + sched = Scheduler(store, slow_runner, on_timeout=on_timeout) + run = await sched.run_task(t, trigger="schedule") + + assert run is not None + assert run.status == "timed_out" + assert "timed out after 0.05s" in (run.error or "") + # Overlap guard must be released + assert t.id not in sched._running_ids + assert timed_out_task_id == t.id + + # Can immediately run again because overlap guard was cleared + async def fast_runner(task, trigger): + return TaskRun(task_id=task.id, status="ok") + + sched.runner = fast_runner + run2 = await sched.run_task(t, trigger="manual") + assert run2 is not None + assert run2.status == "ok" + + +@pytest.mark.asyncio +async def test_scheduler_error_retry_exponential_backoff(tmp_path): + store = TaskStore(tmp_path / "auto.db") + t = _task( + max_retries=2, + retry_backoff_seconds=10.0, + schedule=Schedule(kind="cron", cron="0 9 * * *"), + ) + store.save(t) + + attempts = 0 + + async def failing_runner(task, trigger): + nonlocal attempts + attempts += 1 + return TaskRun(task_id=task.id, status="error", error="service unavailable") + + sched = Scheduler(store, failing_runner) + + # Attempt 1: First failure schedules retry 1 with 10s backoff (10 * 2^0 = 10) + before_1 = time.time() + run1 = await sched.run_task(t, trigger="schedule") + assert run1.status == "error" + fresh1 = store.get(t.id) + assert fresh1.retry_count == 1 + assert fresh1.next_run is not None + assert fresh1.next_run >= before_1 + 9.9 + assert fresh1.next_run <= before_1 + 12.0 + + # Attempt 2: Second failure schedules retry 2 with 20s backoff (10 * 2^1 = 20) + before_2 = time.time() + run2 = await sched.run_task(fresh1, trigger="retry") + assert run2.status == "error" + fresh2 = store.get(t.id) + assert fresh2.retry_count == 2 + assert fresh2.next_run >= before_2 + 19.9 + assert fresh2.next_run <= before_2 + 23.0 + + # Attempt 3: Third failure exhausts retries -> retry_count resets to 0 and cron resumes + run3 = await sched.run_task(fresh2, trigger="retry") + assert run3.status == "error" + fresh3 = store.get(t.id) + assert fresh3.retry_count == 0 + # Next run is computed normally from cron (tomorrow at 9am), not backoff + assert fresh3.next_run is not None + + +@pytest.mark.asyncio +async def test_scheduler_force_stop(tmp_path): + store = TaskStore(tmp_path / "auto.db") + t = _task(timeout_seconds=10.0) + store.save(t) + + hang_event = asyncio.Event() + + async def hanging_runner(task, trigger): + await hang_event.wait() + return TaskRun(task_id=task.id, status="ok") + + sched = Scheduler(store, hanging_runner) + run_task = asyncio.create_task(sched.run_task(t, trigger="schedule")) + + # Give task a moment to start and claim guard + await asyncio.sleep(0.05) + assert t.id in sched._running_ids + + # User calls force_stop + stopped = sched.force_stop(t.id) + assert stopped is True + assert t.id not in sched._running_ids + + run = await run_task + assert run is not None + assert run.status == "cancelled" + assert "Force stopped" in (run.error or "") + + +def test_automations_rest_force_stop_and_settings(tmp_path, monkeypatch): + from fastapi.testclient import TestClient + + from coworker.server.app import create_app + from coworker.server.manager import SessionManager + + monkeypatch.setenv("COWORKER_STATE_DIR", str(tmp_path / "state")) + manager = SessionManager(data_dir=tmp_path / "data") + t = _task( + workspace=str(tmp_path / "ws"), + timeout_seconds=300.0, + max_retries=3, + retry_backoff_seconds=30.0, + ) + manager.task_store.save(t) + client = TestClient(create_app(manager)) + + # Get automation details + data = client.get(f"/v1/automations/{t.id}").json()["task"] + assert data["timeout_seconds"] == 300.0 + assert data["max_retries"] == 3 + assert data["retry_backoff_seconds"] == 30.0 + + # Patch timeout and retries + res = client.patch( + f"/v1/automations/{t.id}", + json={"timeout_seconds": 600.0, "max_retries": 5}, + ).json() + assert res["ok"] + assert res["task"]["timeout_seconds"] == 600.0 + assert res["task"]["max_retries"] == 5 + + # Force stop endpoint + stop_res = client.post(f"/v1/automations/{t.id}/stop").json() + assert stop_res["ok"] is True + assert stop_res["task_id"] == t.id +