diff --git a/CHANGELOG b/CHANGELOG new file mode 100644 index 0000000..3c77d9b --- /dev/null +++ b/CHANGELOG @@ -0,0 +1,22 @@ +# Changelog + +All notable user-facing changes to OpenScan3 firmware are documented here. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## Unreleased + +### Added + +- Tasks can now be chained by making one task depend on another. Dependent + tasks wait for successful completion of their prerequisite. + +### Changed + +- Interrupted scan and focus-stacking tasks can be resumed after a restart. + Scan progress and completed focus-stacking batches are reused, and replacing + an old task preserves dependency chains. +- Clarified that the generic task endpoint is intended for experimental and + custom tasks. Scan, focus-stacking, and cloud-upload tasks must use their + project-specific endpoints. diff --git a/docs/TASKS.md b/docs/TASKS.md index 3696f53..3db837f 100644 --- a/docs/TASKS.md +++ b/docs/TASKS.md @@ -273,9 +273,14 @@ only once. No JSON entry or second list of task names is required. ### 4. Make the task available where it is needed -Every registered task can already be started through -`POST /tasks/{task_name}`. If another firmware feature needs to start it, add a -small function for that feature: +Experimental and custom tasks can be started through +`POST /tasks/{task_name}`. This generic endpoint only creates the task; it does +not persist references on domain objects such as scans or projects and does +not perform domain-specific validation. + +Tasks owned by a user-facing workflow must be started through that workflow's +specialized endpoint. If another firmware feature needs to start a task, add a +small service function for that feature: ```python from openscan_firmware.controllers.services.tasks.task_manager import ( diff --git a/openscan_firmware/controllers/services/cloud.py b/openscan_firmware/controllers/services/cloud.py index a3bc4b3..4cfd862 100644 --- a/openscan_firmware/controllers/services/cloud.py +++ b/openscan_firmware/controllers/services/cloud.py @@ -508,6 +508,7 @@ async def upload_project( *, project_manager: ProjectManager | None = None, token: str | None = None, + depends_on: str | None = None, ): """Schedule an upload task for an existing project. @@ -515,6 +516,7 @@ async def upload_project( project_name: Name of the project directory to upload. project_manager: Optional project manager to validate the project exists. token: Optional cloud token override forwarded to the task. + depends_on: Optional ID of a task that must complete successfully first. Returns: Task: The TaskManager model describing the scheduled upload. @@ -540,16 +542,18 @@ async def upload_project( task.task_type == "cloud_upload_task" and task.run_args and task.run_args[0] == project_name - and task.status in {TaskStatus.PENDING, TaskStatus.RUNNING} + and task.status in {TaskStatus.PENDING, TaskStatus.RUNNING, TaskStatus.INTERRUPTED} ): raise CloudServiceError( "An upload for this project is already in progress. Wait for completion or cancel it." ) + task_kwargs = {"depends_on": depends_on} if depends_on is not None else {} task = await task_manager.create_and_run_task( "cloud_upload_task", project_name, token=token, + **task_kwargs, ) return task @@ -599,7 +603,7 @@ async def download_project( task.task_type == "cloud_download_task" and task.run_args and task.run_args[0] == project_name - and task.status in {TaskStatus.PENDING, TaskStatus.RUNNING} + and task.status in {TaskStatus.PENDING, TaskStatus.RUNNING, TaskStatus.INTERRUPTED} ): raise CloudServiceError( "A download for this project is already in progress. Wait for completion or cancel it." diff --git a/openscan_firmware/controllers/services/focus_stacking.py b/openscan_firmware/controllers/services/focus_stacking.py index a26ef42..05f1343 100644 --- a/openscan_firmware/controllers/services/focus_stacking.py +++ b/openscan_firmware/controllers/services/focus_stacking.py @@ -19,12 +19,17 @@ } -async def start_focus_stacking(project_name: str, scan_index: int) -> Task: +async def start_focus_stacking( + project_name: str, + scan_index: int, + depends_on: str | None = None, +) -> Task: """Start a focus stacking task and persist the task reference on the scan. Args: project_name: Name of the project containing the scan. scan_index: Index of the scan to process. + depends_on: Optional ID of a task that must complete successfully first. Returns: The Task representing the focus stacking job. @@ -38,6 +43,7 @@ async def start_focus_stacking(project_name: str, scan_index: int) -> Task: raise ValueError(f"Scan {scan_index} not found in project '{project_name}'") existing = scan.stacking_task_status + replaced_task_id: str | None = None if existing and existing.task_id: existing_task = task_manager.get_task_info(existing.task_id) if existing_task and existing_task.status in _ACTIVE_STATUSES: @@ -50,12 +56,19 @@ async def start_focus_stacking(project_name: str, scan_index: int) -> Task: ) return existing_task + replaced_task_id = existing.task_id + + task_kwargs = {"depends_on": depends_on} if depends_on is not None else {} task = await task_manager.create_and_run_task( "focus_stacking_task", project_name, scan_index, + **task_kwargs, ) + if replaced_task_id: + await task_manager.replace_task(replaced_task_id, task.id) + scan.stacking_task_status = StackingTaskStatus(task_id=task.id, status=task.status) await project_manager.save_scan_state(scan) return task @@ -82,7 +95,7 @@ async def pause_focus_stacking(project_name: str, scan_index: int) -> Optional[T async def resume_focus_stacking(project_name: str, scan_index: int) -> Optional[Task]: - """Resume a paused focus stacking task and update the scan state.""" + """Resume a paused or interrupted focus stacking task and update the scan state.""" task_manager = get_task_manager() project_manager = get_project_manager() @@ -91,11 +104,37 @@ async def resume_focus_stacking(project_name: str, scan_index: int) -> Optional[ if scan is None: raise ValueError(f"Scan {scan_index} not found in project '{project_name}'") - if not scan.stacking_task_status or not scan.stacking_task_status.task_id: - logger.warning("Cannot resume focus stacking for scan %s: no paused task", scan_index) + stacking_status = scan.stacking_task_status + if not stacking_status: + logger.warning("Cannot resume focus stacking for scan %s: no task", scan_index) + return None + + if not stacking_status.task_id: + if stacking_status.status == TaskStatus.INTERRUPTED: + logger.info( + "Starting interrupted focus stacking for project '%s', scan %s.", + project_name, + scan_index, + ) + return await start_focus_stacking(project_name, scan_index) + + logger.warning( + "Cannot resume focus stacking for scan %s: no task ID", + scan_index, + ) + return None + + task = await task_manager.resume_task(stacking_status.task_id) + if task is None: + if stacking_status.status == TaskStatus.INTERRUPTED: + logger.info( + "Recreating missing interrupted focus stacking task for project '%s', scan %s.", + project_name, + scan_index, + ) + return await start_focus_stacking(project_name, scan_index) return None - task = await task_manager.resume_task(scan.stacking_task_status.task_id) scan.stacking_task_status.status = task.status await project_manager.save_scan_state(scan) return task diff --git a/openscan_firmware/controllers/services/projects.py b/openscan_firmware/controllers/services/projects.py index 610d170..7a86fe6 100644 --- a/openscan_firmware/controllers/services/projects.py +++ b/openscan_firmware/controllers/services/projects.py @@ -296,7 +296,7 @@ def _reset_incomplete_scans(self, project: Project) -> None: for scan in project.scans.values(): dirty = False - if scan.status in {TaskStatus.RUNNING, TaskStatus.PENDING}: + if scan.status in {TaskStatus.RUNNING, TaskStatus.PENDING, TaskStatus.PAUSED}: logger.debug( "Resetting scan %s for project %s from %s to interrupted", scan.index, @@ -304,7 +304,6 @@ def _reset_incomplete_scans(self, project: Project) -> None: scan.status.value, ) scan.status = TaskStatus.INTERRUPTED - scan.task_id = None dirty = True stacking_status = scan.stacking_task_status @@ -316,7 +315,6 @@ def _reset_incomplete_scans(self, project: Project) -> None: stacking_status.status.value if stacking_status.status else "unknown", ) stacking_status.status = TaskStatus.INTERRUPTED - stacking_status.task_id = None dirty = True if dirty: diff --git a/openscan_firmware/controllers/services/scans.py b/openscan_firmware/controllers/services/scans.py index a03e12c..9e4f0da 100644 --- a/openscan_firmware/controllers/services/scans.py +++ b/openscan_firmware/controllers/services/scans.py @@ -25,6 +25,7 @@ async def start_scan( scan: Scan, camera_controller: CameraController, start_from_step: int = 0, + depends_on: str | None = None, ) -> Task: """ Creates and starts a new scan task with simplified arguments. @@ -38,6 +39,7 @@ async def start_scan( scan: The scan object to be executed. camera_controller: The camera controller for validation. start_from_step: The step to resume the scan from. + depends_on: Optional ID of a task that must complete successfully first. Returns: The created Task object. @@ -50,6 +52,7 @@ async def start_scan( # If the scan already has a task_id, check its status. # This prevents creating a new task for a scan that is already running, paused, etc. + replaced_task_id: str | None = None if scan.task_id: existing_task = task_manager.get_task_info(scan.task_id) restartable_statuses = { @@ -74,19 +77,27 @@ async def start_scan( scan.task_id, start_from_step, ) - - # Remove the stale terminal task so the TaskManager list reflects only the new run - await task_manager.delete_task(existing_task.id) - scan.task_id = None + replaced_task_id = existing_task.id + else: + # Keep the stale ID long enough to repoint any dependents after the + # replacement task has been created. + replaced_task_id = scan.task_id task_name = "scan_task" + task_kwargs = {"depends_on": depends_on} if depends_on is not None else {} task = await task_manager.create_and_run_task( task_name, - scan, start_from_step + scan, + start_from_step, + **task_kwargs, ) + if replaced_task_id: + await task_manager.replace_task(replaced_task_id, task.id) + # Save the task_id in the scan object for future reference scan.task_id = task.id + scan.status = task.status await project_manager.save_scan_state(scan) logger.info(f"Started scan {scan.index} for project '{scan.project_name}' with task_id {task.id}") @@ -154,4 +165,4 @@ async def cancel_scan(scan: Scan) -> Optional[Task]: scan.status = TaskStatus.CANCELLED project_manager = get_project_manager() await project_manager.save_scan_state(scan) - return await task_manager.cancel_task(scan.task_id) \ No newline at end of file + return await task_manager.cancel_task(scan.task_id) diff --git a/openscan_firmware/controllers/services/tasks/core/focus_stacking_task.py b/openscan_firmware/controllers/services/tasks/core/focus_stacking_task.py index 940c3d1..bb905c7 100644 --- a/openscan_firmware/controllers/services/tasks/core/focus_stacking_task.py +++ b/openscan_firmware/controllers/services/tasks/core/focus_stacking_task.py @@ -82,8 +82,28 @@ async def run(self, project_name: str, scan_index: int) -> AsyncGenerator[TaskPr total_batches = len(batches) logger.info(f"Found {total_batches} focus stack batches to process") + # TaskManager persists the last yielded batch number. Capture it + # before emitting new progress so an interrupted task can skip + # batches whose output was already written before shutdown. + resume_from_batch = min( + max(int(self._task_model.progress.current), 0), + total_batches, + ) + if resume_from_batch: + logger.info( + "Resuming focus stacking for project '%s', scan %s from batch %s/%s", + project_name, + scan_index, + resume_from_batch, + total_batches, + ) + # Yield initial progress - yield TaskProgress(current=0, total=total_batches, message="Starting calibration...") + yield TaskProgress( + current=resume_from_batch, + total=total_batches, + message="Starting calibration...", + ) # Calibration phase (CPU-intensive, run in executor) logger.info(f"Calibrating with {num_calibration_batches} batches...") @@ -103,7 +123,11 @@ async def run(self, project_name: str, scan_index: int) -> AsyncGenerator[TaskPr calibration_path.write_text(json.dumps(calibration_payload, indent=2), encoding="utf-8") logger.info("Calibration complete") - yield TaskProgress(current=0, total=total_batches, message="Calibration complete, starting stacking...") + yield TaskProgress( + current=resume_from_batch, + total=total_batches, + message="Calibration complete, starting stacking...", + ) # Process all batches output_paths = [] @@ -111,6 +135,16 @@ async def run(self, project_name: str, scan_index: int) -> AsyncGenerator[TaskPr for idx, (position, image_paths) in enumerate(sorted(batches.items())): await self.wait_for_pause() + output_path = output_dir / f"stacked_scan{scan_index:02d}_{position:03d}.jpg" + if idx < resume_from_batch and output_path.exists(): + output_paths.append(str(output_path)) + logger.debug( + "Skipping already completed stacking batch %s (position %s)", + idx + 1, + position, + ) + continue + # Check for cancel if self.is_cancelled(): logger.info("Focus stacking cancelled by user") @@ -138,7 +172,6 @@ async def run(self, project_name: str, scan_index: int) -> AsyncGenerator[TaskPr return # Stack this batch (CPU-intensive, run in executor) - output_path = output_dir / f"stacked_scan{scan_index:02d}_{position:03d}.jpg" await loop.run_in_executor( None, self._stack_batch, diff --git a/openscan_firmware/controllers/services/tasks/task_manager.py b/openscan_firmware/controllers/services/tasks/task_manager.py index b20360e..00767d2 100644 --- a/openscan_firmware/controllers/services/tasks/task_manager.py +++ b/openscan_firmware/controllers/services/tasks/task_manager.py @@ -184,11 +184,21 @@ def restore_tasks_from_persistence(self): loaded_count += 1 continue - # Reset state for tasks that were running when the app was last closed - if task_model.status in [TaskStatus.RUNNING, TaskStatus.PAUSED]: + # Any non-terminal task from the previous process must be + # explicitly restarted by the user after startup. This + # includes queued and dependency-blocked tasks, so no + # hardware or other side effect can start automatically. + if task_model.status in [ + TaskStatus.PENDING, + TaskStatus.RUNNING, + TaskStatus.PAUSED, + ]: task_model.status = TaskStatus.INTERRUPTED task_model.error = "Task was interrupted by application shutdown." - logger.warning(f"Task '{task_model.name}' ({task_model.id}) was interrupted. Set to INTERRUPTED.") + logger.warning( + f"Task '{task_model.name}' ({task_model.id}) was interrupted. " + "Set to INTERRUPTED." + ) self._save_task_state(task_model) # Persist the new state interrupted_count += 1 @@ -306,6 +316,135 @@ def get_all_tasks_info(self) -> list[Task]: """Retrieves the data models for all tasks.""" return list(self._tasks.values()) + def _validate_dependency(self, task_model: Task) -> None: + """Validate that a task dependency exists and does not create a cycle.""" + dependency_id = task_model.depends_on + if dependency_id is None: + return + + visited = {task_model.id} + current_id: str | None = dependency_id + while current_id is not None: + if current_id in visited: + raise ValueError(f"Task dependency cycle detected for task '{task_model.id}'.") + + dependency = self._tasks.get(current_id) + if dependency is None: + raise ValueError(f"Dependency task '{current_id}' does not exist.") + + visited.add(current_id) + current_id = dependency.depends_on + + def _dependency_status(self, task_model: Task) -> TaskStatus | None: + """Return the current status of a task's dependency, if any.""" + if task_model.depends_on is None: + return None + + dependency = self._tasks.get(task_model.depends_on) + if dependency is None: + return TaskStatus.ERROR + return dependency.status + + def _dependencies_satisfied(self, task_model: Task) -> bool: + """Return whether the task's dependency has completed successfully.""" + dependency_status = self._dependency_status(task_model) + return dependency_status is None or dependency_status == TaskStatus.COMPLETED + + def _dependency_failed(self, task_model: Task) -> bool: + """Return whether the task's dependency reached an unsuccessful terminal state.""" + dependency_status = self._dependency_status(task_model) + return dependency_status in { + TaskStatus.ERROR, + TaskStatus.CANCELLED, + TaskStatus.INTERRUPTED, + } + + def _is_task_queued(self, task_id: str) -> bool: + """Return whether a task is already present in the scheduler queue.""" + return any( + task_instance.id == task_id + for task_instance, _, _ in self._pending_tasks._queue + ) + + async def _consume_completed_dependency(self, task_model: Task) -> None: + """Remove a successfully completed dependency from a waiting task.""" + if ( + task_model.depends_on is None + or self._dependency_status(task_model) != TaskStatus.COMPLETED + ): + return + + task_model.depends_on = None + self._save_task_state(task_model) + await task_event_publisher.publish(task_model, TaskEventType.UPDATE) + + async def _schedule_task_model(self, task_model: Task) -> None: + """Schedule a dependency-ready task using the normal scheduler rules.""" + if not self._dependencies_satisfied(task_model): + return + + if task_model.status != TaskStatus.PENDING or self._is_task_queued(task_model.id): + return + + await self._consume_completed_dependency(task_model) + + task_class = self._task_registry[task_model.name] + task_instance = task_class(task_model) + task_is_blocking = getattr(task_class, "is_blocking", False) + + should_queue_due_to_pending_exclusive = ( + not task_model.is_exclusive and self._has_pending_exclusive_task() + ) + + if ( + not should_queue_due_to_pending_exclusive + and self._can_run_task(task_model.is_exclusive, task_is_blocking) + ): + logger.info( + "Starting task '%s' (%s) immediately.", + task_model.name, + task_model.id, + ) + self._start_task_execution( + task_instance, + *task_model.run_args, + **task_model.run_kwargs, + ) + else: + logger.info("Queueing task '%s' (%s).", task_model.name, task_model.id) + await self._pending_tasks.put( + (task_instance, task_model.run_args, task_model.run_kwargs) + ) + + async def _mark_dependency_failed(self, task_model: Task, dependency: Task) -> None: + """Complete a waiting task with an error when its dependency failed.""" + if task_model.status != TaskStatus.PENDING: + return + + task_model.status = TaskStatus.ERROR + task_model.error = ( + f"Dependency task '{dependency.id}' finished with status " + f"'{dependency.status.value}'." + ) + task_model.completed_at = datetime.now() + self._save_task_state(task_model) + await task_event_publisher.publish(task_model, TaskEventType.UPDATE) + await self._handle_dependency_completion(task_model) + + async def _handle_dependency_completion(self, dependency: Task) -> None: + """Release, fail, or detach tasks that depend directly on a task.""" + for task_model in list(self._tasks.values()): + if task_model.depends_on != dependency.id: + continue + + if dependency.status == TaskStatus.COMPLETED: + if task_model.status in {TaskStatus.PENDING, TaskStatus.INTERRUPTED}: + await self._consume_completed_dependency(task_model) + if task_model.status == TaskStatus.PENDING: + await self._schedule_task_model(task_model) + elif task_model.status == TaskStatus.PENDING and self._dependency_failed(task_model): + await self._mark_dependency_failed(task_model, dependency) + async def delete_task(self, task_id: str) -> None: """ Deletes a task, removing it from memory and deleting its state from disk. @@ -343,9 +482,35 @@ async def delete_task(self, task_id: str) -> None: # Delete the persisted file self._delete_task_state(task_id) + async def replace_task(self, task_id: str, replacement_task_id: str) -> None: + """Replace a task while preserving references from dependent tasks.""" + if task_id == replacement_task_id: + raise ValueError("A task cannot replace itself.") + + replacement = self.get_task_info(replacement_task_id) + if replacement is None: + raise ValueError( + f"Replacement task '{replacement_task_id}' does not exist." + ) + + for dependent in self._tasks.values(): + if dependent.depends_on != task_id: + continue + + dependent.depends_on = replacement_task_id + self._save_task_state(dependent) + await task_event_publisher.publish(dependent, TaskEventType.UPDATE) + + await self.delete_task(task_id) + + # A very short replacement can complete before the references above + # are updated. Re-run dependency handling for that case. + if replacement.status == TaskStatus.COMPLETED: + await self._handle_dependency_completion(replacement) + async def wait_for_task(self, task_id: str, timeout: float = 20.0) -> Task: """ - Waits for a task to reach a terminal state (Completed, Error, Cancelled). + Waits for a task to reach a terminal state (Completed, Error, Cancelled, Interrupted). Args: task_id: The ID of the task to wait for. @@ -364,7 +529,12 @@ async def wait_for_task(self, task_id: str, timeout: float = 20.0) -> Task: start_time = time.time() while time.time() - start_time < timeout: task_model = self.get_task_info(task_id) - if task_model.status in [TaskStatus.COMPLETED, TaskStatus.ERROR, TaskStatus.CANCELLED]: + if task_model.status in [ + TaskStatus.COMPLETED, + TaskStatus.ERROR, + TaskStatus.CANCELLED, + TaskStatus.INTERRUPTED, + ]: # Give the event loop one last cycle to process any final updates in the wrapper await asyncio.sleep(0) return self.get_task_info(task_id) @@ -405,7 +575,13 @@ def _can_run_task(self, task_is_exclusive: bool, task_is_blocking: bool) -> bool # 4. A new async, non-exclusive task is subject to the async concurrency limit. return len(self._running_async_tasks) < self.max_concurrent_non_exclusive_tasks - async def create_and_run_task(self, task_name: str, *args: Any, **kwargs: Any) -> Task: + async def create_and_run_task( + self, + task_name: str, + *args: Any, + depends_on: str | None = None, + **kwargs: Any, + ) -> Task: """ Creates a new task. If possible, it starts the task immediately. Otherwise, the task is added to a pending queue. @@ -415,6 +591,7 @@ async def create_and_run_task(self, task_name: str, *args: Any, **kwargs: Any) - Args: task_name: The name of the registered task to run. *args: Positional arguments to pass to the task's run method. + depends_on: Optional ID of a task that must complete successfully first. **kwargs: Keyword arguments to pass to the task's run method. Returns: @@ -437,30 +614,29 @@ async def create_and_run_task(self, task_name: str, *args: Any, **kwargs: Any) - task_type=task_name, # Persist the task type for reconstruction is_exclusive=task_model_is_exclusive, is_blocking=task_is_blocking, + depends_on=depends_on, run_args=args, run_kwargs=kwargs ) + self._validate_dependency(task_model) logger.debug(f"Creating new task '{task_model.name}' ({task_model.id}) with args: {args}, kwargs: {kwargs}") - task_instance = task_class(task_model) # BaseTask instance self._tasks[task_model.id] = task_model self._save_task_state(task_model) # Persist immediately on creation await task_event_publisher.publish(task_model, TaskEventType.UPDATE) - # Scheduling Decision Point: - # A new non-exclusive task must be queued if an exclusive task is already pending. - should_queue_due_to_pending_exclusive = not task_model.is_exclusive and self._has_pending_exclusive_task() - - if not should_queue_due_to_pending_exclusive and self._can_run_task(task_model.is_exclusive, task_is_blocking): - logger.info(f"Starting task '{task_model.name}' ({task_model.id}) immediately.") - self._start_task_execution(task_instance, *args, **kwargs) + if self._dependency_failed(task_model): + dependency = self._tasks[task_model.depends_on] + await self._mark_dependency_failed(task_model, dependency) + elif self._dependencies_satisfied(task_model): + await self._schedule_task_model(task_model) else: - if should_queue_due_to_pending_exclusive: - logger.info(f"Queueing task '{task_model.name}' ({task_model.id}) because an exclusive task is pending.") - else: - logger.info(f"Queueing task '{task_model.name}' ({task_model.id}). Conditions not met for immediate start.") - await self._pending_tasks.put((task_instance, args, kwargs)) - # Task remains in PENDING status by default + logger.info( + "Holding task '%s' (%s) until dependency '%s' completes.", + task_model.name, + task_model.id, + task_model.depends_on, + ) return task_model @@ -564,17 +740,9 @@ async def _run_wrapper(self, task_instance: BaseTask, *args: Any, **kwargs: Any) if task_model.is_exclusive and self._active_exclusive_task_id == task_instance.id: self._active_exclusive_task_id = None - try: - loop = asyncio.get_running_loop() - except RuntimeError: - logger.debug( - "No running event loop; skipping scheduling of pending tasks after %s.", - task_model.id, - ) - else: - loop.create_task(self._try_run_pending_tasks()) # Non-blocking attempt to run next task - await task_event_publisher.publish(task_model, TaskEventType.UPDATE) + await self._handle_dependency_completion(task_model) + await self._try_run_pending_tasks() async def _try_run_pending_tasks(self) -> None: """Attempts to run tasks from the pending queue if conditions allow.""" @@ -647,7 +815,10 @@ async def cancel_task(self, task_id: str) -> Task | None: task_model.error = "Task was cancelled by user." self._save_task_state(task_model) await task_event_publisher.publish(task_model, TaskEventType.UPDATE) - # Note: completed_at will be set in _run_wrapper + # The wrapper may be cancelled before it gets to its post-run + # dependency handling, so notify dependents here as well. + await self._handle_dependency_completion(task_model) + # Note: completed_at will still be set in _run_wrapper return task_model # Case 2: Task is pending @@ -676,6 +847,24 @@ async def cancel_task(self, task_id: str) -> Task | None: await self._pending_tasks.put(item) if found_and_removed_from_queue: + await self._handle_dependency_completion(task_model) + return task_model + + # A dependency-blocked task is intentionally not present in the + # scheduler queue yet. It still needs to be cancellable through the + # normal task lifecycle API. + if task_model.status == TaskStatus.PENDING: + logger.info( + "Cancelling dependency-blocked task %s (%s).", + task_model.name, + task_id, + ) + task_model.status = TaskStatus.CANCELLED + task_model.error = "Task was cancelled by user." + task_model.completed_at = datetime.now() + self._save_task_state(task_model) + await task_event_publisher.publish(task_model, TaskEventType.UPDATE) + await self._handle_dependency_completion(task_model) return task_model # Case 3: Task is already completed, failed, or cancelled @@ -721,10 +910,12 @@ async def pause_task(self, task_id: str) -> Task | None: async def resume_task(self, task_id: str) -> Task | None: """ - Resumes a paused task. + Resumes a paused or interrupted task. This method is the single point of control for resuming a task. - It sets the task status back to RUNNING and signals the task to continue. + A paused task is signaled to continue in its existing execution context. + An interrupted task has no execution context after an application restart, + so it is scheduled again with its persisted arguments and progress. Args: task_id: The ID of the task to resume. @@ -732,11 +923,23 @@ async def resume_task(self, task_id: str) -> Task | None: Returns: The updated task model if found and resumed, otherwise None. """ + task_model = self.get_task_info(task_id) + if task_model is None: + logger.warning(f"Cannot resume task {task_id}: task does not exist.") + return None + + if task_model.status == TaskStatus.INTERRUPTED: + logger.info( + "Resuming interrupted task '%s' (%s) from persisted progress.", + task_model.name, + task_id, + ) + return await self._restart_task(task_model, reset_progress=False) + if task_id not in self._running_task_instances: - logger.warning(f"Cannot resume task {task_id}: not currently running or does not exist.") - return self.get_task_info(task_id) + logger.warning(f"Cannot resume task {task_id}: not currently running.") + return task_model - task_model = self._tasks[task_id] task_instance = self._running_task_instances[task_id] if task_model.status != TaskStatus.PAUSED: @@ -754,11 +957,11 @@ async def resume_task(self, task_id: str) -> Task | None: async def restart_task(self, task_id: str) -> Task | None: """ - Restarts a task that is in a CANCELLED or ERROR state. + Restarts a task that is in a CANCELLED, ERROR, or INTERRUPTED state. The task will be re-queued or run immediately with its original arguments. - The task's implementation is responsible for handling the continuation - from its last known progress. + Cancelled and failed tasks start with fresh progress. Interrupted tasks + retain their persisted progress so task implementations can continue. Args: task_id: The ID of the task to restart. @@ -767,7 +970,7 @@ async def restart_task(self, task_id: str) -> Task | None: The updated task model if found, otherwise None. """ task_model = self.get_task_info(task_id) - if not task_model: + if task_model is None: logger.warning(f"Attempted to restart non-existent task {task_id}.") return None @@ -775,31 +978,38 @@ async def restart_task(self, task_id: str) -> Task | None: logger.warning(f"Task {task_id} is not in a restartable state (status: {task_model.status}).") return task_model - # Reset task state for restart, but keep progress and original creation date + return await self._restart_task( + task_model, + reset_progress=task_model.status != TaskStatus.INTERRUPTED, + ) + + async def _restart_task(self, task_model: Task, *, reset_progress: bool) -> Task: + """Reset lifecycle state and schedule a previously stopped task.""" + + # Reset lifecycle state while keeping the original task identity and creation date. task_model.status = TaskStatus.PENDING task_model.started_at = None task_model.completed_at = None task_model.error = None task_model.result = None - task_model.progress = TaskProgress() # Resets progress + if reset_progress: + task_model.progress = TaskProgress() self._save_task_state(task_model) # Persist the reset state before queuing await task_event_publisher.publish(task_model, TaskEventType.UPDATE) - # A new task instance is required to re-run the logic - task_class = self._task_registry[task_model.name] - task_instance = task_class(task_model) - - # Use the original args and kwargs stored in the model - args = task_model.run_args - kwargs = task_model.run_kwargs - - if self._can_run_task(task_model.is_exclusive, task_instance.is_blocking): - logger.info(f"Restarting task '{task_model.name}' ({task_model.id}) immediately.") - self._start_task_execution(task_instance, *args, **kwargs) + if self._dependency_failed(task_model): + dependency = self._tasks[task_model.depends_on] + await self._mark_dependency_failed(task_model, dependency) + elif self._dependencies_satisfied(task_model): + await self._schedule_task_model(task_model) else: - logger.info(f"Queueing restarted task '{task_model.name}' ({task_model.id}).") - await self._pending_tasks.put((task_instance, args, kwargs)) + logger.info( + "Holding restarted task '%s' (%s) until dependency '%s' completes.", + task_model.name, + task_model.id, + task_model.depends_on, + ) return task_model diff --git a/openscan_firmware/models/task.py b/openscan_firmware/models/task.py index e03ee7d..534a028 100644 --- a/openscan_firmware/models/task.py +++ b/openscan_firmware/models/task.py @@ -36,5 +36,9 @@ class Task(BaseModel): created_at: datetime = Field(default_factory=datetime.now) started_at: Optional[datetime] = None completed_at: Optional[datetime] = None + depends_on: Optional[str] = Field( + default=None, + description="Task ID that must complete successfully before this task may run.", + ) run_args: tuple = Field(default_factory=tuple, description="Positional arguments the task was started with.") run_kwargs: dict = Field(default_factory=dict, description="Keyword arguments the task was started with.") diff --git a/openscan_firmware/routers/next/focus_stacking.py b/openscan_firmware/routers/next/focus_stacking.py index 1370521..b029fba 100644 --- a/openscan_firmware/routers/next/focus_stacking.py +++ b/openscan_firmware/routers/next/focus_stacking.py @@ -10,10 +10,18 @@ @router.post("/{project_name}/scans/{scan_index:int}/focus-stacking/start", response_model=Task) -async def start_focus_stacking(project_name: str, scan_index: int) -> Task: +async def start_focus_stacking( + project_name: str, + scan_index: int, + depends_on: str | None = None, +) -> Task: """Start focus stacking for a scan.""" try: - return await focus_service.start_focus_stacking(project_name, scan_index) + return await focus_service.start_focus_stacking( + project_name, + scan_index, + depends_on=depends_on, + ) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except Exception as exc: # pragma: no cover - unexpected errors bubble up as 500 @@ -38,7 +46,7 @@ async def pause_focus_stacking(project_name: str, scan_index: int) -> Task: @router.patch("/{project_name}/scans/{scan_index:int}/focus-stacking/resume", response_model=Task) async def resume_focus_stacking(project_name: str, scan_index: int) -> Task: - """Resume a paused focus stacking task.""" + """Resume a paused or interrupted focus stacking task.""" try: task = await focus_service.resume_focus_stacking(project_name, scan_index) except ValueError as exc: @@ -47,7 +55,7 @@ async def resume_focus_stacking(project_name: str, scan_index: int) -> Task: raise HTTPException(status_code=500, detail=str(exc)) from exc if task is None: - raise HTTPException(status_code=409, detail="Focus stacking is not paused") + raise HTTPException(status_code=409, detail="Focus stacking is not paused or interrupted") return task diff --git a/openscan_firmware/routers/next/projects.py b/openscan_firmware/routers/next/projects.py index dc9f982..02b8f56 100644 --- a/openscan_firmware/routers/next/projects.py +++ b/openscan_firmware/routers/next/projects.py @@ -62,6 +62,7 @@ class ScanCreateRequest(BaseModel): camera_name: str scan_settings: ScanSetting scan_description: str = "" + depends_on: str | None = None @router.get("/", response_model=dict[str, Project]) @@ -157,7 +158,12 @@ async def add_scan( request.scan_settings, request.scan_description, ) - task = await scans.start_scan(project_manager, scan, camera_controller) + task = await scans.start_scan( + project_manager, + scan, + camera_controller, + depends_on=request.depends_on, + ) return task except ValueError as exc: @@ -171,7 +177,11 @@ async def add_scan( @router.post("/{project_name}/upload", response_model=Task) -async def upload_project_to_cloud(project_name: str, token_override: Optional[str] = None) -> Task: +async def upload_project_to_cloud( + project_name: str, + token_override: Optional[str] = None, + depends_on: Optional[str] = None, +) -> Task: """Schedule an asynchronous cloud upload for a project. Args: @@ -182,7 +192,11 @@ async def upload_project_to_cloud(project_name: str, token_override: Optional[st Task: The TaskManager model describing the scheduled upload """ try: - task = await cloud.upload_project(project_name, token=token_override) + task = await cloud.upload_project( + project_name, + token=token_override, + depends_on=depends_on, + ) except cloud.CloudServiceError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc return task @@ -402,7 +416,9 @@ async def pause_scan(project_name: str, scan_index: int) -> Task: @router.patch("/{project_name}/scans/{scan_index:int}/resume", response_model=Task) async def resume_scan(project_name: str, scan_index: int, camera_name: str) -> Task: - """Resume a paused, cancelled or failed scan and return the resulting Task + """Resume a paused or interrupted scan, or restart a cancelled or failed scan. + + Return the resulting Task. Args: project_name: The name of the project diff --git a/openscan_firmware/routers/next/tasks.py b/openscan_firmware/routers/next/tasks.py index 81f181e..414a6f2 100644 --- a/openscan_firmware/routers/next/tasks.py +++ b/openscan_firmware/routers/next/tasks.py @@ -13,6 +13,15 @@ ) +_DOMAIN_TASK_ENDPOINTS = { + "scan_task": "POST /projects/{project_name}/scan", + "focus_stacking_task": ( + "POST /projects/{project_name}/scans/{scan_index}/focus-stacking/start" + ), + "cloud_upload_task": "POST /projects/{project_name}/upload", +} + + @router.get("/", response_model=List[Task]) async def get_all_tasks(): """ @@ -103,7 +112,7 @@ async def pause_task(task_id: str): @router.post("/{task_id}/resume", response_model=Task, summary="Resume a Task") async def resume_task(task_id: str): """ - Resumes a paused task. + Resumes a paused or interrupted task. Args: task_id: The ID of the task to resume. @@ -124,12 +133,21 @@ async def resume_task(task_id: str): async def create_task( task_name: str, args: List[Any] = Body(default=[], description="Positional arguments for the task"), - kwargs: Dict[str, Any] = Body(default={}, description="Keyword arguments for the task") + kwargs: Dict[str, Any] = Body(default={}, description="Keyword arguments for the task"), + depends_on: str | None = Body( + default=None, + description="Optional task ID that must complete successfully before this task runs", + ), ): """ - Create and start a new background task with optional parameters. + Create and start an experimental or custom background task. + + Domain-owned tasks such as scans, focus stacking, and cloud uploads must + be started through their project-specific endpoints. Those endpoints also + persist the task reference and maintain the corresponding domain status. The request body accepts: + - **depends_on**: Optional ID of a prerequisite task - **args**: List of positional arguments (e.g., `["project_name", 0]`) - **kwargs**: Dictionary of keyword arguments (e.g., `{"num_batches": 5}`) @@ -163,9 +181,25 @@ async def create_task( } ``` """ + domain_endpoint = _DOMAIN_TASK_ENDPOINTS.get(task_name) + if domain_endpoint: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=( + f"Task '{task_name}' is managed by a project-specific endpoint. " + f"Use {domain_endpoint} instead. The generic /tasks/{{task_name}} " + "endpoint is intended for experimental or custom tasks." + ), + ) + try: task_manager = get_task_manager() - task = await task_manager.create_and_run_task(task_name, *args, **kwargs) + task = await task_manager.create_and_run_task( + task_name, + *args, + depends_on=depends_on, + **kwargs, + ) return task except ValueError as e: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) \ No newline at end of file + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) diff --git a/openscan_firmware/routers/v0_8/focus_stacking.py b/openscan_firmware/routers/v0_8/focus_stacking.py index 1370521..b029fba 100644 --- a/openscan_firmware/routers/v0_8/focus_stacking.py +++ b/openscan_firmware/routers/v0_8/focus_stacking.py @@ -10,10 +10,18 @@ @router.post("/{project_name}/scans/{scan_index:int}/focus-stacking/start", response_model=Task) -async def start_focus_stacking(project_name: str, scan_index: int) -> Task: +async def start_focus_stacking( + project_name: str, + scan_index: int, + depends_on: str | None = None, +) -> Task: """Start focus stacking for a scan.""" try: - return await focus_service.start_focus_stacking(project_name, scan_index) + return await focus_service.start_focus_stacking( + project_name, + scan_index, + depends_on=depends_on, + ) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except Exception as exc: # pragma: no cover - unexpected errors bubble up as 500 @@ -38,7 +46,7 @@ async def pause_focus_stacking(project_name: str, scan_index: int) -> Task: @router.patch("/{project_name}/scans/{scan_index:int}/focus-stacking/resume", response_model=Task) async def resume_focus_stacking(project_name: str, scan_index: int) -> Task: - """Resume a paused focus stacking task.""" + """Resume a paused or interrupted focus stacking task.""" try: task = await focus_service.resume_focus_stacking(project_name, scan_index) except ValueError as exc: @@ -47,7 +55,7 @@ async def resume_focus_stacking(project_name: str, scan_index: int) -> Task: raise HTTPException(status_code=500, detail=str(exc)) from exc if task is None: - raise HTTPException(status_code=409, detail="Focus stacking is not paused") + raise HTTPException(status_code=409, detail="Focus stacking is not paused or interrupted") return task diff --git a/openscan_firmware/routers/v0_8/projects.py b/openscan_firmware/routers/v0_8/projects.py index 0dcac38..00d8022 100644 --- a/openscan_firmware/routers/v0_8/projects.py +++ b/openscan_firmware/routers/v0_8/projects.py @@ -113,7 +113,8 @@ async def new_project(project_name: str, project_description: Optional[str] = "" async def add_scan_with_description(project_name: str, camera_name: str, scan_settings: ScanSetting, - scan_description: Optional[str] = "") -> Task: + scan_description: Optional[str] = "", + depends_on: Optional[str] = None) -> Task: """Add a new scan to a project and return the created Task Args: @@ -130,7 +131,12 @@ async def add_scan_with_description(project_name: str, try: scan = project_manager.add_scan(project_name, camera_controller, scan_settings, scan_description) - task = await scans.start_scan(project_manager, scan, camera_controller) + task = await scans.start_scan( + project_manager, + scan, + camera_controller, + depends_on=depends_on, + ) return task except ValueError as exc: @@ -144,7 +150,11 @@ async def add_scan_with_description(project_name: str, @router.post("/{project_name}/upload", response_model=Task) -async def upload_project_to_cloud(project_name: str, token_override: Optional[str] = None) -> Task: +async def upload_project_to_cloud( + project_name: str, + token_override: Optional[str] = None, + depends_on: Optional[str] = None, +) -> Task: """Schedule an asynchronous cloud upload for a project. Args: @@ -155,7 +165,11 @@ async def upload_project_to_cloud(project_name: str, token_override: Optional[st Task: The TaskManager model describing the scheduled upload """ try: - task = await cloud.upload_project(project_name, token=token_override) + task = await cloud.upload_project( + project_name, + token=token_override, + depends_on=depends_on, + ) except cloud.CloudServiceError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc return task @@ -366,7 +380,9 @@ async def pause_scan(project_name: str, scan_index: int) -> Task: @router.patch("/{project_name}/scans/{scan_index:int}/resume", response_model=Task) async def resume_scan(project_name: str, scan_index: int, camera_name: str) -> Task: - """Resume a paused, cancelled or failed scan and return the resulting Task + """Resume a paused or interrupted scan, or restart a cancelled or failed scan. + + Return the resulting Task. Args: project_name: The name of the project diff --git a/openscan_firmware/routers/v0_8/tasks.py b/openscan_firmware/routers/v0_8/tasks.py index 81f181e..e313aec 100644 --- a/openscan_firmware/routers/v0_8/tasks.py +++ b/openscan_firmware/routers/v0_8/tasks.py @@ -103,7 +103,7 @@ async def pause_task(task_id: str): @router.post("/{task_id}/resume", response_model=Task, summary="Resume a Task") async def resume_task(task_id: str): """ - Resumes a paused task. + Resumes a paused or interrupted task. Args: task_id: The ID of the task to resume. @@ -124,12 +124,17 @@ async def resume_task(task_id: str): async def create_task( task_name: str, args: List[Any] = Body(default=[], description="Positional arguments for the task"), - kwargs: Dict[str, Any] = Body(default={}, description="Keyword arguments for the task") + kwargs: Dict[str, Any] = Body(default={}, description="Keyword arguments for the task"), + depends_on: str | None = Body( + default=None, + description="Optional task ID that must complete successfully before this task runs", + ), ): """ Create and start a new background task with optional parameters. The request body accepts: + - **depends_on**: Optional ID of a prerequisite task - **args**: List of positional arguments (e.g., `["project_name", 0]`) - **kwargs**: Dictionary of keyword arguments (e.g., `{"num_batches": 5}`) @@ -165,7 +170,12 @@ async def create_task( """ try: task_manager = get_task_manager() - task = await task_manager.create_and_run_task(task_name, *args, **kwargs) + task = await task_manager.create_and_run_task( + task_name, + *args, + depends_on=depends_on, + **kwargs, + ) return task except ValueError as e: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) \ No newline at end of file + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) diff --git a/openscan_firmware/routers/v0_9/focus_stacking.py b/openscan_firmware/routers/v0_9/focus_stacking.py index 1370521..b029fba 100644 --- a/openscan_firmware/routers/v0_9/focus_stacking.py +++ b/openscan_firmware/routers/v0_9/focus_stacking.py @@ -10,10 +10,18 @@ @router.post("/{project_name}/scans/{scan_index:int}/focus-stacking/start", response_model=Task) -async def start_focus_stacking(project_name: str, scan_index: int) -> Task: +async def start_focus_stacking( + project_name: str, + scan_index: int, + depends_on: str | None = None, +) -> Task: """Start focus stacking for a scan.""" try: - return await focus_service.start_focus_stacking(project_name, scan_index) + return await focus_service.start_focus_stacking( + project_name, + scan_index, + depends_on=depends_on, + ) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except Exception as exc: # pragma: no cover - unexpected errors bubble up as 500 @@ -38,7 +46,7 @@ async def pause_focus_stacking(project_name: str, scan_index: int) -> Task: @router.patch("/{project_name}/scans/{scan_index:int}/focus-stacking/resume", response_model=Task) async def resume_focus_stacking(project_name: str, scan_index: int) -> Task: - """Resume a paused focus stacking task.""" + """Resume a paused or interrupted focus stacking task.""" try: task = await focus_service.resume_focus_stacking(project_name, scan_index) except ValueError as exc: @@ -47,7 +55,7 @@ async def resume_focus_stacking(project_name: str, scan_index: int) -> Task: raise HTTPException(status_code=500, detail=str(exc)) from exc if task is None: - raise HTTPException(status_code=409, detail="Focus stacking is not paused") + raise HTTPException(status_code=409, detail="Focus stacking is not paused or interrupted") return task diff --git a/openscan_firmware/routers/v0_9/projects.py b/openscan_firmware/routers/v0_9/projects.py index 0dcac38..00d8022 100644 --- a/openscan_firmware/routers/v0_9/projects.py +++ b/openscan_firmware/routers/v0_9/projects.py @@ -113,7 +113,8 @@ async def new_project(project_name: str, project_description: Optional[str] = "" async def add_scan_with_description(project_name: str, camera_name: str, scan_settings: ScanSetting, - scan_description: Optional[str] = "") -> Task: + scan_description: Optional[str] = "", + depends_on: Optional[str] = None) -> Task: """Add a new scan to a project and return the created Task Args: @@ -130,7 +131,12 @@ async def add_scan_with_description(project_name: str, try: scan = project_manager.add_scan(project_name, camera_controller, scan_settings, scan_description) - task = await scans.start_scan(project_manager, scan, camera_controller) + task = await scans.start_scan( + project_manager, + scan, + camera_controller, + depends_on=depends_on, + ) return task except ValueError as exc: @@ -144,7 +150,11 @@ async def add_scan_with_description(project_name: str, @router.post("/{project_name}/upload", response_model=Task) -async def upload_project_to_cloud(project_name: str, token_override: Optional[str] = None) -> Task: +async def upload_project_to_cloud( + project_name: str, + token_override: Optional[str] = None, + depends_on: Optional[str] = None, +) -> Task: """Schedule an asynchronous cloud upload for a project. Args: @@ -155,7 +165,11 @@ async def upload_project_to_cloud(project_name: str, token_override: Optional[st Task: The TaskManager model describing the scheduled upload """ try: - task = await cloud.upload_project(project_name, token=token_override) + task = await cloud.upload_project( + project_name, + token=token_override, + depends_on=depends_on, + ) except cloud.CloudServiceError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc return task @@ -366,7 +380,9 @@ async def pause_scan(project_name: str, scan_index: int) -> Task: @router.patch("/{project_name}/scans/{scan_index:int}/resume", response_model=Task) async def resume_scan(project_name: str, scan_index: int, camera_name: str) -> Task: - """Resume a paused, cancelled or failed scan and return the resulting Task + """Resume a paused or interrupted scan, or restart a cancelled or failed scan. + + Return the resulting Task. Args: project_name: The name of the project diff --git a/openscan_firmware/routers/v0_9/tasks.py b/openscan_firmware/routers/v0_9/tasks.py index 81f181e..e313aec 100644 --- a/openscan_firmware/routers/v0_9/tasks.py +++ b/openscan_firmware/routers/v0_9/tasks.py @@ -103,7 +103,7 @@ async def pause_task(task_id: str): @router.post("/{task_id}/resume", response_model=Task, summary="Resume a Task") async def resume_task(task_id: str): """ - Resumes a paused task. + Resumes a paused or interrupted task. Args: task_id: The ID of the task to resume. @@ -124,12 +124,17 @@ async def resume_task(task_id: str): async def create_task( task_name: str, args: List[Any] = Body(default=[], description="Positional arguments for the task"), - kwargs: Dict[str, Any] = Body(default={}, description="Keyword arguments for the task") + kwargs: Dict[str, Any] = Body(default={}, description="Keyword arguments for the task"), + depends_on: str | None = Body( + default=None, + description="Optional task ID that must complete successfully before this task runs", + ), ): """ Create and start a new background task with optional parameters. The request body accepts: + - **depends_on**: Optional ID of a prerequisite task - **args**: List of positional arguments (e.g., `["project_name", 0]`) - **kwargs**: Dictionary of keyword arguments (e.g., `{"num_batches": 5}`) @@ -165,7 +170,12 @@ async def create_task( """ try: task_manager = get_task_manager() - task = await task_manager.create_and_run_task(task_name, *args, **kwargs) + task = await task_manager.create_and_run_task( + task_name, + *args, + depends_on=depends_on, + **kwargs, + ) return task except ValueError as e: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) \ No newline at end of file + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) diff --git a/scripts/openapi/openapi_latest.json b/scripts/openapi/openapi_latest.json index 93f4c11..1074f54 100644 --- a/scripts/openapi/openapi_latest.json +++ b/scripts/openapi/openapi_latest.json @@ -1900,6 +1900,22 @@ "default": "", "title": "Scan Description" } + }, + { + "name": "depends_on", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On" + } } ], "requestBody": { @@ -1972,6 +1988,22 @@ ], "title": "Token Override" } + }, + { + "name": "depends_on", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On" + } } ], "responses": { @@ -2424,7 +2456,7 @@ "projects" ], "summary": "Resume Scan", - "description": "Resume a paused, cancelled or failed scan and return the resulting Task\n\nArgs:\n project_name: The name of the project\n scan_index: The index of the scan to resume\n camera_name: The name of the camera to use for the scan\n\nReturns:\n Task: The resumed or restarted task", + "description": "Resume a paused or interrupted scan, or restart a cancelled or failed scan.\n\nReturn the resulting Task.\n\nArgs:\n project_name: The name of the project\n scan_index: The index of the scan to resume\n camera_name: The name of the camera to use for the scan\n\nReturns:\n Task: The resumed or restarted task", "operationId": "resume_scan", "parameters": [ { @@ -3628,7 +3660,7 @@ "tasks" ], "summary": "Resume a Task", - "description": "Resumes a paused task.\n\nArgs:\n task_id: The ID of the task to resume.\n\nReturns:\n Task: The task object with its status and details.", + "description": "Resumes a paused or interrupted task.\n\nArgs:\n task_id: The ID of the task to resume.\n\nReturns:\n Task: The task object with its status and details.", "operationId": "resume_task", "parameters": [ { @@ -3674,7 +3706,7 @@ "tasks" ], "summary": "Create Task", - "description": "Create and start a new background task with optional parameters.\n\nThe request body accepts:\n- **args**: List of positional arguments (e.g., `[\"project_name\", 0]`)\n- **kwargs**: Dictionary of keyword arguments (e.g., `{\"num_batches\": 5}`)\n\nArgs:\n task_name: The name of the task to create, as registered in the TaskManager.\n args: Positional arguments to pass to the task's run method.\n kwargs: Keyword arguments to pass to the task's run method.\n\nReturns:\n The created task object.\n\nExamples:\n ```json\n // No parameters\n {}\n\n // With positional args\n {\n \"args\": [\"MyProject\", 0]\n }\n\n // With keyword args\n {\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n\n // With both\n {\n \"args\": [\"MyProject\", 0],\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n ```", + "description": "Create and start a new background task with optional parameters.\n\nThe request body accepts:\n- **depends_on**: Optional ID of a prerequisite task\n- **args**: List of positional arguments (e.g., `[\"project_name\", 0]`)\n- **kwargs**: Dictionary of keyword arguments (e.g., `{\"num_batches\": 5}`)\n\nArgs:\n task_name: The name of the task to create, as registered in the TaskManager.\n args: Positional arguments to pass to the task's run method.\n kwargs: Keyword arguments to pass to the task's run method.\n\nReturns:\n The created task object.\n\nExamples:\n ```json\n // No parameters\n {}\n\n // With positional args\n {\n \"args\": [\"MyProject\", 0]\n }\n\n // With keyword args\n {\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n\n // With both\n {\n \"args\": [\"MyProject\", 0],\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n ```", "operationId": "create_task", "parameters": [ { @@ -4415,6 +4447,22 @@ "type": "integer", "title": "Scan Index" } + }, + { + "name": "depends_on", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On" + } } ], "responses": { @@ -4499,7 +4547,7 @@ "focus_stacking" ], "summary": "Resume Focus Stacking", - "description": "Resume a paused focus stacking task.", + "description": "Resume a paused or interrupted focus stacking task.", "operationId": "resume_focus_stacking", "parameters": [ { @@ -4683,6 +4731,18 @@ "title": "Kwargs", "description": "Keyword arguments for the task", "default": {} + }, + "depends_on": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On", + "description": "Optional task ID that must complete successfully before this task runs" } }, "type": "object", @@ -6682,6 +6742,18 @@ ], "title": "Completed At" }, + "depends_on": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On", + "description": "Task ID that must complete successfully before this task may run." + }, "run_args": { "items": {}, "type": "array", @@ -6806,7 +6878,7 @@ "reboot_required" ], "title": "UpdateInstallResponse", - "description": "Result of a synchronous user-requested update installation." + "description": "Acceptance or result of a user-requested update installation." }, "UpdateStatusResponse": { "properties": { @@ -6904,4 +6976,4 @@ } } } -} +} \ No newline at end of file diff --git a/scripts/openapi/openapi_next.json b/scripts/openapi/openapi_next.json index 6522811..6215434 100644 --- a/scripts/openapi/openapi_next.json +++ b/scripts/openapi/openapi_next.json @@ -2093,6 +2093,22 @@ ], "title": "Token Override" } + }, + { + "name": "depends_on", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On" + } } ], "responses": { @@ -2551,7 +2567,7 @@ "projects" ], "summary": "Resume Scan", - "description": "Resume a paused, cancelled or failed scan and return the resulting Task\n\nArgs:\n project_name: The name of the project\n scan_index: The index of the scan to resume\n camera_name: The name of the camera to use for the scan\n\nReturns:\n Task: The resumed or restarted task", + "description": "Resume a paused or interrupted scan, or restart a cancelled or failed scan.\n\nReturn the resulting Task.\n\nArgs:\n project_name: The name of the project\n scan_index: The index of the scan to resume\n camera_name: The name of the camera to use for the scan\n\nReturns:\n Task: The resumed or restarted task", "operationId": "resume_scan", "parameters": [ { @@ -3596,7 +3612,7 @@ "tasks" ], "summary": "Resume a Task", - "description": "Resumes a paused task.\n\nArgs:\n task_id: The ID of the task to resume.\n\nReturns:\n Task: The task object with its status and details.", + "description": "Resumes a paused or interrupted task.\n\nArgs:\n task_id: The ID of the task to resume.\n\nReturns:\n Task: The task object with its status and details.", "operationId": "resume_task", "parameters": [ { @@ -3642,7 +3658,7 @@ "tasks" ], "summary": "Create Task", - "description": "Create and start a new background task with optional parameters.\n\nThe request body accepts:\n- **args**: List of positional arguments (e.g., `[\"project_name\", 0]`)\n- **kwargs**: Dictionary of keyword arguments (e.g., `{\"num_batches\": 5}`)\n\nArgs:\n task_name: The name of the task to create, as registered in the TaskManager.\n args: Positional arguments to pass to the task's run method.\n kwargs: Keyword arguments to pass to the task's run method.\n\nReturns:\n The created task object.\n\nExamples:\n ```json\n // No parameters\n {}\n\n // With positional args\n {\n \"args\": [\"MyProject\", 0]\n }\n\n // With keyword args\n {\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n\n // With both\n {\n \"args\": [\"MyProject\", 0],\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n ```", + "description": "Create and start an experimental or custom background task.\n\nDomain-owned tasks such as scans, focus stacking, and cloud uploads must\nbe started through their project-specific endpoints. Those endpoints also\npersist the task reference and maintain the corresponding domain status.\n\nThe request body accepts:\n- **depends_on**: Optional ID of a prerequisite task\n- **args**: List of positional arguments (e.g., `[\"project_name\", 0]`)\n- **kwargs**: Dictionary of keyword arguments (e.g., `{\"num_batches\": 5}`)\n\nArgs:\n task_name: The name of the task to create, as registered in the TaskManager.\n args: Positional arguments to pass to the task's run method.\n kwargs: Keyword arguments to pass to the task's run method.\n\nReturns:\n The created task object.\n\nExamples:\n ```json\n // No parameters\n {}\n\n // With positional args\n {\n \"args\": [\"MyProject\", 0]\n }\n\n // With keyword args\n {\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n\n // With both\n {\n \"args\": [\"MyProject\", 0],\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n ```", "operationId": "create_task", "parameters": [ { @@ -5154,6 +5170,22 @@ "type": "integer", "title": "Scan Index" } + }, + { + "name": "depends_on", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On" + } } ], "responses": { @@ -5238,7 +5270,7 @@ "focus_stacking" ], "summary": "Resume Focus Stacking", - "description": "Resume a paused focus stacking task.", + "description": "Resume a paused or interrupted focus stacking task.", "operationId": "resume_focus_stacking", "parameters": [ { @@ -5494,6 +5526,18 @@ "title": "Kwargs", "description": "Keyword arguments for the task", "default": {} + }, + "depends_on": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On", + "description": "Optional task ID that must complete successfully before this task runs" } }, "type": "object", @@ -7328,6 +7372,17 @@ "type": "string", "title": "Scan Description", "default": "" + }, + "depends_on": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On" } }, "type": "object", @@ -7869,6 +7924,18 @@ ], "title": "Completed At" }, + "depends_on": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On", + "description": "Task ID that must complete successfully before this task may run." + }, "run_args": { "items": {}, "type": "array", @@ -8100,7 +8167,7 @@ "reboot_required" ], "title": "UpdateInstallResponse", - "description": "Result of a synchronous user-requested update installation." + "description": "Acceptance or result of a user-requested update installation." }, "UpdateStatusResponse": { "properties": { @@ -8198,4 +8265,4 @@ } } } -} +} \ No newline at end of file diff --git a/scripts/openapi/openapi_v0.8.json b/scripts/openapi/openapi_v0.8.json index bae2118..7d8fed6 100644 --- a/scripts/openapi/openapi_v0.8.json +++ b/scripts/openapi/openapi_v0.8.json @@ -1604,6 +1604,22 @@ "default": "", "title": "Scan Description" } + }, + { + "name": "depends_on", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On" + } } ], "requestBody": { @@ -1676,6 +1692,22 @@ ], "title": "Token Override" } + }, + { + "name": "depends_on", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On" + } } ], "responses": { @@ -2128,7 +2160,7 @@ "projects" ], "summary": "Resume Scan", - "description": "Resume a paused, cancelled or failed scan and return the resulting Task\n\nArgs:\n project_name: The name of the project\n scan_index: The index of the scan to resume\n camera_name: The name of the camera to use for the scan\n\nReturns:\n Task: The resumed or restarted task", + "description": "Resume a paused or interrupted scan, or restart a cancelled or failed scan.\n\nReturn the resulting Task.\n\nArgs:\n project_name: The name of the project\n scan_index: The index of the scan to resume\n camera_name: The name of the camera to use for the scan\n\nReturns:\n Task: The resumed or restarted task", "operationId": "resume_scan", "parameters": [ { @@ -3263,7 +3295,7 @@ "tasks" ], "summary": "Resume a Task", - "description": "Resumes a paused task.\n\nArgs:\n task_id: The ID of the task to resume.\n\nReturns:\n Task: The task object with its status and details.", + "description": "Resumes a paused or interrupted task.\n\nArgs:\n task_id: The ID of the task to resume.\n\nReturns:\n Task: The task object with its status and details.", "operationId": "resume_task", "parameters": [ { @@ -3309,7 +3341,7 @@ "tasks" ], "summary": "Create Task", - "description": "Create and start a new background task with optional parameters.\n\nThe request body accepts:\n- **args**: List of positional arguments (e.g., `[\"project_name\", 0]`)\n- **kwargs**: Dictionary of keyword arguments (e.g., `{\"num_batches\": 5}`)\n\nArgs:\n task_name: The name of the task to create, as registered in the TaskManager.\n args: Positional arguments to pass to the task's run method.\n kwargs: Keyword arguments to pass to the task's run method.\n\nReturns:\n The created task object.\n\nExamples:\n ```json\n // No parameters\n {}\n\n // With positional args\n {\n \"args\": [\"MyProject\", 0]\n }\n\n // With keyword args\n {\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n\n // With both\n {\n \"args\": [\"MyProject\", 0],\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n ```", + "description": "Create and start a new background task with optional parameters.\n\nThe request body accepts:\n- **depends_on**: Optional ID of a prerequisite task\n- **args**: List of positional arguments (e.g., `[\"project_name\", 0]`)\n- **kwargs**: Dictionary of keyword arguments (e.g., `{\"num_batches\": 5}`)\n\nArgs:\n task_name: The name of the task to create, as registered in the TaskManager.\n args: Positional arguments to pass to the task's run method.\n kwargs: Keyword arguments to pass to the task's run method.\n\nReturns:\n The created task object.\n\nExamples:\n ```json\n // No parameters\n {}\n\n // With positional args\n {\n \"args\": [\"MyProject\", 0]\n }\n\n // With keyword args\n {\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n\n // With both\n {\n \"args\": [\"MyProject\", 0],\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n ```", "operationId": "create_task", "parameters": [ { @@ -3941,6 +3973,22 @@ "type": "integer", "title": "Scan Index" } + }, + { + "name": "depends_on", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On" + } } ], "responses": { @@ -4025,7 +4073,7 @@ "focus_stacking" ], "summary": "Resume Focus Stacking", - "description": "Resume a paused focus stacking task.", + "description": "Resume a paused or interrupted focus stacking task.", "operationId": "resume_focus_stacking", "parameters": [ { @@ -4209,6 +4257,18 @@ "title": "Kwargs", "description": "Keyword arguments for the task", "default": {} + }, + "depends_on": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On", + "description": "Optional task ID that must complete successfully before this task runs" } }, "type": "object", @@ -6032,6 +6092,18 @@ ], "title": "Completed At" }, + "depends_on": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On", + "description": "Task ID that must complete successfully before this task may run." + }, "run_args": { "items": {}, "type": "array", diff --git a/scripts/openapi/openapi_v0.9.json b/scripts/openapi/openapi_v0.9.json index 93f4c11..1074f54 100644 --- a/scripts/openapi/openapi_v0.9.json +++ b/scripts/openapi/openapi_v0.9.json @@ -1900,6 +1900,22 @@ "default": "", "title": "Scan Description" } + }, + { + "name": "depends_on", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On" + } } ], "requestBody": { @@ -1972,6 +1988,22 @@ ], "title": "Token Override" } + }, + { + "name": "depends_on", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On" + } } ], "responses": { @@ -2424,7 +2456,7 @@ "projects" ], "summary": "Resume Scan", - "description": "Resume a paused, cancelled or failed scan and return the resulting Task\n\nArgs:\n project_name: The name of the project\n scan_index: The index of the scan to resume\n camera_name: The name of the camera to use for the scan\n\nReturns:\n Task: The resumed or restarted task", + "description": "Resume a paused or interrupted scan, or restart a cancelled or failed scan.\n\nReturn the resulting Task.\n\nArgs:\n project_name: The name of the project\n scan_index: The index of the scan to resume\n camera_name: The name of the camera to use for the scan\n\nReturns:\n Task: The resumed or restarted task", "operationId": "resume_scan", "parameters": [ { @@ -3628,7 +3660,7 @@ "tasks" ], "summary": "Resume a Task", - "description": "Resumes a paused task.\n\nArgs:\n task_id: The ID of the task to resume.\n\nReturns:\n Task: The task object with its status and details.", + "description": "Resumes a paused or interrupted task.\n\nArgs:\n task_id: The ID of the task to resume.\n\nReturns:\n Task: The task object with its status and details.", "operationId": "resume_task", "parameters": [ { @@ -3674,7 +3706,7 @@ "tasks" ], "summary": "Create Task", - "description": "Create and start a new background task with optional parameters.\n\nThe request body accepts:\n- **args**: List of positional arguments (e.g., `[\"project_name\", 0]`)\n- **kwargs**: Dictionary of keyword arguments (e.g., `{\"num_batches\": 5}`)\n\nArgs:\n task_name: The name of the task to create, as registered in the TaskManager.\n args: Positional arguments to pass to the task's run method.\n kwargs: Keyword arguments to pass to the task's run method.\n\nReturns:\n The created task object.\n\nExamples:\n ```json\n // No parameters\n {}\n\n // With positional args\n {\n \"args\": [\"MyProject\", 0]\n }\n\n // With keyword args\n {\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n\n // With both\n {\n \"args\": [\"MyProject\", 0],\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n ```", + "description": "Create and start a new background task with optional parameters.\n\nThe request body accepts:\n- **depends_on**: Optional ID of a prerequisite task\n- **args**: List of positional arguments (e.g., `[\"project_name\", 0]`)\n- **kwargs**: Dictionary of keyword arguments (e.g., `{\"num_batches\": 5}`)\n\nArgs:\n task_name: The name of the task to create, as registered in the TaskManager.\n args: Positional arguments to pass to the task's run method.\n kwargs: Keyword arguments to pass to the task's run method.\n\nReturns:\n The created task object.\n\nExamples:\n ```json\n // No parameters\n {}\n\n // With positional args\n {\n \"args\": [\"MyProject\", 0]\n }\n\n // With keyword args\n {\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n\n // With both\n {\n \"args\": [\"MyProject\", 0],\n \"kwargs\": {\"num_calibration_batches\": 5}\n }\n ```", "operationId": "create_task", "parameters": [ { @@ -4415,6 +4447,22 @@ "type": "integer", "title": "Scan Index" } + }, + { + "name": "depends_on", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On" + } } ], "responses": { @@ -4499,7 +4547,7 @@ "focus_stacking" ], "summary": "Resume Focus Stacking", - "description": "Resume a paused focus stacking task.", + "description": "Resume a paused or interrupted focus stacking task.", "operationId": "resume_focus_stacking", "parameters": [ { @@ -4683,6 +4731,18 @@ "title": "Kwargs", "description": "Keyword arguments for the task", "default": {} + }, + "depends_on": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On", + "description": "Optional task ID that must complete successfully before this task runs" } }, "type": "object", @@ -6682,6 +6742,18 @@ ], "title": "Completed At" }, + "depends_on": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Depends On", + "description": "Task ID that must complete successfully before this task may run." + }, "run_args": { "items": {}, "type": "array", @@ -6806,7 +6878,7 @@ "reboot_required" ], "title": "UpdateInstallResponse", - "description": "Result of a synchronous user-requested update installation." + "description": "Acceptance or result of a user-requested update installation." }, "UpdateStatusResponse": { "properties": { @@ -6904,4 +6976,4 @@ } } } -} +} \ No newline at end of file diff --git a/tests/controllers/services/tasks/test_focus_stacking_task.py b/tests/controllers/services/tasks/test_focus_stacking_task.py index f7fee63..e81a5a5 100644 --- a/tests/controllers/services/tasks/test_focus_stacking_task.py +++ b/tests/controllers/services/tasks/test_focus_stacking_task.py @@ -6,7 +6,7 @@ import pytest from openscan_firmware.controllers.services.tasks.core.focus_stacking_task import FocusStackingTask -from openscan_firmware.models.task import TaskStatus +from openscan_firmware.models.task import Task, TaskProgress, TaskStatus async def wait_for_status(task_manager, task_id: str, expected_status: TaskStatus, timeout: float = 5.0): @@ -109,6 +109,43 @@ async def test_focus_stacking_task_happy_path( assert updated_scan.stacked_size_bytes > 0 +@pytest.mark.asyncio +async def test_focus_stacking_task_resumes_from_persisted_batch_progress( + monkeypatch, + focus_task_manager, + focus_stacking_environment, + focus_stacking_batches, +): + stack_impl = make_writing_stack_impl() + configure_focus_stacking_task(monkeypatch, focus_stacking_environment, focus_stacking_batches, stack_impl) + + project = focus_stacking_environment["project"] + scan = focus_stacking_environment["scan"] + first_output = focus_stacking_environment["stacked_dir"] / f"stacked_scan{scan.index:02d}_001.jpg" + first_output.write_bytes(b"already completed") + + task_model = Task( + name="focus_stacking_task", + task_type="focus_stacking_task", + status=TaskStatus.INTERRUPTED, + progress=TaskProgress(current=1, total=2, message="Stacking batch 1 of 2"), + run_args=(project.name, scan.index), + ) + focus_task_manager._tasks[task_model.id] = task_model + focus_task_manager._save_task_state(task_model) + + resumed_task = await focus_task_manager.resume_task(task_model.id) + assert resumed_task is not None + + final_state = await wait_for_status(focus_task_manager, task_model.id, TaskStatus.COMPLETED) + + second_output = focus_stacking_environment["stacked_dir"] / f"stacked_scan{scan.index:02d}_002.jpg" + assert first_output.read_bytes() == b"already completed" + assert second_output.read_bytes() == b"stacked" + assert stack_impl.call_counter["value"] == 1 + assert final_state.result["stacked_image_count"] == 2 + + @pytest.mark.asyncio async def test_focus_stacking_task_pause_and_resume( monkeypatch, diff --git a/tests/controllers/services/test_cloud.py b/tests/controllers/services/test_cloud.py index 1703343..45c9950 100644 --- a/tests/controllers/services/test_cloud.py +++ b/tests/controllers/services/test_cloud.py @@ -86,11 +86,14 @@ async def test_upload_project_rejects_uploaded_project(monkeypatch, project_mana @pytest.mark.asyncio -async def test_upload_project_rejects_running_task(monkeypatch, project_manager, task_manager): +@pytest.mark.parametrize("status", [TaskStatus.RUNNING, TaskStatus.INTERRUPTED]) +async def test_upload_project_rejects_active_or_interrupted_task( + monkeypatch, project_manager, task_manager, status +): task = Task( name="cloud_upload_task", task_type="cloud_upload_task", - status=TaskStatus.RUNNING, + status=status, run_args=("demo",), ) task_manager.add_task(task) @@ -111,6 +114,7 @@ async def test_upload_project_starts_when_not_blocked(monkeypatch, project_manag created_task = Task(name="cloud_upload_task", task_type="cloud_upload_task") async def fake_create_and_run(task_name, project_name, **kwargs): + assert kwargs["depends_on"] == "task-prerequisite" task_manager.add_task(created_task) return created_task @@ -124,7 +128,7 @@ async def fake_create_and_run(task_name, project_name, **kwargs): ) monkeypatch.setattr(task_manager, "create_and_run_task", fake_create_and_run) - task = await upload_project("demo") + task = await upload_project("demo", depends_on="task-prerequisite") assert task is created_task @@ -143,7 +147,10 @@ async def test_download_project_requires_remote(monkeypatch, project_manager, ta @pytest.mark.asyncio -async def test_download_project_rejects_running_task(monkeypatch, project_manager, task_manager): +@pytest.mark.parametrize("status", [TaskStatus.RUNNING, TaskStatus.INTERRUPTED]) +async def test_download_project_rejects_active_or_interrupted_task( + monkeypatch, project_manager, task_manager, status +): project = project_manager.get_project_by_name("demo") project.cloud_project_name = "demo-remote.zip" @@ -155,7 +162,7 @@ async def test_download_project_rejects_running_task(monkeypatch, project_manage task = Task( name="cloud_download_task", task_type="cloud_download_task", - status=TaskStatus.RUNNING, + status=status, run_args=("demo",), ) task_manager.add_task(task) diff --git a/tests/controllers/services/test_focus_stacking_service.py b/tests/controllers/services/test_focus_stacking_service.py index 450eaa0..a15d43a 100644 --- a/tests/controllers/services/test_focus_stacking_service.py +++ b/tests/controllers/services/test_focus_stacking_service.py @@ -49,6 +49,7 @@ def patch_project_manager(monkeypatch, scan: Scan): def patch_task_manager(monkeypatch): task_manager = MagicMock() task_manager.create_and_run_task = AsyncMock() + task_manager.replace_task = AsyncMock() task_manager.pause_task = AsyncMock() task_manager.resume_task = AsyncMock() task_manager.cancel_task = AsyncMock() @@ -65,9 +66,19 @@ async def test_start_focus_stacking_persists_task_reference(scan: Scan, patch_pr id="task-123", ) - task = await service.start_focus_stacking("demo", 1) + task = await service.start_focus_stacking( + "demo", + 1, + depends_on="task-prerequisite", + ) patch_project_manager.get_scan_by_index.assert_called_once_with("demo", 1) + patch_task_manager.create_and_run_task.assert_awaited_once_with( + "focus_stacking_task", + "demo", + 1, + depends_on="task-prerequisite", + ) assert scan.stacking_task_status == StackingTaskStatus(task_id="task-123", status=TaskStatus.RUNNING) patch_project_manager.save_scan_state.assert_awaited_once_with(scan) assert task.id == "task-123" @@ -86,6 +97,72 @@ async def test_start_focus_stacking_returns_existing_active_task(scan: Scan, pat assert task.id == "task-999" +@pytest.mark.asyncio +async def test_start_focus_stacking_removes_replaced_task( + scan: Scan, + patch_project_manager, + patch_task_manager, +): + """A new focus run removes the old terminal task record first.""" + scan.stacking_task_status = StackingTaskStatus( + task_id="task-interrupted", + status=TaskStatus.INTERRUPTED, + ) + patch_task_manager.get_task_info.return_value = Task( + name="focus_stacking_task", + task_type="core", + status=TaskStatus.INTERRUPTED, + id="task-interrupted", + ) + patch_task_manager.create_and_run_task.return_value = Task( + name="focus_stacking_task", + task_type="core", + status=TaskStatus.RUNNING, + id="task-new", + ) + + task = await service.start_focus_stacking("demo", 1) + + patch_task_manager.replace_task.assert_awaited_once_with( + "task-interrupted", + "task-new", + ) + assert task.id == "task-new" + + +@pytest.mark.asyncio +async def test_start_focus_stacking_replacement_does_not_inherit_dependency( + scan: Scan, + patch_project_manager, + patch_task_manager, +): + scan.stacking_task_status = StackingTaskStatus( + task_id="task-interrupted", + status=TaskStatus.INTERRUPTED, + ) + patch_task_manager.get_task_info.return_value = Task( + name="focus_stacking_task", + task_type="core", + status=TaskStatus.INTERRUPTED, + id="task-interrupted", + depends_on="task-prerequisite", + ) + patch_task_manager.create_and_run_task.return_value = Task( + name="focus_stacking_task", + task_type="core", + status=TaskStatus.PENDING, + id="task-new", + ) + + await service.start_focus_stacking("demo", 1) + + patch_task_manager.create_and_run_task.assert_awaited_once_with("focus_stacking_task", "demo", 1) + patch_task_manager.replace_task.assert_awaited_once_with( + "task-interrupted", + "task-new", + ) + + @pytest.mark.asyncio @pytest.mark.parametrize( "service_fn, manager_attr, expected_status", @@ -122,3 +199,35 @@ async def test_resume_focus_stacking_without_task_returns_none(scan: Scan, patch assert result is None patch_task_manager.resume_task.assert_not_called() patch_project_manager.save_scan_state.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_resume_interrupted_focus_stacking_without_task_id_starts_new_task( + scan: Scan, + patch_project_manager, + patch_task_manager, +): + """An interrupted focus task detached during startup is recreated by the service.""" + scan.stacking_task_status = StackingTaskStatus(status=TaskStatus.INTERRUPTED) + new_task = Task( + name="focus_stacking_task", + task_type="core", + status=TaskStatus.RUNNING, + id="task-recreated", + ) + patch_task_manager.create_and_run_task.return_value = new_task + + result = await service.resume_focus_stacking("demo", 1) + + assert result is new_task + patch_task_manager.resume_task.assert_not_called() + patch_task_manager.create_and_run_task.assert_awaited_once_with( + "focus_stacking_task", + "demo", + 1, + ) + assert scan.stacking_task_status == StackingTaskStatus( + task_id="task-recreated", + status=TaskStatus.RUNNING, + ) + patch_project_manager.save_scan_state.assert_awaited_once_with(scan) diff --git a/tests/controllers/services/test_project_manager.py b/tests/controllers/services/test_project_manager.py index e150bfd..aa0c8b6 100644 --- a/tests/controllers/services/test_project_manager.py +++ b/tests/controllers/services/test_project_manager.py @@ -163,7 +163,7 @@ async def test_pm_init_loads_existing_project(project_manager: ProjectManager, assert actual_scan.current_step == 10 -@pytest.mark.parametrize("initial_status", [TaskStatus.RUNNING, TaskStatus.PENDING]) +@pytest.mark.parametrize("initial_status", [TaskStatus.RUNNING, TaskStatus.PENDING, TaskStatus.PAUSED]) def test_pm_recovers_incomplete_scans( tmp_path: Path, sample_scan_settings: ScanSetting, initial_status: TaskStatus ): @@ -315,7 +315,7 @@ async def test_pm_save_scan_state_persists_stacking_status( assert reloaded_scan is not None assert reloaded_scan.stacking_task_status is not None - assert reloaded_scan.stacking_task_status.task_id is None + assert reloaded_scan.stacking_task_status.task_id == "stack-123" assert reloaded_scan.stacking_task_status.status == TaskStatus.INTERRUPTED diff --git a/tests/controllers/services/test_scans_service.py b/tests/controllers/services/test_scans_service.py index 5c6550a..e54f2e9 100644 --- a/tests/controllers/services/test_scans_service.py +++ b/tests/controllers/services/test_scans_service.py @@ -27,17 +27,61 @@ async def test_start_scan_restarts_interrupted_task(sample_scan_model: Scan) -> task_manager_mock = MagicMock() task_manager_mock.get_task_info.return_value = existing_task task_manager_mock.create_and_run_task = AsyncMock(return_value=new_task) - task_manager_mock.delete_task = AsyncMock() + task_manager_mock.replace_task = AsyncMock() with patch("openscan_firmware.controllers.services.scans.get_task_manager", return_value=task_manager_mock): - result = await scans.start_scan(project_manager, scan, camera_controller, start_from_step=3) + result = await scans.start_scan( + project_manager, + scan, + camera_controller, + start_from_step=3, + depends_on="task-prerequisite", + ) assert result is new_task - task_manager_mock.create_and_run_task.assert_awaited_once_with("scan_task", scan, 3) + task_manager_mock.replace_task.assert_awaited_once_with( + "task-interrupted", + "task-new", + ) + task_manager_mock.create_and_run_task.assert_awaited_once_with( + "scan_task", + scan, + 3, + depends_on="task-prerequisite", + ) assert scan.task_id == new_task.id project_manager.save_scan_state.assert_awaited_once_with(scan) +@pytest.mark.asyncio +async def test_start_scan_replacement_does_not_inherit_dependency(sample_scan_model: Scan) -> None: + scan = sample_scan_model + scan.task_id = "task-interrupted" + scan.camera_name = "mock-cam" + + camera_controller = MagicMock() + camera_controller.camera.name = "mock-cam" + project_manager = MagicMock() + project_manager.save_scan_state = AsyncMock() + + existing_task = Task( + name="scan_task", + task_type="core", + status=TaskStatus.INTERRUPTED, + id="task-interrupted", + depends_on="task-prerequisite", + ) + new_task = Task(name="scan_task", task_type="core", status=TaskStatus.PENDING, id="task-new") + task_manager_mock = MagicMock() + task_manager_mock.get_task_info.return_value = existing_task + task_manager_mock.create_and_run_task = AsyncMock(return_value=new_task) + task_manager_mock.replace_task = AsyncMock() + + with patch("openscan_firmware.controllers.services.scans.get_task_manager", return_value=task_manager_mock): + await scans.start_scan(project_manager, scan, camera_controller) + + task_manager_mock.create_and_run_task.assert_awaited_once_with("scan_task", scan, 0) + @pytest.mark.asyncio async def test_pause_scan_updates_status_and_persists(sample_scan_model: Scan) -> None: scan = sample_scan_model diff --git a/tests/controllers/services/test_task_manager.py b/tests/controllers/services/test_task_manager.py index ccdfa5d..b3c673c 100644 --- a/tests/controllers/services/test_task_manager.py +++ b/tests/controllers/services/test_task_manager.py @@ -314,6 +314,269 @@ def print_state(tag: str): assert tm.get_task_info(another_non_exclusive_task.id).status == TaskStatus.PENDING +async def test_dependent_task_waits_until_dependency_completes(task_manager_fixture: TaskManager): + """A dependency-blocked task is not scheduled before its dependency succeeds.""" + tm = task_manager_fixture + dependency_release = asyncio.Event() + + dependency = await tm.create_and_run_task( + "controlled_async_task", + completion_event=dependency_release, + ) + dependent = await tm.create_and_run_task( + "hello_world_progress_task", + total_steps=1, + depends_on=dependency.id, + ) + + await asyncio.sleep(0.05) + assert tm.get_task_info(dependency.id).status == TaskStatus.RUNNING + assert tm.get_task_info(dependent.id).status == TaskStatus.PENDING + assert not tm._is_task_queued(dependent.id) + + dependency_release.set() + + dependency_state = await wait_for_task_completion(tm, dependency.id) + dependent_state = await wait_for_task_completion(tm, dependent.id) + assert dependency_state.status == TaskStatus.COMPLETED + assert dependent_state.status == TaskStatus.COMPLETED + assert dependent_state.depends_on is None + + +async def test_dependent_task_fails_when_dependency_is_cancelled(task_manager_fixture: TaskManager): + """A dependency-blocked task must not run after its dependency is cancelled.""" + tm = task_manager_fixture + dependency_release = asyncio.Event() + + dependency = await tm.create_and_run_task( + "controlled_async_task", + completion_event=dependency_release, + ) + dependent = await tm.create_and_run_task( + "hello_world_progress_task", + total_steps=1, + depends_on=dependency.id, + ) + + await tm.cancel_task(dependency.id) + dependency_state = await wait_for_task_completion(tm, dependency.id) + dependent_state = await wait_for_task_completion(tm, dependent.id) + + assert dependency_state.status == TaskStatus.CANCELLED + assert dependent_state.status == TaskStatus.ERROR + assert "dependency" in dependent_state.error.lower() + + +async def test_dependent_task_is_started_when_dependency_already_completed(task_manager_fixture: TaskManager): + """A task created after a successful dependency should start immediately.""" + tm = task_manager_fixture + + dependency = await tm.create_and_run_task( + "hello_world_progress_task", + total_steps=1, + interval=0.01, + ) + dependency_state = await wait_for_task_completion(tm, dependency.id) + assert dependency_state.status == TaskStatus.COMPLETED + + dependent = await tm.create_and_run_task( + "hello_world_progress_task", + total_steps=1, + interval=0.01, + depends_on=dependency.id, + ) + assert dependent.status in {TaskStatus.RUNNING, TaskStatus.COMPLETED} + + dependent_state = await wait_for_task_completion(tm, dependent.id) + assert dependent_state.status == TaskStatus.COMPLETED + assert dependent_state.depends_on is None + + with open(TASKS_STORAGE_PATH / f"{dependent.id}.json", "r") as task_file: + persisted_dependent = json.load(task_file) + assert persisted_dependent["depends_on"] is None + + +async def test_dependent_task_fails_when_dependency_already_failed(task_manager_fixture: TaskManager): + """A task created after a failed dependency must fail without starting.""" + tm = task_manager_fixture + + dependency = await tm.create_and_run_task( + "failing_task", + error_message="dependency failed before child creation", + ) + dependency_state = await wait_for_task_completion(tm, dependency.id) + assert dependency_state.status == TaskStatus.ERROR + + dependent = await tm.create_and_run_task( + "hello_world_progress_task", + total_steps=1, + depends_on=dependency.id, + ) + + assert dependent.status == TaskStatus.ERROR + assert dependent.started_at is None + assert "Dependency task" in dependent.error + assert dependent.depends_on == dependency.id + + +async def test_cancelling_queued_dependency_fails_waiting_dependents(task_manager_fixture: TaskManager): + """Cancelling a queued dependency must fail tasks waiting on it.""" + tm = task_manager_fixture + + exclusive_task = await tm.create_and_run_task("exclusive_demo_task", duration=2) + dependency = await tm.create_and_run_task( + "hello_world_progress_task", + total_steps=1, + interval=0.1, + ) + dependent = await tm.create_and_run_task( + "hello_world_progress_task", + total_steps=1, + depends_on=dependency.id, + ) + + await asyncio.sleep(0.05) + assert tm.get_task_info(dependency.id).status == TaskStatus.PENDING + assert tm._is_task_queued(dependency.id) + assert tm.get_task_info(dependent.id).status == TaskStatus.PENDING + + await tm.cancel_task(dependency.id) + + dependency_state = await wait_for_task_completion(tm, dependency.id) + dependent_state = await wait_for_task_completion(tm, dependent.id) + assert dependency_state.status == TaskStatus.CANCELLED + assert dependent_state.status == TaskStatus.ERROR + assert dependent_state.started_at is None + + await tm.cancel_task(exclusive_task.id) + + +async def test_cancelling_waiting_dependent_prevents_later_start(task_manager_fixture: TaskManager): + """A cancelled dependency-blocked task must stay cancelled after its parent succeeds.""" + tm = task_manager_fixture + dependency_release = asyncio.Event() + + dependency = await tm.create_and_run_task( + "controlled_async_task", + completion_event=dependency_release, + ) + dependent = await tm.create_and_run_task( + "hello_world_progress_task", + total_steps=1, + depends_on=dependency.id, + ) + + cancelled_dependent = await tm.cancel_task(dependent.id) + assert cancelled_dependent.status == TaskStatus.CANCELLED + + dependency_release.set() + await wait_for_task_completion(tm, dependency.id) + await asyncio.sleep(0.05) + + dependent_state = tm.get_task_info(dependent.id) + assert dependent_state.status == TaskStatus.CANCELLED + assert dependent_state.started_at is None + + +async def test_create_task_rejects_missing_dependency(task_manager_fixture: TaskManager): + """A dependency ID must refer to a known task.""" + tm = task_manager_fixture + + with pytest.raises(ValueError, match="does not exist"): + await tm.create_and_run_task( + "hello_world_progress_task", + total_steps=1, + depends_on="missing-task-id", + ) + + +async def test_dependency_cycle_is_rejected(task_manager_fixture: TaskManager): + """The dependency validator must reject a cycle in the existing task graph.""" + tm = task_manager_fixture + first = Task(name="hello_world_progress_task", task_type="hello_world_progress_task") + second = Task( + name="hello_world_progress_task", + task_type="hello_world_progress_task", + depends_on=first.id, + ) + first.depends_on = second.id + tm._tasks[first.id] = first + tm._tasks[second.id] = second + + cycle_candidate = Task( + name="hello_world_progress_task", + task_type="hello_world_progress_task", + depends_on=first.id, + ) + + with pytest.raises(ValueError, match="cycle detected"): + tm._validate_dependency(cycle_candidate) + + +async def test_replacing_task_repoints_dependents(task_manager_fixture: TaskManager): + """Replacing a task keeps dependent tasks attached to the new task ID.""" + tm = task_manager_fixture + interrupted = Task( + name="hello_world_progress_task", + task_type="hello_world_progress_task", + status=TaskStatus.INTERRUPTED, + ) + replacement = Task( + name="hello_world_progress_task", + task_type="hello_world_progress_task", + status=TaskStatus.PENDING, + ) + dependent = Task( + name="hello_world_progress_task", + task_type="hello_world_progress_task", + status=TaskStatus.PENDING, + depends_on=interrupted.id, + ) + dependent_dependent = Task( + name="hello_world_progress_task", + task_type="hello_world_progress_task", + status=TaskStatus.PENDING, + depends_on=dependent.id, + ) + tm._tasks.update( + { + interrupted.id: interrupted, + replacement.id: replacement, + dependent.id: dependent, + dependent_dependent.id: dependent_dependent, + } + ) + tm._save_task_state(interrupted) + tm._save_task_state(replacement) + tm._save_task_state(dependent) + tm._save_task_state(dependent_dependent) + + await tm.replace_task(interrupted.id, replacement.id) + + assert tm.get_task_info(interrupted.id) is None + assert dependent.depends_on == replacement.id + assert dependent_dependent.depends_on == dependent.id + with open(TASKS_STORAGE_PATH / f"{dependent.id}.json") as task_file: + persisted_dependent = json.load(task_file) + assert persisted_dependent["depends_on"] == replacement.id + assert not os.path.exists(TASKS_STORAGE_PATH / f"{interrupted.id}.json") + + +async def test_wait_for_interrupted_task_returns_immediately(task_manager_fixture: TaskManager): + """Interrupted is a terminal persisted state until an explicit resume.""" + tm = task_manager_fixture + task = Task( + name="hello_world_progress_task", + task_type="hello_world_progress_task", + status=TaskStatus.INTERRUPTED, + ) + tm._tasks[task.id] = task + + waited_for = await tm.wait_for_task(task.id, timeout=0.01) + + assert waited_for is task + + async def test_pause_and_resume_task(task_manager_fixture: TaskManager): """ Tests pausing and resuming a running task. @@ -348,6 +611,29 @@ async def test_pause_and_resume_task(task_manager_fixture: TaskManager): assert final_task_state.status == TaskStatus.COMPLETED +async def test_resume_interrupted_task_uses_persisted_progress(task_manager_fixture: TaskManager): + """An interrupted task can be resumed after its in-memory execution is gone.""" + tm = task_manager_fixture + task = Task( + name="hello_world_progress_task", + task_type="hello_world_progress_task", + status=TaskStatus.INTERRUPTED, + progress=TaskProgress(current=2, total=4, message="Interrupted"), + run_kwargs={"total_steps": 4, "interval": 0.01}, + ) + tm._save_task_state(task) + tm.restore_tasks_from_persistence() + + resumed_task = await tm.resume_task(task.id) + + assert resumed_task.id == task.id + assert resumed_task.status in (TaskStatus.RUNNING, TaskStatus.PENDING) + assert resumed_task.progress.current == 2 + + final_task_state = await tm.wait_for_task(task.id, timeout=2) + assert final_task_state.status == TaskStatus.COMPLETED + + async def test_streaming_task_progress(task_manager_fixture: TaskManager): """ Tests that a task using an async generator correctly streams progress updates. @@ -584,6 +870,17 @@ async def test_tasks_are_reloaded_on_startup(task_manager_fixture: TaskManager): completed_task = Task(name="completed_task", task_type="hello_world_progress_task", status=TaskStatus.COMPLETED) running_task = Task(name="running_task", task_type="hello_world_progress_task", status=TaskStatus.RUNNING) paused_task = Task(name="paused_task", task_type="hello_world_progress_task", status=TaskStatus.PAUSED) + dependency_task = Task( + name="dependency_task", + task_type="hello_world_progress_task", + status=TaskStatus.RUNNING, + ) + pending_task = Task( + name="pending_task", + task_type="hello_world_progress_task", + status=TaskStatus.PENDING, + depends_on=dependency_task.id, + ) with open(TASKS_STORAGE_PATH / f"{completed_task.id}.json", 'w') as f: f.write(completed_task.model_dump_json()) @@ -591,6 +888,10 @@ async def test_tasks_are_reloaded_on_startup(task_manager_fixture: TaskManager): f.write(running_task.model_dump_json()) with open(TASKS_STORAGE_PATH / f"{paused_task.id}.json", 'w') as f: f.write(paused_task.model_dump_json()) + with open(TASKS_STORAGE_PATH / f"{dependency_task.id}.json", 'w') as f: + f.write(dependency_task.model_dump_json()) + with open(TASKS_STORAGE_PATH / f"{pending_task.id}.json", 'w') as f: + f.write(pending_task.model_dump_json()) # --- Simulate Application Restart --- # Instead of creating a new instance, we clear the internal state of the @@ -601,11 +902,13 @@ async def test_tasks_are_reloaded_on_startup(task_manager_fixture: TaskManager): # --- Verification --- # Check if all non-completed tasks were loaded all_loaded_tasks = tm.get_all_tasks_info() - assert len(all_loaded_tasks) == 2 # Completed task should be cleaned up + assert len(all_loaded_tasks) == 4 # Completed task should be cleaned up loaded_task_ids = {t.id for t in all_loaded_tasks} assert running_task.id in loaded_task_ids assert paused_task.id in loaded_task_ids + assert dependency_task.id in loaded_task_ids + assert pending_task.id in loaded_task_ids assert completed_task.id not in loaded_task_ids # Verify the completed task's file was deleted @@ -615,6 +918,10 @@ async def test_tasks_are_reloaded_on_startup(task_manager_fixture: TaskManager): for task in all_loaded_tasks: assert task.status == TaskStatus.INTERRUPTED + restored_pending_task = tm.get_task_info(pending_task.id) + assert restored_pending_task.depends_on == dependency_task.id + assert tm._pending_tasks.empty() + async def test_cancelled_task_state_is_persisted(task_manager_fixture: TaskManager): """Tests that a cancelled task's final state is saved to its JSON file.""" @@ -776,6 +1083,80 @@ async def test_restart_interrupted_task_after_shutdown(task_manager_fixture: Tas assert final_state.progress.current == total_steps +async def test_restart_restored_pending_task_after_shutdown(task_manager_fixture: TaskManager): + """A task that was pending at shutdown can be explicitly restarted afterwards.""" + tm = task_manager_fixture + task = Task( + name="hello_world_progress_task", + task_type="hello_world_progress_task", + status=TaskStatus.PENDING, + run_kwargs={"total_steps": 1, "interval": 0.01}, + ) + tm._save_task_state(task) + + tm.restore_tasks_from_persistence() + + restored_task = tm.get_task_info(task.id) + assert restored_task.status == TaskStatus.INTERRUPTED + assert tm._pending_tasks.empty() + + restarted_task = await tm.restart_task(task.id) + assert restarted_task.status == TaskStatus.RUNNING + + final_state = await tm.wait_for_task(task.id, timeout=2) + assert final_state.status == TaskStatus.COMPLETED + + +async def test_restart_waiting_task_after_restarted_dependency_completed( + task_manager_fixture: TaskManager, +): + """A restarted dependent task starts immediately after its restarted dependency completed.""" + tm = task_manager_fixture + dependency = Task( + name="hello_world_progress_task", + task_type="hello_world_progress_task", + status=TaskStatus.RUNNING, + run_kwargs={"total_steps": 1, "interval": 0.01}, + ) + dependent = Task( + name="hello_world_progress_task", + task_type="hello_world_progress_task", + status=TaskStatus.PENDING, + depends_on=dependency.id, + run_kwargs={"total_steps": 1, "interval": 0.01}, + ) + tm._save_task_state(dependency) + tm._save_task_state(dependent) + + tm.restore_tasks_from_persistence() + + restored_dependency = tm.get_task_info(dependency.id) + restored_dependent = tm.get_task_info(dependent.id) + assert restored_dependency.status == TaskStatus.INTERRUPTED + assert restored_dependent.status == TaskStatus.INTERRUPTED + assert restored_dependent.depends_on == dependency.id + + restarted_dependency = await tm.restart_task(dependency.id) + assert restarted_dependency.status == TaskStatus.RUNNING + dependency_state = await wait_for_task_completion(tm, dependency.id, timeout=2) + assert dependency_state.status == TaskStatus.COMPLETED + + # Recovery is intentionally manual; the dependent is not auto-started by + # the predecessor's completion event because it was also interrupted. + restored_dependent = tm.get_task_info(dependent.id) + assert restored_dependent.status == TaskStatus.INTERRUPTED + assert restored_dependent.depends_on is None + + with open(TASKS_STORAGE_PATH / f"{dependent.id}.json", "r") as task_file: + persisted_dependent = json.load(task_file) + assert persisted_dependent["depends_on"] is None + + restarted_dependent = await tm.restart_task(dependent.id) + assert restarted_dependent.status == TaskStatus.RUNNING + dependent_state = await wait_for_task_completion(tm, dependent.id, timeout=2) + assert dependent_state.status == TaskStatus.COMPLETED + + async def test_cancel_pending_task_in_full_queue(task_manager_fixture: TaskManager): """ Tests that a task waiting due to concurrency limits can be cancelled. diff --git a/tests/routers/test_focus_stacking_router.py b/tests/routers/test_focus_stacking_router.py index 8903947..be7e0d7 100644 --- a/tests/routers/test_focus_stacking_router.py +++ b/tests/routers/test_focus_stacking_router.py @@ -53,7 +53,7 @@ async def _stub(*args, **kwargs): @pytest.mark.parametrize("endpoint", [ ("pause", "Focus stacking is not running"), ("cancel", "Focus stacking is not running"), - ("resume", "Focus stacking is not paused"), + ("resume", "Focus stacking is not paused or interrupted"), ]) def test_focus_stacking_conflict( monkeypatch, diff --git a/tests/routers/test_next_projects_request_bodies.py b/tests/routers/test_next_projects_request_bodies.py index 8fad49d..146d142 100644 --- a/tests/routers/test_next_projects_request_bodies.py +++ b/tests/routers/test_next_projects_request_bodies.py @@ -48,6 +48,7 @@ def test_add_scan_accepts_all_input_in_json_body(monkeypatch) -> None: "camera_name": "cam0", "scan_settings": ScanSetting().model_dump(mode="json"), "scan_description": "Created from a JSON request body.", + "depends_on": "previous-task", }, ) @@ -58,6 +59,12 @@ def test_add_scan_accepts_all_input_in_json_body(monkeypatch) -> None: ScanSetting(), "Created from a JSON request body.", ) + projects_router.scans.start_scan.assert_awaited_once_with( + project_manager, + scan, + camera_controller, + depends_on="previous-task", + ) def test_next_projects_openapi_uses_json_request_bodies() -> None: @@ -87,6 +94,7 @@ def test_versioned_projects_openapi_contracts_remain_unchanged() -> None: "project_name", "camera_name", "scan_description", + "depends_on", ] assert scan_post["requestBody"]["content"]["application/json"]["schema"] == { "$ref": "#/components/schemas/ScanSetting" diff --git a/tests/routers/test_tasks_router.py b/tests/routers/test_tasks_router.py new file mode 100644 index 0000000..73c8997 --- /dev/null +++ b/tests/routers/test_tasks_router.py @@ -0,0 +1,281 @@ +import asyncio +from importlib import import_module +from unittest.mock import AsyncMock, MagicMock + +import httpx +import pytest +import pytest_asyncio +from fastapi.testclient import TestClient + +import openscan_firmware.controllers.services.tasks.task_manager as task_manager_module +from openscan_firmware.controllers.services.tasks.task_manager import TaskManager +from openscan_firmware.main import app +from openscan_firmware.models.task import Task, TaskStatus + + +@pytest.fixture +def client() -> TestClient: + with TestClient(app) as test_client: + yield test_client + + +@pytest_asyncio.fixture +async def real_task_manager(monkeypatch: pytest.MonkeyPatch) -> TaskManager: + task_manager = TaskManager() + # get_task_manager() returns the module-level singleton, so point the API + # at the isolated real manager used by this integration test. + monkeypatch.setattr(task_manager_module, "task_manager", task_manager) + from openscan_firmware.controllers.services.tasks.examples import demo_examples + + task_manager.register_task("hello_world_progress_task", demo_examples.HelloWorldProgressTask) + task_manager.register_task("failing_task", demo_examples.FailingTask) + yield task_manager + + +@pytest_asyncio.fixture +async def async_client(real_task_manager: TaskManager) -> httpx.AsyncClient: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + yield client + + +async def _create_task( + client: httpx.AsyncClient, + api_version: str, + task_name: str, + *, + kwargs: dict | None = None, + depends_on: str | None = None, +) -> Task: + payload = {"args": [], "kwargs": kwargs or {}} + if depends_on is not None: + payload["depends_on"] = depends_on + + response = await client.post(f"/{api_version}/tasks/{task_name}", json=payload) + assert response.status_code == 202, response.text + return Task.model_validate(response.json()) + + +async def _wait_for_task(task_manager: TaskManager, task_id: str) -> Task: + return await task_manager.wait_for_task(task_id, timeout=5) + + +@pytest.mark.parametrize( + ("api_version", "router_module"), + [ + ("v0.8", "openscan_firmware.routers.v0_8.tasks"), + ("v0.9", "openscan_firmware.routers.v0_9.tasks"), + ("next", "openscan_firmware.routers.next.tasks"), + ], +) +def test_create_task_accepts_dependency( + client: TestClient, + monkeypatch: pytest.MonkeyPatch, + api_version: str, + router_module: str, +) -> None: + task_manager = MagicMock() + task_manager.create_and_run_task = AsyncMock( + return_value=Task(name="hello_world_progress_task", task_type="hello_world_progress_task") + ) + monkeypatch.setattr(import_module(router_module), "get_task_manager", lambda: task_manager) + + response = client.post( + f"/{api_version}/tasks/hello_world_progress_task", + json={ + "args": ["demo"], + "kwargs": {"some_option": True}, + "depends_on": "scan-task-id", + }, + ) + + assert response.status_code == 202 + task_manager.create_and_run_task.assert_awaited_once_with( + "hello_world_progress_task", + "demo", + depends_on="scan-task-id", + some_option=True, + ) + + +@pytest.mark.parametrize( + ("api_version", "router_module"), + [ + ("v0.8", "openscan_firmware.routers.v0_8.tasks"), + ("v0.9", "openscan_firmware.routers.v0_9.tasks"), + ("next", "openscan_firmware.routers.next.tasks"), + ], +) +def test_create_task_dependency_is_optional( + client: TestClient, + monkeypatch: pytest.MonkeyPatch, + api_version: str, + router_module: str, +) -> None: + task_manager = MagicMock() + task_manager.create_and_run_task = AsyncMock( + return_value=Task(name="hello_world_progress_task", task_type="hello_world_progress_task") + ) + monkeypatch.setattr(import_module(router_module), "get_task_manager", lambda: task_manager) + + response = client.post( + f"/{api_version}/tasks/hello_world_progress_task", + json={"args": [], "kwargs": {}}, + ) + + assert response.status_code == 202 + task_manager.create_and_run_task.assert_awaited_once_with( + "hello_world_progress_task", + depends_on=None, + ) + + +@pytest.mark.parametrize( + ("task_name", "expected_endpoint"), + [ + ("scan_task", "POST /projects/{project_name}/scan"), + ( + "focus_stacking_task", + "POST /projects/{project_name}/scans/{scan_index}/focus-stacking/start", + ), + ("cloud_upload_task", "POST /projects/{project_name}/upload"), + ], +) +def test_domain_tasks_must_use_project_specific_endpoints( + client: TestClient, + task_name: str, + expected_endpoint: str, +) -> None: + response = client.post( + f"/next/tasks/{task_name}", + json={"args": [], "kwargs": {}}, + ) + + assert response.status_code == 400 + assert expected_endpoint in response.json()["detail"] + assert "experimental or custom tasks" in response.json()["detail"] + + +@pytest.mark.asyncio +async def test_task_api_runs_real_three_step_chain( + async_client: httpx.AsyncClient, + real_task_manager: TaskManager, +) -> None: + first = await _create_task( + async_client, + "next", + "hello_world_progress_task", + kwargs={"total_steps": 2, "interval": 0.15}, + ) + second = await _create_task( + async_client, + "next", + "hello_world_progress_task", + kwargs={"total_steps": 1, "interval": 0.01}, + depends_on=first.id, + ) + third = await _create_task( + async_client, + "next", + "hello_world_progress_task", + kwargs={"total_steps": 1, "interval": 0.01}, + depends_on=second.id, + ) + + assert real_task_manager.get_task_info(first.id).status == TaskStatus.RUNNING + assert real_task_manager.get_task_info(second.id).status == TaskStatus.PENDING + assert real_task_manager.get_task_info(third.id).status == TaskStatus.PENDING + + first_result, second_result, third_result = await asyncio.gather( + _wait_for_task(real_task_manager, first.id), + _wait_for_task(real_task_manager, second.id), + _wait_for_task(real_task_manager, third.id), + ) + + assert first_result.status == TaskStatus.COMPLETED + assert second_result.status == TaskStatus.COMPLETED + assert third_result.status == TaskStatus.COMPLETED + assert real_task_manager.get_task_info(second.id).depends_on is None + assert real_task_manager.get_task_info(third.id).depends_on is None + + +@pytest.mark.asyncio +async def test_task_api_propagates_failure_through_real_chain( + async_client: httpx.AsyncClient, + real_task_manager: TaskManager, +) -> None: + first = await _create_task( + async_client, + "next", + "failing_task", + kwargs={"error_message": "first task failed"}, + ) + second = await _create_task( + async_client, + "next", + "hello_world_progress_task", + kwargs={"total_steps": 1}, + depends_on=first.id, + ) + third = await _create_task( + async_client, + "next", + "hello_world_progress_task", + kwargs={"total_steps": 1}, + depends_on=second.id, + ) + + first_result, second_result, third_result = await asyncio.gather( + _wait_for_task(real_task_manager, first.id), + _wait_for_task(real_task_manager, second.id), + _wait_for_task(real_task_manager, third.id), + ) + + assert first_result.status == TaskStatus.ERROR + assert first_result.error == "first task failed" + assert second_result.status == TaskStatus.ERROR + assert third_result.status == TaskStatus.ERROR + assert "Dependency task" in second_result.error + assert "Dependency task" in third_result.error + assert real_task_manager.get_task_info(second.id).started_at is None + assert real_task_manager.get_task_info(third.id).started_at is None + + +@pytest.mark.asyncio +async def test_task_api_propagates_middle_task_failure_to_downstream_task( + async_client: httpx.AsyncClient, + real_task_manager: TaskManager, +) -> None: + first = await _create_task( + async_client, + "next", + "hello_world_progress_task", + kwargs={"total_steps": 1, "interval": 0.05}, + ) + second = await _create_task( + async_client, + "next", + "failing_task", + kwargs={"error_message": "middle task failed"}, + depends_on=first.id, + ) + third = await _create_task( + async_client, + "next", + "hello_world_progress_task", + kwargs={"total_steps": 1}, + depends_on=second.id, + ) + + first_result, second_result, third_result = await asyncio.gather( + _wait_for_task(real_task_manager, first.id), + _wait_for_task(real_task_manager, second.id), + _wait_for_task(real_task_manager, third.id), + ) + + assert first_result.status == TaskStatus.COMPLETED + assert second_result.status == TaskStatus.ERROR + assert second_result.error == "middle task failed" + assert third_result.status == TaskStatus.ERROR + assert "Dependency task" in third_result.error + assert real_task_manager.get_task_info(third.id).started_at is None