Add restart option for failed/dismissed jobs

CHANGES.md

@@ -1,5 +1,12 @@
 ## Changes in version 0.2.1 (in development)
 
+### Enhancements
+
+- **Wraptile** now supports restarting failed or dismissed jobs with
+  `POST /jobs/{jobId}/restart`. Local services reuse the original process
+  request, while Airflow services replay the original DAG-run configuration.
+  (#41)
+
 ### Fixes
 
 - Fixed `client.show_app()` forcing a redundant interactive OAuth2/PKCE login

gavicore/src/gavicore/service/service.py

@@ -1,6 +1,6 @@
 # generated by gen_server.py:
 #   filename:  service.py:
-#   timestamp: 2026-05-27T07:55:14.570257
+#   timestamp: 2026-08-03T11:32:13.433039
 
 from abc import ABC, abstractmethod
 
@@ -97,3 +97,9 @@ async def get_job_results(self, job_id: str, *args, **kwargs) -> JobResults:
 
         For more information, see [OGC API — Processes — Part 1 Section 7.13](https://docs.ogc.org/is/18-062r2/18-062r2.html#sc_retrieve_job_results).
         """
+
+    @abstractmethod
+    async def restart_job(self, job_id: str, *args, **kwargs) -> JobInfo:
+        """Create a new job using the original process request of a failed or
+        dismissed job.
+        """

gavicore/tests/test_service.py

@@ -18,6 +18,7 @@
     "get_processes",
     "get_job",
     "get_jobs",
+    "restart_job",
 }
 
 

tools/gen_server.py

@@ -55,7 +55,7 @@ def main():
         [
             "from abc import ABC, abstractmethod\n",
             "\n",
-            f"from .models import {model_list}\n",
+            f"from gavicore.models import {model_list}\n",
             "\n",
             "class Service(ABC):\n",
             service_code,

tools/openapi.yaml

@@ -382,6 +382,50 @@ paths:
             text/html:
               schema:
                 type: string
+  /jobs/{jobId}/restart:
+    post:
+      tags:
+      - Restart
+      summary: restart a failed or dismissed job
+      description: |
+        Create a new job using the original process request of a failed or
+        dismissed job.
+      operationId: restartJob
+      parameters:
+      - name: jobId
+        in: path
+        description: Local identifier of the job to restart
+        required: true
+        schema:
+          type: string
+      responses:
+        "201":
+          description: The restarted job.
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/JobInfo'
+        "403":
+          description: The job cannot be restarted.
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiError'
+        "404":
+          description: The requested job was not found.
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiError'
+        "500":
+          description: A server error occurred.
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiError'
+            text/html:
+              schema:
+                type: string
 components:
   schemas:
     #------------------------------------------------------

wraptile/src/wraptile/routes.py

@@ -1,6 +1,6 @@
 # generated by gen_server.py:
 #   filename:  routes.py:
-#   timestamp: 2026-05-27T07:55:14.099104
+#   timestamp: 2026-07-31T14:05:53.544409
 
 import fastapi
 
@@ -169,3 +169,20 @@ async def get_job_results(
     return await service.get_job_results(
         job_id=jobId, request=request, response=response
     )
+
+
+# noinspection PyPep8Naming
+@app.post(
+    "/jobs/{jobId}/restart",
+    status_code=201,
+    response_model=JobInfo,
+    response_model_exclude_none=True,
+    response_model_exclude_unset=True,
+)
+async def restart_job(
+    jobId: str,
+    request: fastapi.Request,
+    response: fastapi.Response,
+    service: Service = fastapi.Depends(get_service),  # noqa B008
+):
+    return await service.restart_job(job_id=jobId, request=request, response=response)

wraptile/src/wraptile/services/airflow/airflow_service.py

@@ -230,6 +230,25 @@ async def dismiss_job(self, job_id: str, *args, **kwargs) -> JobInfo:
             ) from e
         return self.dag_run_to_job_info(dag_run)
 
+    async def restart_job(self, job_id: str, *args, **kwargs) -> JobInfo:
+        """Create a new DAG run using the original run configuration."""
+        dag_id = self.get_dag_id_from_job_id(job_id)
+        try:
+            dag_run = self.airflow_dag_run_api.get_dag_run(dag_id, job_id)
+        except ApiException as e:
+            raise ServiceException(
+                e.status, e.reason, exception=e, is_job_problem=True
+            ) from e
+        job_info = self.dag_run_to_job_info(dag_run)
+        if job_info.status not in (JobStatus.failed, JobStatus.dismissed):
+            raise ServiceException(
+                403,
+                detail=f"Job {job_id!r} cannot be restarted unless it failed or was dismissed",
+                is_job_problem=True,
+            )
+        process_request = ProcessRequest(inputs=dag_run.conf or {})
+        return await self.execute_process(dag_id, process_request)
+
     async def get_job_results(self, job_id: str, *args, **kwargs) -> JobResults:
         dag_id = self.get_dag_id_from_job_id(job_id)
         return_value: Optional[Any] = None

wraptile/src/wraptile/services/local/local_service.py

@@ -40,6 +40,7 @@ def __init__(
 
         self.process_registry = process_registry or ProcessRegistry()
         self.jobs: dict[str, Job] = {}
+        self.job_requests: dict[str, ProcessRequest] = {}
         self.job_results: dict[str, JobResults | None] = {}
         self.job_uses_processes: dict[str, bool] = {}
         self._executor_uses_processes = False
@@ -117,6 +118,7 @@ async def execute_process(
                     ),
                 )
         self.jobs[job_id] = job
+        self.job_requests[job_id] = process_request.model_copy(deep=True)
         self.job_uses_processes[job_id] = use_processes
         if use_processes:
             assert self.service_ref is not None
@@ -160,10 +162,35 @@ async def dismiss_job(self, job_id: str, *args, **_kwargs) -> JobInfo:
             JobStatus.failed,
         ):
             del self.jobs[job_id]
+            self.job_requests.pop(job_id, None)
             self.job_results.pop(job_id, None)
             self.job_uses_processes.pop(job_id, None)
         return job.job_info
 
+    async def restart_job(self, job_id: str, *args, **_kwargs) -> JobInfo:
+        """Create a new job using a failed or dismissed job's request."""
+        job = self._get_job(
+            job_id,
+            forbidden_status_codes={
+                JobStatus.accepted: "cannot be restarted before it has finished",
+                JobStatus.running: "cannot be restarted while it is running",
+                JobStatus.successful: "cannot be restarted after succeeding",
+            },
+        )
+        assert job.job_info.status in (JobStatus.failed, JobStatus.dismissed)
+        process_id = job.job_info.processID
+        process_request = self.job_requests.get(job_id)
+        if process_id is None or process_request is None:
+            raise ServiceException(
+                500,
+                detail=f"Original request for job {job_id!r} is not available",
+                is_job_problem=True,
+            )
+        return await self.execute_process(
+            process_id,
+            process_request.model_copy(deep=True),
+        )
+
     async def get_job_results(self, job_id: str, *args, **_kwargs) -> JobResults:
         job = self._get_job(
             job_id,

wraptile/tests/services/base/test_service_base.py

@@ -51,6 +51,9 @@ async def get_job(self, job_id: str, *args, **kwargs) -> JobInfo:
     async def dismiss_job(self, job_id: str, *args, **kwargs) -> JobInfo:
         raise NotImplementedError
 
+    async def restart_job(self, job_id: str, *args, **kwargs) -> JobInfo:
+        raise NotImplementedError
+
     async def get_job_results(self, job_id: str, *args, **kwargs) -> JobResults:
         raise NotImplementedError
 

wraptile/tests/services/local/test_local_service.py

@@ -266,9 +266,59 @@ async def test_dismiss_finished_job_removes_cached_state(self):
         self.service.job_uses_processes[job_id] = False
         await self.service.dismiss_job(job_id=job_id, request=self.get_request())
         self.assertNotIn(job_id, self.service.jobs)
+        self.assertNotIn(job_id, self.service.job_requests)
         self.assertNotIn(job_id, self.service.job_results)
         self.assertNotIn(job_id, self.service.job_uses_processes)
 
+    async def test_restart_failed_job_reuses_original_request(self):
+        request = ProcessRequest(inputs={"max_val": 20})
+        failed_job = await self.service.execute_process(
+            process_id="primes_between",
+            process_request=request,
+            request=self.get_request(),
+        )
+        self.service.jobs[failed_job.jobID].job_info.status = JobStatus.failed
+
+        restarted_job = await self.service.restart_job(
+            job_id=failed_job.jobID,
+            request=self.get_request(),
+        )
+
+        self.assertNotEqual(failed_job.jobID, restarted_job.jobID)
+        self.assertEqual("primes_between", restarted_job.processID)
+        self.assertEqual(request, self.service.job_requests[restarted_job.jobID])
+
+    async def test_restart_non_failed_job_fails(self):
+        job_info = await self.service.execute_process(
+            process_id="primes_between",
+            process_request=ProcessRequest(inputs={"max_val": 20}),
+            request=self.get_request(),
+        )
+        self.service.jobs[job_info.jobID].job_info.status = JobStatus.successful
+
+        with pytest.raises(
+            ServiceException, match="cannot be restarted after succeeding"
+        ):
+            await self.service.restart_job(
+                job_id=job_info.jobID,
+                request=self.get_request(),
+            )
+
+    async def test_restart_without_original_request_fails(self):
+        job_info = await self.service.execute_process(
+            process_id="primes_between",
+            process_request=ProcessRequest(inputs={"max_val": 20}),
+            request=self.get_request(),
+        )
+        self.service.jobs[job_info.jobID].job_info.status = JobStatus.failed
+        del self.service.job_requests[job_info.jobID]
+
+        with pytest.raises(ServiceException, match="Original request.*not available"):
+            await self.service.restart_job(
+                job_id=job_info.jobID,
+                request=self.get_request(),
+            )
+
     def test_ensure_executor_reconfigures_missing_executor(self):
         self.service.executor = None
         executor = self.service._ensure_executor()

wraptile/tests/test_app.py

@@ -7,6 +7,7 @@
 
 from fastapi.testclient import TestClient
 
+from gavicore.models import JobStatus
 from wraptile.logging import LogMessageFilter
 from wraptile.main import app
 from wraptile.provider import ServiceProvider
@@ -71,6 +72,16 @@ def test_dismiss_job(self):
         response = client.delete(f"/jobs/{job_id}")
         self.assertEqual(200, response.status_code)
 
+    def test_restart_failed_job(self):
+        response = client.post("/processes/primes_between/execution", json={})
+        job_id = response.json()["jobID"]
+        service.jobs[job_id].job_info.status = JobStatus.failed
+
+        response = client.post(f"/jobs/{job_id}/restart")
+
+        self.assertEqual(201, response.status_code)
+        self.assertNotEqual(job_id, response.json()["jobID"])
+
     def test_get_job_results(self):
         response = client.post("/processes/primes_between/execution", json={})
         job_info = response.json()